MAPREDUCE-3251. Network ACLs can prevent some clients to talk to MR AM. Improved the earlier patch to not to JobHistoryServer repeatedly. Contributed by Anupam Seth.

svn merge --ignore-ancestry -c 1229787 ../../trunk/ git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.23@1229789 13f79535-47bb-0310-9956-ffa450edef68
2012-01-10 23:09:30 +00:00 · 2012-01-10 23:09:30 +00:00 · 92975b63e5
parent dd807a1e8a
commit 92975b63e5
7 changed files with 89 additions and 68 deletions
--- a/hadoop-mapreduce-project/CHANGES.txt
+++ b/hadoop-mapreduce-project/CHANGES.txt
@ -108,6 +108,10 @@ Release 0.23.1 - Unreleased
    MAPREDUCE-3299. Added AMInfo table to the MR AM job pages to list all the
    job-attempts when AM restarts and recovers. (Jonathan Eagles via vinodkv)

+    MAPREDUCE-3251. Network ACLs can prevent some clients to talk to MR AM.
+    Improved the earlier patch to not to JobHistoryServer repeatedly.
+    (Anupam Seth via vinodkv)
+
  OPTIMIZATIONS

    MAPREDUCE-3567. Extraneous JobConf objects in AM heap. (Vinod Kumar
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java
@ -370,6 +370,11 @@ public interface MRJobConfig {
    MR_AM_PREFIX  + "job.reduce.preemption.limit";
  public static final float DEFAULT_MR_AM_JOB_REDUCE_PREEMPTION_LIMIT = 0.5f;
  
+  /** AM ACL disabled. **/
+  public static final String JOB_AM_ACCESS_DISABLED = 
+    "mapreduce.job.am-access-disabled";
+  public static final boolean DEFAULT_JOB_AM_ACCESS_DISABLED = false;
+
  /**
   * Limit reduces starting until a certain percentage of maps have finished.
   *  Percentage between 0.0 and 1.0
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/ClientServiceDelegate.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/ClientServiceDelegate.java
@ -35,6 +35,7 @@ import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.JobID;
 import org.apache.hadoop.mapreduce.JobStatus;
+import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.TaskAttemptID;
 import org.apache.hadoop.mapreduce.TaskType;
 import org.apache.hadoop.mapreduce.TypeConverter;
@ -68,7 +69,6 @@ import org.apache.hadoop.yarn.YarnException;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
-import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
 import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
@ -94,6 +94,8 @@ public class ClientServiceDelegate {
  private static String UNKNOWN_USER = "Unknown User";
  private String trackingUrl;

+  private boolean amAclDisabledStatusLogged = false;
+
  public ClientServiceDelegate(Configuration conf, ResourceMgrDelegate rm,
      JobID jobId, MRClientProtocol historyServerProxy) {
    this.conf = new Configuration(conf); // Cloning for modifying.
@ -157,7 +159,7 @@ public class ClientServiceDelegate {
          application = rm.getApplicationReport(appId);
          continue;
        }
-        if(!conf.getBoolean(YarnConfiguration.RM_AM_NETWORK_ACL_CLOSED, false)) {
+        if(!conf.getBoolean(MRJobConfig.JOB_AM_ACCESS_DISABLED, false)) {
          UserGroupInformation newUgi = UserGroupInformation.createRemoteUser(
              UserGroupInformation.getCurrentUser().getUserName());
          serviceAddr = application.getHost() + ":" + application.getRpcPort();
@ -182,11 +184,13 @@ public class ClientServiceDelegate {
              return instantiateAMProxy(tempStr);
            }
          });
-	} else {
-           logApplicationReportInfo(application); 
-           LOG.info("Network ACL closed to AM for job " + jobId
-             + ". Redirecting to job history server.");
-           return checkAndGetHSProxy(null, JobState.RUNNING);
+        } else {
+          if (!amAclDisabledStatusLogged) {
+            LOG.info("Network ACL closed to AM for job " + jobId
+                + ". Not going to try to reach the AM.");
+            amAclDisabledStatusLogged = true;
+          }
+          return getNotRunningJob(null, JobState.RUNNING);
        }
        return realProxy;
      } catch (IOException e) {
@ -248,55 +252,10 @@ public class ClientServiceDelegate {
    return realProxy;
  }

-  private void logApplicationReportInfo(ApplicationReport application) {
-    if(application == null) {
-      return;
-    }
-    LOG.info("AppId: " + application.getApplicationId()
-      + " # reserved containers: " 
-      + application.getApplicationResourceUsageReport().getNumReservedContainers()
-      + " # used containers: " 
-      + application.getApplicationResourceUsageReport().getNumUsedContainers()
-      + " Needed resources (memory): "
-      + application.getApplicationResourceUsageReport().getNeededResources().getMemory()
-      + " Reserved resources (memory): "
-      + application.getApplicationResourceUsageReport().getReservedResources().getMemory()
-      + " Used resources (memory): "
-      + application.getApplicationResourceUsageReport().getUsedResources().getMemory()
-      + " Diagnostics: " 
-      + application.getDiagnostics()
-      + " Start time: "
-      + application.getStartTime()
-      + " Finish time: "
-      + application.getFinishTime()
-      + " Host: "
-      + application.getHost()
-      + " Name: "
-      + application.getName()
-      + " Orig. tracking url: "
-      + application.getOriginalTrackingUrl()
-      + " Queue: "
-      + application.getQueue()
-      + " RPC port: "
-      + application.getRpcPort()
-      + " Tracking url: "
-      + application.getTrackingUrl()
-      + " User: "
-      + application.getUser()
-      + " Client token: "
-      + application.getClientToken()
-      + " Final appl. status: "
-      + application.getFinalApplicationStatus()
-      + " Yarn appl. state: "
-      + application.getYarnApplicationState()
-    );
-  }
-
  private MRClientProtocol checkAndGetHSProxy(
      ApplicationReport applicationReport, JobState state) {
    if (null == historyServerProxy) {
-      LOG.warn("Job History Server is not configured or " +
-      		"job information not yet available on History Server.");
+      LOG.warn("Job History Server is not configured.");
      return getNotRunningJob(applicationReport, state);
    }
    return historyServerProxy;
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientServiceDelegate.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientServiceDelegate.java
@ -19,7 +19,12 @@
 package org.apache.hadoop.mapred;

 import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;

 import java.io.IOException;
 import java.util.Arrays;
@ -31,6 +36,7 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.JobID;
 import org.apache.hadoop.mapreduce.JobStatus;
 import org.apache.hadoop.mapreduce.MRConfig;
+import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.TypeConverter;
 import org.apache.hadoop.mapreduce.v2.api.MRClientProtocol;
 import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetCountersRequest;
@ -254,6 +260,58 @@ public class TestClientServiceDelegate {
        any(String.class));
  }
  
+  @Test
+  public void testAMAccessDisabled() throws IOException {
+    //test only applicable when AM not reachable
+    if(isAMReachableFromClient) {
+      return;
+    }
+
+    MRClientProtocol historyServerProxy = mock(MRClientProtocol.class);
+    when(historyServerProxy.getJobReport(getJobReportRequest())).thenReturn(                      
+        getJobReportResponseFromHistoryServer());                                                 
+
+    ResourceMgrDelegate rmDelegate = mock(ResourceMgrDelegate.class);
+    when(rmDelegate.getApplicationReport(jobId.getAppId())).thenReturn(
+        getRunningApplicationReport("am1", 78)).thenReturn(
+          getRunningApplicationReport("am1", 78)).thenReturn(
+            getRunningApplicationReport("am1", 78)).thenReturn(
+        getFinishedApplicationReport());
+
+    ClientServiceDelegate clientServiceDelegate = spy(getClientServiceDelegate(
+        historyServerProxy, rmDelegate));
+
+    JobStatus jobStatus = clientServiceDelegate.getJobStatus(oldJobId);
+    Assert.assertNotNull(jobStatus);
+    Assert.assertEquals("N/A", jobStatus.getJobName());
+    
+    verify(clientServiceDelegate, times(0)).instantiateAMProxy(
+        any(String.class));
+
+    // Should not reach AM even for second and third times too.
+    jobStatus = clientServiceDelegate.getJobStatus(oldJobId);
+    Assert.assertNotNull(jobStatus);
+    Assert.assertEquals("N/A", jobStatus.getJobName());    
+    verify(clientServiceDelegate, times(0)).instantiateAMProxy(
+        any(String.class));
+    jobStatus = clientServiceDelegate.getJobStatus(oldJobId);
+    Assert.assertNotNull(jobStatus);
+    Assert.assertEquals("N/A", jobStatus.getJobName());    
+    verify(clientServiceDelegate, times(0)).instantiateAMProxy(
+        any(String.class));
+
+    // The third time around, app is completed, so should go to JHS
+    JobStatus jobStatus1 = clientServiceDelegate.getJobStatus(oldJobId);
+    Assert.assertNotNull(jobStatus1);
+    Assert.assertEquals("TestJobFilePath", jobStatus1.getJobFile());                               
+    Assert.assertEquals("http://TestTrackingUrl", jobStatus1.getTrackingUrl());                    
+    Assert.assertEquals(1.0f, jobStatus1.getMapProgress());                                        
+    Assert.assertEquals(1.0f, jobStatus1.getReduceProgress());
+    
+    verify(clientServiceDelegate, times(0)).instantiateAMProxy(
+        any(String.class));
+  }
+  
  private GetJobReportRequest getJobReportRequest() {
    GetJobReportRequest request = Records.newRecord(GetJobReportRequest.class);
    request.setJobId(jobId);
@ -300,7 +358,7 @@ public class TestClientServiceDelegate {
      MRClientProtocol historyServerProxy, ResourceMgrDelegate rm) {
    Configuration conf = new YarnConfiguration();
    conf.set(MRConfig.FRAMEWORK_NAME, MRConfig.YARN_FRAMEWORK_NAME);
-    conf.setBoolean(YarnConfiguration.RM_AM_NETWORK_ACL_CLOSED, !isAMReachableFromClient);
+    conf.setBoolean(MRJobConfig.JOB_AM_ACCESS_DISABLED, !isAMReachableFromClient);
    ClientServiceDelegate clientServiceDelegate = new ClientServiceDelegate(
        conf, rm, oldJobId, historyServerProxy);
    return clientServiceDelegate;
--- a/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
+++ b/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
@ -145,11 +145,6 @@ public class YarnConfiguration extends Configuration {
  /** ACL used in case none is found. Allows nothing. */
  public static final String DEFAULT_YARN_APP_ACL = " ";

-  /** RM-AM ACL disabled. **/
-  public static final String RM_AM_NETWORK_ACL_CLOSED = 
-    RM_PREFIX + "am.acl.disabled";
-  public static final boolean DEFAULT_RM_AM_NETWORK_ACL_CLOSED = false;
-
  /** The address of the RM admin interface.*/
  public static final String RM_ADMIN_ADDRESS = 
    RM_PREFIX + "admin.address";
--- a/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/resources/yarn-default.xml
+++ b/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/resources/yarn-default.xml
@ -116,12 +116,6 @@
    <value>*</value>
  </property>

-  <property>
-    <description>Network ACL to AM closed.</description>
-    <name>yarn.resourcemanager.am.acl.disabled</name>
-    <value>false</value>
-  </property> 
-
  <property>
    <description>The address of the RM admin interface.</description>
    <name>yarn.resourcemanager.admin.address</name>
--- a/hadoop-mapreduce-project/src/java/mapred-default.xml
+++ b/hadoop-mapreduce-project/src/java/mapred-default.xml
@ -959,6 +959,12 @@
  </description>
 </property>

+<property>
+  <description>Network ACL to AM closed.</description>
+  <name>mapreduce.job.am-access-disabled</name>
+  <value>false</value>
+</property>
+
 <property>
  <name>mapreduce.job.acl-modify-job</name>
  <value> </value>