hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jl...@apache.org
Subject [1/2] YARN-90. NodeManager should identify failed disks becoming good again. Contributed by Varun Vasudev (cherry picked from commit 6f2028bd1514d90b831f889fd0ee7f2ba5c15000)
Date Tue, 21 Oct 2014 17:44:31 GMT
Repository: hadoop
Updated Branches:
  refs/heads/branch-2.6 fcdeba204 -> 413fe4cb0


http://git-wip-us.apache.org/repos/asf/hadoop/blob/413fe4cb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java
index ab86a18..8a5441a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java
@@ -25,6 +25,7 @@ import static org.mockito.Matchers.anyMap;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Matchers.isA;
 import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -58,6 +59,8 @@ import org.junit.Assert;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.AbstractFileSystem;
+import org.apache.hadoop.fs.FileContext;
 import org.apache.hadoop.fs.FileContext;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
@@ -105,6 +108,7 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.TestContainerM
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.Application;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEvent;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.TestNonAggregatingLogHandler;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.event.LogHandlerAppFinishedEvent;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.event.LogHandlerAppStartedEvent;
@@ -136,12 +140,19 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
     super();
     this.remoteRootLogDir.mkdir();
   }
+  
+  DrainDispatcher dispatcher;
+  EventHandler<ApplicationEvent> appEventHandler;
 
   @Override
+  @SuppressWarnings("unchecked")
   public void setup() throws IOException {
     super.setup();
     NodeId nodeId = NodeId.newInstance("0.0.0.0", 5555);
     ((NMContext)context).setNodeId(nodeId);
+    dispatcher = createDispatcher();
+    appEventHandler = mock(EventHandler.class);
+    dispatcher.register(ApplicationEventType.class, appEventHandler);
   }
 
   @Override
@@ -149,10 +160,12 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
     super.tearDown();
     createContainerExecutor().deleteAsUser(user,
         new Path(remoteRootLogDir.getAbsolutePath()), new Path[] {});
+    dispatcher.await();
+    dispatcher.stop();
+    dispatcher.close();
   }
 
   @Test
-  @SuppressWarnings("unchecked")
   public void testLocalFileDeletionAfterUpload() throws Exception {
     this.delSrvc = new DeletionService(createContainerExecutor());
     delSrvc = spy(delSrvc);
@@ -161,10 +174,6 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
     this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
         this.remoteRootLogDir.getAbsolutePath());
     
-    DrainDispatcher dispatcher = createDispatcher();
-    EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class);
-    dispatcher.register(ApplicationEventType.class, appEventHandler);
-    
     LogAggregationService logAggregationService = spy(
         new LogAggregationService(dispatcher, this.context, this.delSrvc,
                                   super.dirsHandler));
@@ -236,16 +245,11 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
   }
 
   @Test
-  @SuppressWarnings("unchecked")
   public void testNoContainerOnNode() throws Exception {
     this.conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDir.getAbsolutePath());
     this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
         this.remoteRootLogDir.getAbsolutePath());
     
-    DrainDispatcher dispatcher = createDispatcher();
-    EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class);
-    dispatcher.register(ApplicationEventType.class, appEventHandler);
-    
     LogAggregationService logAggregationService =
         new LogAggregationService(dispatcher, this.context, this.delSrvc,
                                   super.dirsHandler);
@@ -285,6 +289,7 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
     };
     checkEvents(appEventHandler, expectedEvents, true, "getType", "getApplicationID");
     dispatcher.stop();
+    logAggregationService.close();
   }
 
   @Test
@@ -294,6 +299,7 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
     this.conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDir.getAbsolutePath());
     this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
         this.remoteRootLogDir.getAbsolutePath());
+    
     String[] fileNames = new String[] { "stdout", "stderr", "syslog" };
     DrainDispatcher dispatcher = createDispatcher();
     EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class);
@@ -432,17 +438,12 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
   }
   
   @Test
-  @SuppressWarnings("unchecked")
   public void testVerifyAndCreateRemoteDirsFailure()
       throws Exception {
     this.conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDir.getAbsolutePath());
     this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
         this.remoteRootLogDir.getAbsolutePath());
     
