hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From varunsax...@apache.org
Subject [43/50] [abbrv] hadoop git commit: YARN-6130. [ATSv2 Security] Generate a delegation token for AM when app collector is created and pass it to AM via NM and RM. Contributed by Varun Saxena.
Date Wed, 30 Aug 2017 20:16:08 GMT
http://git-wip-us.apache.org/repos/asf/hadoop/blob/15712234/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/applicationsmanager/MockAsm.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/applicationsmanager/MockAsm.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/applicationsmanager/MockAsm.java
index 3234d6f..f826631 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/applicationsmanager/MockAsm.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/applicationsmanager/MockAsm.java
@@ -31,6 +31,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
 import org.apache.hadoop.yarn.api.records.ApplicationTimeoutType;
+import org.apache.hadoop.yarn.api.records.CollectorInfo;
 import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
@@ -239,6 +240,11 @@ public abstract class MockAsm extends MockApps {
     public boolean isAppInCompletedStates() {
       throw new UnsupportedOperationException("Not supported yet.");
     }
+
+    @Override
+    public CollectorInfo getCollectorInfo() {
+      throw new UnsupportedOperationException("Not supported yet.");
+    }
   }
 
   public static RMApp newApplication(int i) {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/15712234/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
index 9365e54..17cafef 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
@@ -32,6 +32,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
 import org.apache.hadoop.yarn.api.records.ApplicationTimeoutType;
+import org.apache.hadoop.yarn.api.records.CollectorInfo;
 import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
 import org.apache.hadoop.yarn.api.records.LogAggregationStatus;
 import org.apache.hadoop.yarn.api.records.NodeId;
@@ -321,13 +322,13 @@ public class MockRMApp implements RMApp {
     return false;
   }
 
-  public String getCollectorAddr() {
+  @Override
+  public AppCollectorData getCollectorData() {
     throw new UnsupportedOperationException("Not supported yet.");
   }
 
   @Override
-  public AppCollectorData getCollectorData() {
+  public CollectorInfo getCollectorInfo() {
     throw new UnsupportedOperationException("Not supported yet.");
   }
-
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/15712234/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/TestTimelineServiceClientIntegration.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/TestTimelineServiceClientIntegration.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/TestTimelineServiceClientIntegration.java
index 07058f6..eb4381d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/TestTimelineServiceClientIntegration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/TestTimelineServiceClientIntegration.java
@@ -80,7 +80,7 @@ public class TestTimelineServiceClientIntegration {
       auxService =
           PerNodeTimelineCollectorsAuxService.launchServer(new String[0],
               collectorManager, conf);
-      auxService.addApplication(ApplicationId.newInstance(0, 1));
+      auxService.addApplication(ApplicationId.newInstance(0, 1), "user");
     } catch (ExitUtil.ExitException e) {
       fail();
     }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/15712234/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/security/TestTimelineAuthFilterForV2.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/security/TestTimelineAuthFilterForV2.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/security/TestTimelineAuthFilterForV2.java
index da76958..84d892d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/security/TestTimelineAuthFilterForV2.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/security/TestTimelineAuthFilterForV2.java
@@ -23,7 +23,10 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.io.BufferedReader;
@@ -40,6 +43,7 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.http.HttpConfig;
+import org.apache.hadoop.io.Text;
 import org.apache.hadoop.minikdc.MiniKdc;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -51,10 +55,12 @@ import org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity;
 import org.apache.hadoop.yarn.client.api.TimelineV2Client;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier;
 import org.apache.hadoop.yarn.server.api.CollectorNodemanagerProtocol;
 import org.apache.hadoop.yarn.server.api.protocolrecords.GetTimelineCollectorContextRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.GetTimelineCollectorContextResponse;
 import org.apache.hadoop.yarn.server.timeline.security.TimelineAuthenticationFilterInitializer;
+import org.apache.hadoop.yarn.server.timelineservice.collector.AppLevelTimelineCollector;
 import org.apache.hadoop.yarn.server.timelineservice.collector.NodeTimelineCollectorManager;
 import org.apache.hadoop.yarn.server.timelineservice.collector.PerNodeTimelineCollectorsAuxService;
 import org.apache.hadoop.yarn.server.timelineservice.storage.FileSystemTimelineReaderImpl;
@@ -76,7 +82,6 @@ public class TestTimelineAuthFilterForV2 {
 
   private static final String FOO_USER = "foo";
   private static final String HTTP_USER = "HTTP";
-
   private static final File TEST_ROOT_DIR = new File(
       System.getProperty("test.build.dir", "target" + File.separator +
           "test-dir"), UUID.randomUUID().toString());
@@ -88,21 +93,35 @@ public class TestTimelineAuthFilterForV2 {
   private static String httpSpnegoPrincipal = KerberosTestUtils.
       getServerPrincipal();
 
+  // First param indicates whether HTTPS access or HTTP access and second param
+  // indicates whether it is kerberos access or token based access.
   @Parameterized.Parameters
-  public static Collection<Object[]> withSsl() {
-    return Arrays.asList(new Object[][] {{false}, {true}});
+  public static Collection<Object[]> params() {
+    return Arrays.asList(new Object[][] {{false, true}, {false, false},
+        {true, false}, {true, true}});
   }
 
   private static MiniKdc testMiniKDC;
   private static String keystoresDir;
   private static String sslConfDir;
   private static Configuration conf;
+  private static UserGroupInformation nonKerberosUser;
+  static {
+    try {
+      nonKerberosUser = UserGroupInformation.getCurrentUser();
+    } catch (IOException e) {}
+  }
+  // Indicates whether HTTPS or HTTP access.
   private boolean withSsl;
+  // Indicates whether Kerberos based login is used or token based access is
+  // done.
+  private boolean withKerberosLogin;
   private NodeTimelineCollectorManager collectorManager;
   private PerNodeTimelineCollectorsAuxService auxService;
-
-  public TestTimelineAuthFilterForV2(boolean withSsl) {
+  public TestTimelineAuthFilterForV2(boolean withSsl,
+      boolean withKerberosLogin) {
     this.withSsl = withSsl;
+    this.withKerberosLogin = withKerberosLogin;
   }
 
   @BeforeClass
@@ -143,8 +162,6 @@ public class TestTimelineAuthFilterForV2 {
       conf.set("hadoop.proxyuser.HTTP.hosts", "*");
       conf.set("hadoop.proxyuser.HTTP.users", FOO_USER);
       UserGroupInformation.setConfiguration(conf);
-      SecurityUtil.login(conf, YarnConfiguration.TIMELINE_SERVICE_KEYTAB,
-          YarnConfiguration.TIMELINE_SERVICE_PRINCIPAL, "localhost");
     } catch (Exception e) {
       fail("Couldn't setup TimelineServer V2.");
     }
@@ -166,9 +183,27 @@ public class TestTimelineAuthFilterForV2 {
       conf.set(YarnConfiguration.YARN_HTTP_POLICY_KEY,
           HttpConfig.Policy.HTTP_ONLY.name());
     }
+    UserGroupInformation.setConfiguration(conf);
     collectorManager = new DummyNodeTimelineCollectorManager();
-    auxService = PerNodeTimelineCollectorsAuxService.launchServer(new String[0],
-        collectorManager, conf);
+    auxService = PerNodeTimelineCollectorsAuxService.launchServer(
+        new String[0], collectorManager, conf);
+    if (withKerberosLogin) {
+      SecurityUtil.login(conf, YarnConfiguration.TIMELINE_SERVICE_KEYTAB,
+          YarnConfiguration.TIMELINE_SERVICE_PRINCIPAL, "localhost");
+    }
+    ApplicationId appId = ApplicationId.newInstance(0, 1);
+    auxService.addApplication(
+        appId, UserGroupInformation.getCurrentUser().getUserName());
+    if (!withKerberosLogin) {
+      AppLevelTimelineCollector collector =
+          (AppLevelTimelineCollector)collectorManager.get(appId);
+      org.apache.hadoop.security.token.Token
+          <TimelineDelegationTokenIdentifier> token =
+              collector.getDelegationTokenForApp();
+      token.setService(new Text("localhost" + token.getService().toString().
+          substring(token.getService().toString().indexOf(":"))));
+      UserGroupInformation.getCurrentUser().addToken(token);
+    }
   }
 
   private TimelineV2Client createTimelineClientForUGI(ApplicationId appId) {
@@ -199,9 +234,14 @@ public class TestTimelineAuthFilterForV2 {
     }
     if (withSsl) {
       KeyStoreTestUtil.cleanupSSLConfig(keystoresDir, sslConfDir);
-      File base = new File(BASEDIR);
-      FileUtil.fullyDelete(base);
+      FileUtil.fullyDelete(new File(BASEDIR));
+    }
+    if (withKerberosLogin) {
+      UserGroupInformation.getCurrentUser().logoutUserFromKeytab();
     }
+    // Reset the user for next run.
+    UserGroupInformation.setLoginUser(
+        UserGroupInformation.createRemoteUser(nonKerberosUser.getUserName()));
   }
 
   private static TimelineEntity createEntity(String id, String type) {
@@ -241,35 +281,44 @@ public class TestTimelineAuthFilterForV2 {
     }
   }
 
+  private void publishAndVerifyEntity(ApplicationId appId, File entityTypeDir,
+      String entityType) throws Exception {
+    TimelineV2Client client = createTimelineClientForUGI(appId);
+    try {
+    // Sync call. Results available immediately.
+      client.putEntities(createEntity("entity1", entityType));
+      assertEquals(1, entityTypeDir.listFiles().length);
+      verifyEntity(entityTypeDir, "entity1", entityType);
+      // Async call.
+      client.putEntitiesAsync(createEntity("entity2", entityType));
+    } finally {
+      client.stop();
+    }
+  }
+
   @Test
   public void testPutTimelineEntities() throws Exception {
-    final ApplicationId appId = ApplicationId.newInstance(0, 1);
-    auxService.addApplication(appId);
     final String entityType = "dummy_type";
+    final ApplicationId appId = ApplicationId.newInstance(0, 1);
     final File entityTypeDir = new File(TEST_ROOT_DIR.getAbsolutePath() +
         File.separator + "entities" + File.separator +
-        YarnConfiguration.DEFAULT_RM_CLUSTER_ID + File.separator + "test_user" +
+        YarnConfiguration.DEFAULT_RM_CLUSTER_ID + File.separator +
+        UserGroupInformation.getCurrentUser().getUserName() +
         File.separator + "test_flow_name" + File.separator +
         "test_flow_version" + File.separator + "1" + File.separator +
         appId.toString() + File.separator + entityType);
     try {
-      KerberosTestUtils.doAs(HTTP_USER + "/localhost", new Callable<Void>() {
-        @Override
-        public Void call() throws Exception {
-          TimelineV2Client client = createTimelineClientForUGI(appId);
-          try {
-            // Sync call. Results available immediately.
-            client.putEntities(createEntity("entity1", entityType));
-            assertEquals(1, entityTypeDir.listFiles().length);
-            verifyEntity(entityTypeDir, "entity1", entityType);
-            // Async call.
-            client.putEntitiesAsync(createEntity("entity2", entityType));
+      if (withKerberosLogin) {
+        KerberosTestUtils.doAs(HTTP_USER + "/localhost", new Callable<Void>() {
+          @Override
+          public Void call() throws Exception {
+            publishAndVerifyEntity(appId, entityTypeDir, entityType);
             return null;
-          } finally {
-            client.stop();
           }
-        }
-      });
+        });
+      } else {
+        publishAndVerifyEntity(appId, entityTypeDir, entityType);
+      }
       // Wait for async entity to be published.
       for (int i = 0; i < 50; i++) {
         if (entityTypeDir.listFiles().length == 2) {
@@ -279,6 +328,11 @@ public class TestTimelineAuthFilterForV2 {
       }
       assertEquals(2, entityTypeDir.listFiles().length);
       verifyEntity(entityTypeDir, "entity2", entityType);
+      AppLevelTimelineCollector collector =
+          (AppLevelTimelineCollector)collectorManager.get(appId);
+      auxService.removeApplication(appId);
+      verify(collectorManager.getTokenManagerService()).cancelToken(
+          eq(collector.getDelegationTokenForApp()), any(String.class));
     } finally {
       FileUtils.deleteQuietly(entityTypeDir);
     }
@@ -291,12 +345,19 @@ public class TestTimelineAuthFilterForV2 {
     }
 
     @Override
+    protected TimelineV2DelegationTokenSecretManagerService
+        createTokenManagerService() {
+      return spy(new TimelineV2DelegationTokenSecretManagerService());
+    }
+
+    @Override
     protected CollectorNodemanagerProtocol getNMCollectorService() {
       CollectorNodemanagerProtocol protocol =
           mock(CollectorNodemanagerProtocol.class);
       try {
         GetTimelineCollectorContextResponse response =
-            GetTimelineCollectorContextResponse.newInstance("test_user",
+            GetTimelineCollectorContextResponse.newInstance(
+                UserGroupInformation.getCurrentUser().getUserName(),
                 "test_flow_name", "test_flow_version", 1L);
         when(protocol.getTimelineCollectorContext(any(
             GetTimelineCollectorContextRequest.class))).thenReturn(response);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/15712234/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/AppLevelTimelineCollector.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/AppLevelTimelineCollector.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/AppLevelTimelineCollector.java
index c481dbe..13426b9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/AppLevelTimelineCollector.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/AppLevelTimelineCollector.java
@@ -24,9 +24,12 @@ import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 
 /**
@@ -41,13 +44,20 @@ public class AppLevelTimelineCollector extends TimelineCollector {
   private static final Log LOG = LogFactory.getLog(TimelineCollector.class);
 
   private final ApplicationId appId;
+  private final String appUser;
   private final TimelineCollectorContext context;
   private UserGroupInformation currentUser;
+  private Token<TimelineDelegationTokenIdentifier> delegationTokenForApp;
 
   public AppLevelTimelineCollector(ApplicationId appId) {
+    this(appId, null);
+  }
+
+  public AppLevelTimelineCollector(ApplicationId appId, String user) {
     super(AppLevelTimelineCollector.class.getName() + " - " + appId.toString());
     Preconditions.checkNotNull(appId, "AppId shouldn't be null");
     this.appId = appId;
+    this.appUser = user;
     context = new TimelineCollectorContext();
   }
 
@@ -55,6 +65,20 @@ public class AppLevelTimelineCollector extends TimelineCollector {
     return currentUser;
   }
 
+  public String getAppUser() {
+    return appUser;
+  }
+
+  void setDelegationTokenForApp(
+      Token<TimelineDelegationTokenIdentifier> token) {
+    this.delegationTokenForApp = token;
+  }
+
+  @VisibleForTesting
+  public Token<TimelineDelegationTokenIdentifier> getDelegationTokenForApp() {
+    return this.delegationTokenForApp;
+  }
+
   @Override
   protected void serviceInit(Configuration conf) throws Exception {
     context.setClusterId(conf.get(YarnConfiguration.RM_CLUSTER_ID,

http://git-wip-us.apache.org/repos/asf/hadoop/blob/15712234/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/AppLevelTimelineCollectorWithAgg.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/AppLevelTimelineCollectorWithAgg.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/AppLevelTimelineCollectorWithAgg.java
index ac91275..6c0d693 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/AppLevelTimelineCollectorWithAgg.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/AppLevelTimelineCollectorWithAgg.java
@@ -56,8 +56,8 @@ public class AppLevelTimelineCollectorWithAgg
   private ScheduledThreadPoolExecutor appAggregationExecutor;
   private AppLevelAggregator appAggregator;
 
-  public AppLevelTimelineCollectorWithAgg(ApplicationId appId) {
-    super(appId);
+  public AppLevelTimelineCollectorWithAgg(ApplicationId appId, String user) {
+    super(appId, user);
   }
 
   private static Set<String> initializeSkipSet() {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/15712234/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/NodeTimelineCollectorManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/NodeTimelineCollectorManager.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/NodeTimelineCollectorManager.java
index 50ebb0f..cad993d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/NodeTimelineCollectorManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/NodeTimelineCollectorManager.java
@@ -30,14 +30,17 @@ import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.http.HttpServer2;
+import org.apache.hadoop.io.Text;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
 import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier;
 import org.apache.hadoop.yarn.server.api.CollectorNodemanagerProtocol;
 import org.apache.hadoop.yarn.server.api.protocolrecords.GetTimelineCollectorContextRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.GetTimelineCollectorContextResponse;
@@ -71,6 +74,8 @@ public class NodeTimelineCollectorManager extends TimelineCollectorManager
{
 
   private final boolean runningAsAuxService;
 
+  private UserGroupInformation loginUGI;
+
   static final String COLLECTOR_MANAGER_ATTR_KEY = "collector.manager";
 
   @VisibleForTesting
@@ -85,25 +90,40 @@ public class NodeTimelineCollectorManager extends TimelineCollectorManager
{
 
   @Override
   protected void serviceInit(Configuration conf) throws Exception {
-    tokenMgrService = new TimelineV2DelegationTokenSecretManagerService();
+    tokenMgrService = createTokenManagerService();
     addService(tokenMgrService);
+    this.loginUGI = UserGroupInformation.getCurrentUser();
     super.serviceInit(conf);
   }
 
   @Override
   protected void serviceStart() throws Exception {
-    if (UserGroupInformation.isSecurityEnabled() && !runningAsAuxService) {
+    if (UserGroupInformation.isSecurityEnabled()) {
       // Do security login for cases where collector is running outside NM.
-      try {
-        doSecureLogin();
-      } catch(IOException ie) {
-        throw new YarnRuntimeException("Failed to login", ie);
+      if (!runningAsAuxService) {
+        try {
+          doSecureLogin();
+        } catch(IOException ie) {
+          throw new YarnRuntimeException("Failed to login", ie);
+        }
       }
+      this.loginUGI = UserGroupInformation.getLoginUser();
     }
     super.serviceStart();
     startWebApp();
   }
 
+  protected TimelineV2DelegationTokenSecretManagerService
+      createTokenManagerService() {
+    return new TimelineV2DelegationTokenSecretManagerService();
+  }
+
+  @VisibleForTesting
+  public TimelineV2DelegationTokenSecretManagerService
+      getTokenManagerService() {
+    return tokenMgrService;
+  }
+
   private void doSecureLogin() throws IOException {
     Configuration conf = getConfig();
     InetSocketAddress addr = NetUtils.createSocketAddr(conf.getTrimmed(
@@ -122,13 +142,45 @@ public class NodeTimelineCollectorManager extends TimelineCollectorManager
{
     super.serviceStop();
   }
 
+  @VisibleForTesting
+  public Token<TimelineDelegationTokenIdentifier> generateTokenForAppCollector(
+      String user) {
+    Token<TimelineDelegationTokenIdentifier> token  = tokenMgrService.
+        generateToken(UserGroupInformation.createRemoteUser(user),
+            loginUGI.getShortUserName());
+    token.setService(new Text(timelineRestServerBindAddress));
+    return token;
+  }
+
+  @VisibleForTesting
+  public void cancelTokenForAppCollector(
+      AppLevelTimelineCollector appCollector) throws IOException {
+    if (appCollector.getDelegationTokenForApp() != null) {
+      tokenMgrService.cancelToken(appCollector.getDelegationTokenForApp(),
+          appCollector.getAppUser());
+    }
+  }
+
   @Override
   protected void doPostPut(ApplicationId appId, TimelineCollector collector) {
     try {
       // Get context info from NM
       updateTimelineCollectorContext(appId, collector);
+      // Generate token for app collector.
+      org.apache.hadoop.yarn.api.records.Token token = null;
+      if (UserGroupInformation.isSecurityEnabled() &&
+          collector instanceof AppLevelTimelineCollector) {
+        AppLevelTimelineCollector appCollector =
+            (AppLevelTimelineCollector)collector;
+        Token<TimelineDelegationTokenIdentifier> timelineToken =
+            generateTokenForAppCollector(appCollector.getAppUser());
+        appCollector.setDelegationTokenForApp(timelineToken);
+        token = org.apache.hadoop.yarn.api.records.Token.newInstance(
+            timelineToken.getIdentifier(), timelineToken.getKind().toString(),
+            timelineToken.getPassword(), timelineToken.getService().toString());
+      }
       // Report to NM if a new collector is added.
-      reportNewCollectorToNM(appId);
+      reportNewCollectorToNM(appId, token);
     } catch (YarnException | IOException e) {
       // throw exception here as it cannot be used if failed communicate with NM
       LOG.error("Failed to communicate with NM Collector Service for " + appId);
@@ -136,6 +188,18 @@ public class NodeTimelineCollectorManager extends TimelineCollectorManager
{
     }
   }
 
+  @Override
+  protected void postRemove(ApplicationId appId, TimelineCollector collector) {
+    if (collector instanceof AppLevelTimelineCollector) {
+      try {
+        cancelTokenForAppCollector((AppLevelTimelineCollector)collector);
+      } catch (IOException e) {
+        LOG.warn("Failed to cancel token for app collector with appId " +
+            appId, e);
+      }
+    }
+  }
+
   /**
    * Launch the REST web server for this collector manager.
    */
@@ -180,11 +244,12 @@ public class NodeTimelineCollectorManager extends TimelineCollectorManager
{
         timelineRestServerBindAddress);
   }
 
-  private void reportNewCollectorToNM(ApplicationId appId)
+  private void reportNewCollectorToNM(ApplicationId appId,
+      org.apache.hadoop.yarn.api.records.Token token)
       throws YarnException, IOException {
     ReportNewCollectorInfoRequest request =
         ReportNewCollectorInfoRequest.newInstance(appId,
-            this.timelineRestServerBindAddress);
+            this.timelineRestServerBindAddress, token);
     LOG.info("Report a new collector for application: " + appId +
         " to the NM Collector Service.");
     getNMCollectorService().reportNewCollectorInfo(request);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/15712234/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/PerNodeTimelineCollectorsAuxService.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/PerNodeTimelineCollectorsAuxService.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/PerNodeTimelineCollectorsAuxService.java
index 669e752..cb48e72 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/PerNodeTimelineCollectorsAuxService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/PerNodeTimelineCollectorsAuxService.java
@@ -114,11 +114,12 @@ public class PerNodeTimelineCollectorsAuxService extends AuxiliaryService
{
    * exists, no new service is created.
    *
    * @param appId Application Id to be added.
+   * @param user Application Master container user.
    * @return whether it was added successfully
    */
-  public boolean addApplication(ApplicationId appId) {
+  public boolean addApplication(ApplicationId appId, String user) {
     AppLevelTimelineCollector collector =
-        new AppLevelTimelineCollectorWithAgg(appId);
+        new AppLevelTimelineCollectorWithAgg(appId, user);
     return (collectorManager.putIfAbsent(appId, collector)
         == collector);
   }
@@ -147,7 +148,7 @@ public class PerNodeTimelineCollectorsAuxService extends AuxiliaryService
{
     if (context.getContainerType() == ContainerType.APPLICATION_MASTER) {
       ApplicationId appId = context.getContainerId().
           getApplicationAttemptId().getApplicationId();
-      addApplication(appId);
+      addApplication(appId, context.getUser());
     }
   }
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/15712234/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/security/TimelineV2DelegationTokenSecretManagerService.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/security/TimelineV2DelegationTokenSecretManagerService.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/security/TimelineV2DelegationTokenSecretManagerService.java
index eef8436..de7db58 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/security/TimelineV2DelegationTokenSecretManagerService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/security/TimelineV2DelegationTokenSecretManagerService.java
@@ -18,8 +18,13 @@
 
 package org.apache.hadoop.yarn.server.timelineservice.security;
 
+import java.io.IOException;
+
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenSecretManager;
 import org.apache.hadoop.yarn.security.client.TimelineDelegationTokenIdentifier;
 import org.apache.hadoop.yarn.server.timeline.security.TimelineDelgationTokenSecretManagerService;
@@ -43,6 +48,17 @@ public class TimelineV2DelegationTokenSecretManagerService extends
         tokenMaxLifetime, tokenRenewInterval, tokenRemovalScanInterval);
   }
 
+  public Token<TimelineDelegationTokenIdentifier> generateToken(
+      UserGroupInformation ugi, String renewer) {
+    return ((TimelineV2DelegationTokenSecretManager)
+        getTimelineDelegationTokenSecretManager()).generateToken(ugi, renewer);
+  }
+
+  public void cancelToken(Token<TimelineDelegationTokenIdentifier> token,
+      String canceller) throws IOException {
+    getTimelineDelegationTokenSecretManager().cancelToken(token, canceller);
+  }
+
   /**
    * Delegation token secret manager for ATSv2.
    */
@@ -70,6 +86,21 @@ public class TimelineV2DelegationTokenSecretManagerService extends
           delegationTokenRenewInterval, delegationTokenRemoverScanInterval);
     }
 
+    public Token<TimelineDelegationTokenIdentifier> generateToken(
+        UserGroupInformation ugi, String renewer) {
+      Text realUser = null;
+      if (ugi.getRealUser() != null) {
+        realUser = new Text(ugi.getRealUser().getUserName());
+      }
+      TimelineDelegationTokenIdentifier identifier = createIdentifier();
+      identifier.setOwner(new Text(ugi.getUserName()));
+      identifier.setRenewer(new Text(renewer));
+      identifier.setRealUser(realUser);
+      byte[] password = createPassword(identifier);
+      return new Token<TimelineDelegationTokenIdentifier>(identifier.getBytes(),
+          password, identifier.getKind(), null);
+    }
+
     @Override
     public TimelineDelegationTokenIdentifier createIdentifier() {
       return new TimelineDelegationTokenIdentifier();

http://git-wip-us.apache.org/repos/asf/hadoop/blob/15712234/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/collector/TestNMTimelineCollectorManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/collector/TestNMTimelineCollectorManager.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/collector/TestNMTimelineCollectorManager.java
index a59f8c1..af9acce 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/collector/TestNMTimelineCollectorManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/collector/TestNMTimelineCollectorManager.java
@@ -95,7 +95,7 @@ public class TestNMTimelineCollectorManager {
       Callable<Boolean> task = new Callable<Boolean>() {
         public Boolean call() {
           AppLevelTimelineCollector collector =
-              new AppLevelTimelineCollectorWithAgg(appId);
+              new AppLevelTimelineCollectorWithAgg(appId, "user");
           return (collectorManager.putIfAbsent(appId, collector) == collector);
         }
       };
@@ -126,7 +126,7 @@ public class TestNMTimelineCollectorManager {
       Callable<Boolean> task = new Callable<Boolean>() {
         public Boolean call() {
           AppLevelTimelineCollector collector =
-              new AppLevelTimelineCollectorWithAgg(appId);
+              new AppLevelTimelineCollectorWithAgg(appId, "user");
           boolean successPut =
               (collectorManager.putIfAbsent(appId, collector) == collector);
           return successPut && collectorManager.remove(appId);


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message