import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public void run() {
  List<String> rootLogDirs =
      NonAggregatingLogHandler.this.dirsHandler.getLogDirs();
  Path[] localAppLogDirs = new Path[rootLogDirs.size()];
  int index = 0;
  for (String rootLogDir : rootLogDirs) {
    localAppLogDirs[index] = new Path(rootLogDir, applicationId.toString());
    index++;
  }
  // Inform the application before the actual delete itself, so that links
  // to logs will no longer be there on NM web-UI. 
  NonAggregatingLogHandler.this.dispatcher.getEventHandler().handle(
      new ApplicationEvent(this.applicationId,
          ApplicationEventType.APPLICATION_LOG_HANDLING_FINISHED));
  NonAggregatingLogHandler.this.delService.delete(user, null,
      localAppLogDirs);
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void run() {
  try {
    doAppLogAggregation();
  } catch (Exception e) {
    // do post clean up of log directories on any exception
    LOG.error("Error occured while aggregating the log for the application "
        + appId, e);
    doAppLogAggregationPostCleanUp();
  } finally {
    if (!this.appAggregationFinished.get() && !this.aborted.get()) {
      LOG.warn("Log aggregation did not complete for application " + appId);
      this.dispatcher.getEventHandler().handle(
          new ApplicationEvent(this.appId,
              ApplicationEventType.APPLICATION_LOG_HANDLING_FAILED));
    }
    this.appAggregationFinished.set(true);
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private void stopApp(ApplicationId appId) {

  // App is complete. Finish up any containers' pending log aggregation and
  // close the application specific logFile.

  AppLogAggregator aggregator = this.appLogAggregators.get(appId);
  if (aggregator == null) {
    LOG.warn("Log aggregation is not initialized for " + appId
        + ", did it fail to start?");
    this.dispatcher.getEventHandler().handle(
        new ApplicationEvent(appId,
            ApplicationEventType.APPLICATION_LOG_HANDLING_FAILED));
    return;
  }
  aggregator.finishLogAggregation();
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public void run() {
  List<Path> localAppLogDirs = new ArrayList<Path>();
  FileContext lfs = getLocalFileContext(getConfig());
  for (String rootLogDir : dirsHandler.getLogDirsForCleanup()) {
    Path logDir = new Path(rootLogDir, applicationId.toString());
    try {
      lfs.getFileStatus(logDir);
      localAppLogDirs.add(logDir);
    } catch (UnsupportedFileSystemException ue) {
      LOG.warn("Unsupported file system used for log dir " + logDir, ue);
      continue;
    } catch (IOException ie) {
      continue;
    }
  }

  // Inform the application before the actual delete itself, so that links
  // to logs will no longer be there on NM web-UI.
  NonAggregatingLogHandler.this.dispatcher.getEventHandler().handle(
    new ApplicationEvent(this.applicationId,
      ApplicationEventType.APPLICATION_LOG_HANDLING_FINISHED));
  if (localAppLogDirs.size() > 0) {
    NonAggregatingLogHandler.this.delService.delete(user, null,
      (Path[]) localAppLogDirs.toArray(new Path[localAppLogDirs.size()]));
  }
  try {
    NonAggregatingLogHandler.this.stateStore.removeLogDeleter(
        this.applicationId);
  } catch (IOException e) {
    LOG.error("Error removing log deletion state", e);
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private void doAppLogAggregation() {
  while (!this.appFinishing.get() && !this.aborted.get()) {
    synchronized(this) {
      try {
        waiting.set(true);
        if (logAggregationInRolling) {
          wait(this.rollingMonitorInterval * 1000);
          if (this.appFinishing.get() || this.aborted.get()) {
            break;
          }
          uploadLogsForContainers(false);
        } else {
          wait(THREAD_SLEEP_TIME);
        }
      } catch (InterruptedException e) {
        LOG.warn("PendingContainers queue is interrupted");
        this.appFinishing.set(true);
      }
    }
  }

  if (this.aborted.get()) {
    return;
  }

  // App is finished, upload the container logs.
  uploadLogsForContainers(true);

  doAppLogAggregationPostCleanUp();

  this.dispatcher.getEventHandler().handle(
      new ApplicationEvent(this.appId,
          ApplicationEventType.APPLICATION_LOG_HANDLING_FINISHED));
  this.appAggregationFinished.set(true);
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@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();
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@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);
  UserGroupInformation.setConfiguration(conf);
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
/**
 * Creates an instance of the service.
 * 
 * @param nmContext
 * @param dispatcher
 */
public AMRMProxyService(Context nmContext, AsyncDispatcher dispatcher) {
  super(AMRMProxyService.class.getName());
  Preconditions.checkArgument(nmContext != null, "nmContext is null");
  Preconditions.checkArgument(dispatcher != null, "dispatcher is null");
  this.nmContext = nmContext;
  this.dispatcher = dispatcher;
  this.applPipelineMap =
      new ConcurrentHashMap<ApplicationId, RequestInterceptorChainWrapper>();

  this.dispatcher.register(ApplicationEventType.class,
      new ApplicationEventHandler());
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
private void finishApplication(ApplicationId appId,
    LogAggregationService logAggregationService) throws Exception {
  dispatcher.await();
  ApplicationEvent expectedInitEvents[] =
      new ApplicationEvent[] { new ApplicationEvent(appId,
          ApplicationEventType.APPLICATION_LOG_HANDLING_INITED) };
  checkEvents(appEventHandler, expectedInitEvents, false, "getType",
      "getApplicationID");
  reset(appEventHandler);

  logAggregationService.handle(new LogHandlerAppFinishedEvent(appId));
  logAggregationService.stop();
  assertEquals(0, logAggregationService.getNumAggregators());
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
private void verifyLogAggFinishEvent(ApplicationId appId) throws Exception {
  dispatcher.await();

  ApplicationEvent[] expectedFinishedEvents =
      new ApplicationEvent[] { new ApplicationEvent(appId,
          ApplicationEventType.APPLICATION_LOG_HANDLING_FINISHED) };
  checkEvents(appEventHandler, expectedFinishedEvents, false, "getType",
          "getApplicationID");
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void handle(LogHandlerEvent event) {
  switch (event.getType()) {
    case APPLICATION_STARTED:
      LogHandlerAppStartedEvent appStartedEvent =
          (LogHandlerAppStartedEvent) event;
      this.appOwners.put(appStartedEvent.getApplicationId(),
          appStartedEvent.getUser());
      this.dispatcher.getEventHandler().handle(
          new ApplicationEvent(appStartedEvent.getApplicationId(),
              ApplicationEventType.APPLICATION_LOG_HANDLING_INITED));
      break;
    case CONTAINER_FINISHED:
      // Ignore
      break;
    case APPLICATION_FINISHED:
      LogHandlerAppFinishedEvent appFinishedEvent =
          (LogHandlerAppFinishedEvent) event;
      // Schedule - so that logs are available on the UI till they're deleted.
      LOG.info("Scheduling Log Deletion for application: "
          + appFinishedEvent.getApplicationId() + ", with delay of "
          + this.deleteDelaySeconds + " seconds");
      LogDeleterRunnable logDeleter =
          new LogDeleterRunnable(appOwners.remove(appFinishedEvent
                .getApplicationId()), appFinishedEvent.getApplicationId());
      try {
        sched.schedule(logDeleter, this.deleteDelaySeconds,
            TimeUnit.SECONDS);
      } catch (RejectedExecutionException e) {
        // Handling this event in local thread before starting threads
        // or after calling sched.shutdownNow().
        logDeleter.run();
      }
      break;
    default:
      ; // Ignore
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public void run() {
  List<Path> localAppLogDirs = new ArrayList<Path>();
  FileContext lfs = getLocalFileContext(getConfig());
  for (String rootLogDir : dirsHandler.getLogDirsForCleanup()) {
    Path logDir = new Path(rootLogDir, applicationId.toString());
    try {
      lfs.getFileStatus(logDir);
      localAppLogDirs.add(logDir);
    } catch (UnsupportedFileSystemException ue) {
      LOG.warn("Unsupported file system used for log dir " + logDir, ue);
      continue;
    } catch (IOException ie) {
      continue;
    }
  }

  // Inform the application before the actual delete itself, so that links
  // to logs will no longer be there on NM web-UI.
  NonAggregatingLogHandler.this.dispatcher.getEventHandler().handle(
    new ApplicationEvent(this.applicationId,
      ApplicationEventType.APPLICATION_LOG_HANDLING_FINISHED));
  if (localAppLogDirs.size() > 0) {
    NonAggregatingLogHandler.this.delService.delete(user, null,
      (Path[]) localAppLogDirs.toArray(new Path[localAppLogDirs.size()]));
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@SuppressWarnings({"unchecked"})
private void handleDestroyApplicationResources(Application application) {
  String userName;
  String appIDStr;
  LocalResourcesTracker appLocalRsrcsTracker =
    appRsrc.remove(ConverterUtils.toString(application.getAppId()));
  if (null == appLocalRsrcsTracker) {
    LOG.warn("Removing uninitialized application " + application);
  }
  // TODO: What to do with appLocalRsrcsTracker?

  // Delete the application directories
  userName = application.getUser();
  appIDStr = application.toString();
  for (String localDir : dirsHandler.getLocalDirs()) {

    // Delete the user-owned app-dir
    Path usersdir = new Path(localDir, ContainerLocalizer.USERCACHE);
    Path userdir = new Path(usersdir, userName);
    Path allAppsdir = new Path(userdir, ContainerLocalizer.APPCACHE);
    Path appDir = new Path(allAppsdir, appIDStr);
    delService.delete(userName, appDir, new Path[] {});

    // Delete the nmPrivate app-dir
    Path sysDir = new Path(localDir, NM_PRIVATE_DIR);
    Path appSysDir = new Path(sysDir, appIDStr);
    delService.delete(null, appSysDir, new Path[] {});
  }

  // TODO: decrement reference counts of all resources associated with this
  // app

  dispatcher.getEventHandler().handle(new ApplicationEvent(
        application.getAppId(),
        ApplicationEventType.APPLICATION_RESOURCES_CLEANEDUP));
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void handle(LogHandlerEvent event) {
  switch (event.getType()) {
    case APPLICATION_STARTED:
      LogHandlerAppStartedEvent appStartedEvent =
          (LogHandlerAppStartedEvent) event;
      this.appOwners.put(appStartedEvent.getApplicationId(),
          appStartedEvent.getUser());
      this.dispatcher.getEventHandler().handle(
          new ApplicationEvent(appStartedEvent.getApplicationId(),
              ApplicationEventType.APPLICATION_LOG_HANDLING_INITED));
      break;
    case CONTAINER_FINISHED:
      // Ignore
      break;
    case APPLICATION_FINISHED:
      LogHandlerAppFinishedEvent appFinishedEvent =
          (LogHandlerAppFinishedEvent) event;
      // Schedule - so that logs are available on the UI till they're deleted.
      LOG.info("Scheduling Log Deletion for application: "
          + appFinishedEvent.getApplicationId() + ", with delay of "
          + this.deleteDelaySeconds + " seconds");
      sched.schedule(
          new LogDeleterRunnable(appOwners.remove(appFinishedEvent
              .getApplicationId()), appFinishedEvent.getApplicationId()),
          this.deleteDelaySeconds, TimeUnit.SECONDS);
      break;
    default:
      ; // Ignore
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@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);
  logAggregationService.init(this.conf);
  logAggregationService.start();

  ApplicationId application1 = BuilderUtils.newApplicationId(1234, 1);
  logAggregationService.handle(new LogHandlerAppStartedEvent(
          application1, this.user, null,
          ContainerLogsRetentionPolicy.ALL_CONTAINERS, this.acls));

  logAggregationService.stop();
  assertEquals(0, logAggregationService.getNumAggregators());
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@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);
  logAggregationService.init(this.conf);
  logAggregationService.start();

  ApplicationId application1 = BuilderUtils.newApplicationId(1234, 1);
  logAggregationService.handle(new LogHandlerAppStartedEvent(
          application1, this.user, null,
          ContainerLogsRetentionPolicy.ALL_CONTAINERS, this.acls));

  logAggregationService.handle(new LogHandlerAppFinishedEvent(application1));
  dispatcher.await();
  int timeToWait = 20 * 1000;
  while (timeToWait > 0 && logAggregationService.getNumAggregators() > 0) {
    Thread.sleep(100);
    timeToWait -= 100;
  }
  Assert.assertEquals("Log aggregator failed to cleanup!", 0,
      logAggregationService.getNumAggregators());
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public void run() {
  List<Path> localAppLogDirs = new ArrayList<Path>();
  FileContext lfs = getLocalFileContext(getConfig());
  for (String rootLogDir : dirsHandler.getLogDirsForCleanup()) {
    Path userLogDir = new Path(rootLogDir, userFolder);
    Path logDir = new Path(userLogDir, applicationId.toString());
    try {
      lfs.getFileStatus(logDir);
      localAppLogDirs.add(logDir);
    } catch (UnsupportedFileSystemException ue) {
      LOG.warn("Unsupported file system used for log dir " + logDir, ue);
      continue;
    } catch (IOException ie) {
      continue;
    }
  }

  // Inform the application before the actual delete itself, so that links
  // to logs will no longer be there on NM web-UI.
  NonAggregatingLogHandler.this.dispatcher.getEventHandler().handle(
    new ApplicationEvent(this.applicationId,
      ApplicationEventType.APPLICATION_LOG_HANDLING_FINISHED));
  if (localAppLogDirs.size() > 0) {
    NonAggregatingLogHandler.this.delService.delete(user, null,
      (Path[]) localAppLogDirs.toArray(new Path[localAppLogDirs.size()]));
  }
  try {
    NonAggregatingLogHandler.this.stateStore.removeLogDeleter(
        this.applicationId);
  } catch (IOException e) {
    LOG.error("Error removing log deletion state", e);
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType; //导入依赖的package包/类
@Before
@SuppressWarnings("unchecked")
public void setup() {
  mockDelService = mock(DeletionService.class);
  conf = new YarnConfiguration();
  dispatcher = createDispatcher(conf);
  appEventHandler = new ApplicationEventHandler();
  dispatcher.register(ApplicationEventType.class, appEventHandler);
  appId = BuilderUtils.newApplicationId(1234, 1);
  appAttemptId = BuilderUtils.newApplicationAttemptId(appId, 1);
  container11 = BuilderUtils.newContainerId(appAttemptId, 1);
  dirsHandler = new LocalDirsHandlerService();
}