-    DrainDispatcher dispatcher = createDispatcher();
-    EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class);
-    dispatcher.register(ApplicationEventType.class, appEventHandler);
-    
     LogAggregationService logAggregationService = spy(
         new LogAggregationService(dispatcher, this.context, this.delSrvc,
                                   super.dirsHandler));
@@ -456,8 +457,9 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
     logAggregationService.start();
     
     // Now try to start an application
-    ApplicationId appId = BuilderUtils.newApplicationId(
-        System.currentTimeMillis(), (int)Math.random());
+    ApplicationId appId =
+        BuilderUtils.newApplicationId(System.currentTimeMillis(),
+          (int) (Math.random() * 1000));
     logAggregationService.handle(new LogHandlerAppStartedEvent(appId,
         this.user, null,
         ContainerLogsRetentionPolicy.AM_AND_FAILED_CONTAINERS_ONLY,
@@ -475,8 +477,9 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
     Mockito.reset(logAggregationService);
     
     // Now try to start another one
-    ApplicationId appId2 = BuilderUtils.newApplicationId(
-        System.currentTimeMillis(), (int)Math.random());
+    ApplicationId appId2 =
+        BuilderUtils.newApplicationId(System.currentTimeMillis(),
+          (int) (Math.random() * 1000));
     File appLogDir =
         new File(localLogDir, ConverterUtils.toString(appId2));
     appLogDir.mkdir();
@@ -578,6 +581,8 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
     aggSvc.handle(new LogHandlerAppStartedEvent(appId3, this.user, null,
         ContainerLogsRetentionPolicy.ALL_CONTAINERS, this.acls));
     verify(spyFs, never()).mkdirs(eq(appDir3), isA(FsPermission.class));
+    aggSvc.stop();
+    aggSvc.close();
   }
 
   @Test
@@ -588,19 +593,16 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
         localLogDir.getAbsolutePath());
     this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
         this.remoteRootLogDir.getAbsolutePath());
-
-    DrainDispatcher dispatcher = createDispatcher();
-    EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class);
-    dispatcher.register(ApplicationEventType.class, appEventHandler);
-
+    
     LogAggregationService logAggregationService = spy(
         new LogAggregationService(dispatcher, this.context, this.delSrvc,
                                   super.dirsHandler));
     logAggregationService.init(this.conf);
     logAggregationService.start();
 
-    ApplicationId appId = BuilderUtils.newApplicationId(
-        System.currentTimeMillis(), (int)Math.random());
+    ApplicationId appId =
+        BuilderUtils.newApplicationId(System.currentTimeMillis(),
+          (int) (Math.random() * 1000));
     doThrow(new YarnRuntimeException("KABOOM!"))
       .when(logAggregationService).initAppAggregator(
           eq(appId), eq(user), any(Credentials.class),
@@ -634,26 +636,22 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
   }
 
   @Test
-  @SuppressWarnings("unchecked")
   public void testLogAggregationCreateDirsFailsWithoutKillingNM()
       throws Exception {
     
     this.conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDir.getAbsolutePath());
     this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
         this.remoteRootLogDir.getAbsolutePath());
-    
-    DrainDispatcher dispatcher = createDispatcher();
-    EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class);
-    dispatcher.register(ApplicationEventType.class, appEventHandler);
-    
+        
     LogAggregationService logAggregationService = spy(
         new LogAggregationService(dispatcher, this.context, this.delSrvc,
                                   super.dirsHandler));
     logAggregationService.init(this.conf);
     logAggregationService.start();
     
-    ApplicationId appId = BuilderUtils.newApplicationId(
-        System.currentTimeMillis(), (int)Math.random());
+    ApplicationId appId =
+        BuilderUtils.newApplicationId(System.currentTimeMillis(),
+          (int) (Math.random() * 1000));
     Exception e = new RuntimeException("KABOOM!");
     doThrow(e)
       .when(logAggregationService).createAppDir(any(String.class),
@@ -905,18 +903,13 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
   }
 
   @Test(timeout=20000)
-  @SuppressWarnings("unchecked")
   public void testStopAfterError() throws Exception {
     DeletionService delSrvc = mock(DeletionService.class);
 
     // get the AppLogAggregationImpl thread to crash
     LocalDirsHandlerService mockedDirSvc = mock(LocalDirsHandlerService.class);
     when(mockedDirSvc.getLogDirs()).thenThrow(new RuntimeException());
-
-    DrainDispatcher dispatcher = createDispatcher();
-    EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class);
-    dispatcher.register(ApplicationEventType.class, appEventHandler);
-
+    
     LogAggregationService logAggregationService =
         new LogAggregationService(dispatcher, this.context, delSrvc,
                                   mockedDirSvc);
@@ -930,20 +923,16 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
 
     logAggregationService.stop();
     assertEquals(0, logAggregationService.getNumAggregators());
+    logAggregationService.close();
   }
 
   @Test
-  @SuppressWarnings("unchecked")
   public void testLogAggregatorCleanup() throws Exception {
     DeletionService delSrvc = mock(DeletionService.class);
 
     // get the AppLogAggregationImpl thread to crash
     LocalDirsHandlerService mockedDirSvc = mock(LocalDirsHandlerService.class);
 
-    DrainDispatcher dispatcher = createDispatcher();
-    EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class);
-    dispatcher.register(ApplicationEventType.class, appEventHandler);
-
     LogAggregationService logAggregationService =
         new LogAggregationService(dispatcher, this.context, delSrvc,
                                   mockedDirSvc);
@@ -964,6 +953,8 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
     }
     Assert.assertEquals("Log aggregator failed to cleanup!", 0,
         logAggregationService.getNumAggregators());
+    logAggregationService.stop();
+    logAggregationService.close();
   }
   
   @SuppressWarnings("unchecked")
@@ -1039,6 +1030,72 @@ public class TestLogAggregationService extends BaseContainerManagerTest
{
     return sb.toString();
   }
 
+  /*
+   * Test to make sure we handle cases where the directories we get back from
+   * the LocalDirsHandler may have issues including the log dir not being
+   * present as well as other issues. The test uses helper functions from
+   * TestNonAggregatingLogHandler.
+   */
+  @Test
+  public void testFailedDirsLocalFileDeletionAfterUpload() throws Exception {
+
+    // setup conf and services
+    DeletionService mockDelService = mock(DeletionService.class);
+    File[] localLogDirs =
+        TestNonAggregatingLogHandler.getLocalLogDirFiles(this.getClass()
+          .getName(), 7);
+    final List<String> localLogDirPaths =
+        new ArrayList<String>(localLogDirs.length);
+    for (int i = 0; i < localLogDirs.length; i++) {
+      localLogDirPaths.add(localLogDirs[i].getAbsolutePath());
+    }
+
+    String localLogDirsString = StringUtils.join(localLogDirPaths, ",");
+
+    this.conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDirsString);
+    this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
+      this.remoteRootLogDir.getAbsolutePath());
+    this.conf.setLong(YarnConfiguration.NM_DISK_HEALTH_CHECK_INTERVAL_MS, 500);
+
+    ApplicationId application1 = BuilderUtils.newApplicationId(1234, 1);
+    ApplicationAttemptId appAttemptId =
+        BuilderUtils.newApplicationAttemptId(application1, 1);
+
+    this.dirsHandler = new LocalDirsHandlerService();
+    LocalDirsHandlerService mockDirsHandler = mock(LocalDirsHandlerService.class);
+
+    LogAggregationService logAggregationService =
+        spy(new LogAggregationService(dispatcher, this.context, mockDelService,
+          mockDirsHandler));
+    AbstractFileSystem spylfs =
+        spy(FileContext.getLocalFSFileContext().getDefaultFileSystem());
+    FileContext lfs = FileContext.getFileContext(spylfs, conf);
+    doReturn(lfs).when(logAggregationService).getLocalFileContext(
+      isA(Configuration.class));
+
+    logAggregationService.init(this.conf);
+    logAggregationService.start();
+
+    TestNonAggregatingLogHandler.runMockedFailedDirs(logAggregationService,
+      application1, user, mockDelService, mockDirsHandler, conf, spylfs, lfs,
+      localLogDirs);
+
+    logAggregationService.stop();
+    assertEquals(0, logAggregationService.getNumAggregators());
+    verify(logAggregationService).closeFileSystems(
+      any(UserGroupInformation.class));
+
+    ApplicationEvent expectedEvents[] =
+        new ApplicationEvent[] {
+            new ApplicationEvent(appAttemptId.getApplicationId(),
+              ApplicationEventType.APPLICATION_LOG_HANDLING_INITED),
+            new ApplicationEvent(appAttemptId.getApplicationId(),
+              ApplicationEventType.APPLICATION_LOG_HANDLING_FINISHED) };
+
+    checkEvents(appEventHandler, expectedEvents, true, "getType",
+      "getApplicationID");
+  }
+
   @Test (timeout = 50000)
   @SuppressWarnings("unchecked")
   public void testLogAggregationServiceWithPatterns() throws Exception {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/413fe4cb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/loghandler/TestNonAggregatingLogHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/loghandler/TestNonAggregatingLogHandler.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/loghandler/TestNonAggregatingLogHandler.java
index 300ca28..d0f6472 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/loghandler/TestNonAggregatingLogHandler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/loghandler/TestNonAggregatingLogHandler.java
@@ -19,15 +19,36 @@ package org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler;
 
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 
 import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.NotSerializableException;
+import java.io.ObjectInputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.builder.EqualsBuilder;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.AbstractFileSystem;
+import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.UnsupportedFileSystemException;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -45,25 +66,52 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.eve
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.event.LogHandlerAppStartedEvent;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.event.LogHandlerContainerFinishedEvent;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
+import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
+import org.mockito.ArgumentMatcher;
+import org.mockito.Mockito;
 import org.mockito.exceptions.verification.WantedButNotInvoked;
+import org.mockito.internal.matchers.VarargMatcher;
 
 public class TestNonAggregatingLogHandler {
+  
+  DeletionService mockDelService;
+  Configuration conf;
+  DrainDispatcher dispatcher;
+  EventHandler<ApplicationEvent> appEventHandler;
+  String user = "testuser";
+  ApplicationId appId;
+  ApplicationAttemptId appAttemptId;
+  ContainerId container11;
+  LocalDirsHandlerService dirsHandler;
+
+  @Before
+  @SuppressWarnings("unchecked")
+  public void setup() {
+    mockDelService = mock(DeletionService.class);
+    conf = new YarnConfiguration();
+    dispatcher = createDispatcher(conf);
+    appEventHandler = mock(EventHandler.class);
+    dispatcher.register(ApplicationEventType.class, appEventHandler);
+    appId = BuilderUtils.newApplicationId(1234, 1);
+    appAttemptId = BuilderUtils.newApplicationAttemptId(appId, 1);
+    container11 = BuilderUtils.newContainerId(appAttemptId, 1);
+    dirsHandler = new LocalDirsHandlerService();
+  }
+
+  @After
+  public void tearDown() throws IOException {
+    dirsHandler.stop();
+    dirsHandler.close();
+    dispatcher.await();
+    dispatcher.stop();
+    dispatcher.close();
+  }  
 
   @Test
-  @SuppressWarnings("unchecked")
-  public void testLogDeletion() {
-    DeletionService delService = mock(DeletionService.class);
-    Configuration conf = new YarnConfiguration();
-    String user = "testuser";
-
-    File[] localLogDirs = new File[2];
-    localLogDirs[0] =
-        new File("target", this.getClass().getName() + "-localLogDir0")
-            .getAbsoluteFile();
-    localLogDirs[1] =
-        new File("target", this.getClass().getName() + "-localLogDir1")
-            .getAbsoluteFile();
+  public void testLogDeletion() throws IOException {
+    File[] localLogDirs = getLocalLogDirFiles(this.getClass().getName(), 2);
     String localLogDirsString =
         localLogDirs[0].getAbsolutePath() + ","
             + localLogDirs[1].getAbsolutePath();
@@ -72,72 +120,50 @@ public class TestNonAggregatingLogHandler {
     conf.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, false);
     conf.setLong(YarnConfiguration.NM_LOG_RETAIN_SECONDS, 0l);
 
-    DrainDispatcher dispatcher = createDispatcher(conf);
-    EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class);
-    dispatcher.register(ApplicationEventType.class, appEventHandler);
-
-    LocalDirsHandlerService dirsHandler = new LocalDirsHandlerService();
     dirsHandler.init(conf);
 
-    ApplicationId appId1 = BuilderUtils.newApplicationId(1234, 1);
-    ApplicationAttemptId appAttemptId1 =
-        BuilderUtils.newApplicationAttemptId(appId1, 1);
-    ContainerId container11 = BuilderUtils.newContainerId(appAttemptId1, 1);
+    NonAggregatingLogHandler rawLogHandler =
+        new NonAggregatingLogHandler(dispatcher, mockDelService, dirsHandler);
+    NonAggregatingLogHandler logHandler = spy(rawLogHandler);
+    AbstractFileSystem spylfs =
+        spy(FileContext.getLocalFSFileContext().getDefaultFileSystem());
+    FileContext lfs = FileContext.getFileContext(spylfs, conf);
+    doReturn(lfs).when(logHandler)
+      .getLocalFileContext(isA(Configuration.class));
+    FsPermission defaultPermission =
+        FsPermission.getDirDefault().applyUMask(lfs.getUMask());
+    final FileStatus fs =
+        new FileStatus(0, true, 1, 0, System.currentTimeMillis(), 0,
+          defaultPermission, "", "",
+          new Path(localLogDirs[0].getAbsolutePath()));
+    doReturn(fs).when(spylfs).getFileStatus(isA(Path.class));
 
-    NonAggregatingLogHandler logHandler =
-        new NonAggregatingLogHandler(dispatcher, delService, dirsHandler);
     logHandler.init(conf);
     logHandler.start();
 
-    logHandler.handle(new LogHandlerAppStartedEvent(appId1, user, null,
+    logHandler.handle(new LogHandlerAppStartedEvent(appId, user, null,
         ContainerLogsRetentionPolicy.ALL_CONTAINERS, null));
 
     logHandler.handle(new LogHandlerContainerFinishedEvent(container11, 0));
 
-    logHandler.handle(new LogHandlerAppFinishedEvent(appId1));
+    logHandler.handle(new LogHandlerAppFinishedEvent(appId));
 
     Path[] localAppLogDirs = new Path[2];
     localAppLogDirs[0] =
-        new Path(localLogDirs[0].getAbsolutePath(), appId1.toString());
+        new Path(localLogDirs[0].getAbsolutePath(), appId.toString());
     localAppLogDirs[1] =
-        new Path(localLogDirs[1].getAbsolutePath(), appId1.toString());
+        new Path(localLogDirs[1].getAbsolutePath(), appId.toString());
 
-    // 5 seconds for the delete which is a separate thread.
-    long verifyStartTime = System.currentTimeMillis();
-    WantedButNotInvoked notInvokedException = null;
-    boolean matched = false;
-    while (!matched && System.currentTimeMillis() < verifyStartTime + 5000l) {
-      try {
-        verify(delService).delete(eq(user), (Path) eq(null),
-            eq(localAppLogDirs[0]), eq(localAppLogDirs[1]));
-        matched = true;
-      } catch (WantedButNotInvoked e) {
-        notInvokedException = e;
-        try {
-          Thread.sleep(50l);
-        } catch (InterruptedException i) {
-        }
-      }
-    }
-    if (!matched) {
-      throw notInvokedException;
+    testDeletionServiceCall(mockDelService, user, 5000, localAppLogDirs);
+    logHandler.close();
+    for (int i = 0; i < localLogDirs.length; i++) {
+      FileUtils.deleteDirectory(localLogDirs[i]);
     }
   }
 
   @Test
-  @SuppressWarnings("unchecked")
-  public void testDelayedDelete() {
-    DeletionService delService = mock(DeletionService.class);
-    Configuration conf = new YarnConfiguration();
-    String user = "testuser";
-
-    File[] localLogDirs = new File[2];
-    localLogDirs[0] =
-        new File("target", this.getClass().getName() + "-localLogDir0")
-            .getAbsoluteFile();
-    localLogDirs[1] =
-        new File("target", this.getClass().getName() + "-localLogDir1")
-            .getAbsoluteFile();
+  public void testDelayedDelete() throws IOException {
+    File[] localLogDirs = getLocalLogDirFiles(this.getClass().getName(), 2);
     String localLogDirsString =
         localLogDirs[0].getAbsolutePath() + ","
             + localLogDirs[1].getAbsolutePath();
@@ -148,42 +174,36 @@ public class TestNonAggregatingLogHandler {
     conf.setLong(YarnConfiguration.NM_LOG_RETAIN_SECONDS,
             YarnConfiguration.DEFAULT_NM_LOG_RETAIN_SECONDS);
 
-    DrainDispatcher dispatcher = createDispatcher(conf);
-    EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class);
-    dispatcher.register(ApplicationEventType.class, appEventHandler);
-
-    LocalDirsHandlerService dirsHandler = new LocalDirsHandlerService();
     dirsHandler.init(conf);
 
-    ApplicationId appId1 = BuilderUtils.newApplicationId(1234, 1);
-    ApplicationAttemptId appAttemptId1 =
-        BuilderUtils.newApplicationAttemptId(appId1, 1);
-    ContainerId container11 = BuilderUtils.newContainerId(appAttemptId1, 1);
-
     NonAggregatingLogHandler logHandler =
-        new NonAggregatingLogHandlerWithMockExecutor(dispatcher, delService,
+        new NonAggregatingLogHandlerWithMockExecutor(dispatcher, mockDelService,
                                                      dirsHandler);
     logHandler.init(conf);
     logHandler.start();
 
-    logHandler.handle(new LogHandlerAppStartedEvent(appId1, user, null,
+    logHandler.handle(new LogHandlerAppStartedEvent(appId, user, null,
         ContainerLogsRetentionPolicy.ALL_CONTAINERS, null));
 
     logHandler.handle(new LogHandlerContainerFinishedEvent(container11, 0));
 
-    logHandler.handle(new LogHandlerAppFinishedEvent(appId1));
+    logHandler.handle(new LogHandlerAppFinishedEvent(appId));
 
     Path[] localAppLogDirs = new Path[2];
     localAppLogDirs[0] =
-        new Path(localLogDirs[0].getAbsolutePath(), appId1.toString());
+        new Path(localLogDirs[0].getAbsolutePath(), appId.toString());
     localAppLogDirs[1] =
-        new Path(localLogDirs[1].getAbsolutePath(), appId1.toString());
+        new Path(localLogDirs[1].getAbsolutePath(), appId.toString());
 
     ScheduledThreadPoolExecutor mockSched =
         ((NonAggregatingLogHandlerWithMockExecutor) logHandler).mockSched;
 
     verify(mockSched).schedule(any(Runnable.class), eq(10800l),
         eq(TimeUnit.SECONDS));
+    logHandler.close();
+    for (int i = 0; i < localLogDirs.length; i++) {
+      FileUtils.deleteDirectory(localLogDirs[i]);
+    }
   }
   
   @Test
@@ -202,25 +222,25 @@ public class TestNonAggregatingLogHandler {
     verify(logHandler.mockSched)
         .awaitTermination(eq(10l), eq(TimeUnit.SECONDS));
     verify(logHandler.mockSched).shutdownNow();
+    logHandler.close();
+    aggregatingLogHandler.close();
   }
 
   @Test
-  public void testHandlingApplicationFinishedEvent() {
-    Configuration conf = new Configuration();
-    LocalDirsHandlerService dirsService  = new LocalDirsHandlerService();
+  public void testHandlingApplicationFinishedEvent() throws IOException {
     DeletionService delService = new DeletionService(null);
     NonAggregatingLogHandler aggregatingLogHandler =
         new NonAggregatingLogHandler(new InlineDispatcher(),
             delService,
-            dirsService);
+            dirsHandler);
 
-    dirsService.init(conf);
-    dirsService.start();
+    dirsHandler.init(conf);
+    dirsHandler.start();
     delService.init(conf);
     delService.start();
     aggregatingLogHandler.init(conf);
     aggregatingLogHandler.start();
-    ApplicationId appId = BuilderUtils.newApplicationId(1234, 1);
+    
     // It should NOT throw RejectedExecutionException
     aggregatingLogHandler.handle(new LogHandlerAppFinishedEvent(appId));
     aggregatingLogHandler.stop();
@@ -228,6 +248,7 @@ public class TestNonAggregatingLogHandler {
     // It should NOT throw RejectedExecutionException after stopping
     // handler service.
     aggregatingLogHandler.handle(new LogHandlerAppFinishedEvent(appId));
+    aggregatingLogHandler.close();
   }
 
   private class NonAggregatingLogHandlerWithMockExecutor extends
@@ -255,4 +276,201 @@ public class TestNonAggregatingLogHandler {
     dispatcher.start();
     return dispatcher;
   }
+  
+  /*
+   * Test to ensure that we handle the cleanup of directories that may not have
+   * the application log dirs we're trying to delete or may have other problems.
+   * Test creates 7 log dirs, and fails the directory check for 4 of them and
+   * then checks to ensure we tried to delete only the ones that passed the
+   * check.
+   */
+  @Test
+  public void testFailedDirLogDeletion() throws Exception {
+
+    File[] localLogDirs = getLocalLogDirFiles(this.getClass().getName(), 7);
+    final List<String> localLogDirPaths =
+        new ArrayList<String>(localLogDirs.length);
+    for (int i = 0; i < localLogDirs.length; i++) {
+      localLogDirPaths.add(localLogDirs[i].getAbsolutePath());
+    }
+
+    String localLogDirsString = StringUtils.join(localLogDirPaths, ",");
+
+    conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDirsString);
+    conf.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, false);
+    conf.setLong(YarnConfiguration.NM_LOG_RETAIN_SECONDS, 0l);
+
+    LocalDirsHandlerService mockDirsHandler = mock(LocalDirsHandlerService.class);
+
+    NonAggregatingLogHandler rawLogHandler =
+        new NonAggregatingLogHandler(dispatcher, mockDelService, mockDirsHandler);
+    NonAggregatingLogHandler logHandler = spy(rawLogHandler);
+    AbstractFileSystem spylfs =
+        spy(FileContext.getLocalFSFileContext().getDefaultFileSystem());
+    FileContext lfs = FileContext.getFileContext(spylfs, conf);
+    doReturn(lfs).when(logHandler)
+      .getLocalFileContext(isA(Configuration.class));
+    logHandler.init(conf);
+    logHandler.start();
+    runMockedFailedDirs(logHandler, appId, user, mockDelService,
+      mockDirsHandler, conf, spylfs, lfs, localLogDirs);
+    logHandler.close();
+  }
+  
+  /**
+   * Function to run a log handler with directories failing the getFileStatus
+   * call. The function accepts the log handler, setup the mocks to fail with
+   * specific exceptions and ensures the deletion service has the correct calls.
+   * 
+   * @param logHandler the logHandler implementation to test
+   * 
+   * @param appId the application id that we wish when sending events to the log
+   * handler
+   * 
+   * @param user the user name to use
+   * 
+   * @param mockDelService a mock of the DeletionService which we will verify
+   * the delete calls against
+   * 
+   * @param dirsHandler a spy or mock on the LocalDirsHandler service used to
+   * when creating the logHandler. It needs to be a spy so that we can intercept
+   * the getAllLogDirs() call.
+   * 
+   * @param conf the configuration used
+   * 
+   * @param spylfs a spy on the AbstractFileSystem object used when creating lfs
+   * 
+   * @param lfs the FileContext object to be used to mock the getFileStatus()
+   * calls
+   * 
+   * @param localLogDirs list of the log dirs to run the test against, must have
+   * at least 7 entries
+   */
+  public static void runMockedFailedDirs(LogHandler logHandler,
+      ApplicationId appId, String user, DeletionService mockDelService,
+      LocalDirsHandlerService dirsHandler, Configuration conf,
+      AbstractFileSystem spylfs, FileContext lfs, File[] localLogDirs)
+      throws Exception {
+    Map<ApplicationAccessType, String> appAcls = new HashMap<ApplicationAccessType,
String>();
+    if (localLogDirs.length < 7) {
+      throw new IllegalArgumentException(
+        "Argument localLogDirs must be at least of length 7");
+    }
+    Path[] localAppLogDirPaths = new Path[localLogDirs.length];
+    for (int i = 0; i < localAppLogDirPaths.length; i++) {
+      localAppLogDirPaths[i] =
+          new Path(localLogDirs[i].getAbsolutePath(), appId.toString());
+    }
+    final List<String> localLogDirPaths =
+        new ArrayList<String>(localLogDirs.length);
+    for (int i = 0; i < localLogDirs.length; i++) {
+      localLogDirPaths.add(localLogDirs[i].getAbsolutePath());
+    }
+
+    // setup mocks
+    FsPermission defaultPermission =
+        FsPermission.getDirDefault().applyUMask(lfs.getUMask());
+    final FileStatus fs =
+        new FileStatus(0, true, 1, 0, System.currentTimeMillis(), 0,
+          defaultPermission, "", "",
+          new Path(localLogDirs[0].getAbsolutePath()));
+    doReturn(fs).when(spylfs).getFileStatus(isA(Path.class));
+    doReturn(localLogDirPaths).when(dirsHandler).getLogDirsForCleanup();
+
+    logHandler.handle(new LogHandlerAppStartedEvent(appId, user, null,
+      ContainerLogsRetentionPolicy.ALL_CONTAINERS, appAcls));
+
+    // test case where some dirs have the log dir to delete
+    // mock some dirs throwing various exceptions
+    // verify deletion happens only on the others
+    Mockito.doThrow(new FileNotFoundException()).when(spylfs)
+      .getFileStatus(eq(localAppLogDirPaths[0]));
+    doReturn(fs).when(spylfs).getFileStatus(eq(localAppLogDirPaths[1]));
+    Mockito.doThrow(new AccessControlException()).when(spylfs)
+      .getFileStatus(eq(localAppLogDirPaths[2]));
+    doReturn(fs).when(spylfs).getFileStatus(eq(localAppLogDirPaths[3]));
+    Mockito.doThrow(new IOException()).when(spylfs)
+      .getFileStatus(eq(localAppLogDirPaths[4]));
+    Mockito.doThrow(new UnsupportedFileSystemException("test")).when(spylfs)
+      .getFileStatus(eq(localAppLogDirPaths[5]));
+    doReturn(fs).when(spylfs).getFileStatus(eq(localAppLogDirPaths[6]));
+
+    logHandler.handle(new LogHandlerAppFinishedEvent(appId));
+
+    testDeletionServiceCall(mockDelService, user, 5000, localAppLogDirPaths[1],
+      localAppLogDirPaths[3], localAppLogDirPaths[6]);
+
+    return;
+  }
+
+  static class DeletePathsMatcher extends ArgumentMatcher<Path[]> implements
+      VarargMatcher {
+    
+    // to get rid of serialization warning
+    static final long serialVersionUID = 0;
+
+    private transient Path[] matchPaths;
+
+    DeletePathsMatcher(Path... matchPaths) {
+      this.matchPaths = matchPaths;
+    }
+
+    @Override
+    public boolean matches(Object varargs) {
+      return new EqualsBuilder().append(matchPaths, varargs).isEquals();
+    }
+
+    // function to get rid of FindBugs warning
+    private void readObject(ObjectInputStream os) throws NotSerializableException {
+      throw new NotSerializableException(this.getClass().getName());
+    }
+  }
+
+  /**
+   * Function to verify that the DeletionService object received the right
+   * requests.
+   * 
+   * @param delService the DeletionService mock which we verify against
+   * 
+   * @param user the user name to use when verifying the deletion
+   * 
+   * @param timeout amount in milliseconds to wait before we decide the calls
+   * didn't come through
+   * 
+   * @param matchPaths the paths to match in the delete calls
+   * 
+   * @throws WantedButNotInvoked if the calls could not be verified
+   */
+  static void testDeletionServiceCall(DeletionService delService, String user,
+      long timeout, Path... matchPaths) {
+
+    long verifyStartTime = System.currentTimeMillis();
+    WantedButNotInvoked notInvokedException = null;
+    boolean matched = false;
+    while (!matched && System.currentTimeMillis() < verifyStartTime + timeout)
{
+      try {
+        verify(delService).delete(eq(user), (Path) eq(null),
+          Mockito.argThat(new DeletePathsMatcher(matchPaths)));
+        matched = true;
+      } catch (WantedButNotInvoked e) {
+        notInvokedException = e;
+        try {
+          Thread.sleep(50l);
+        } catch (InterruptedException i) {
+        }
+      }
+    }
+    if (!matched) {
+      throw notInvokedException;
+    }
+    return;
+  }
+
+  public static File[] getLocalLogDirFiles(String name, int number) {
+    File[] dirs = new File[number];
+    for (int i = 0; i < dirs.length; i++) {
+      dirs[i] = new File("target", name + "-localLogDir" + i).getAbsoluteFile();
+    }
+    return dirs;
+  }
 }


Mime
View raw message