簡(jiǎn)介

ContainerManager主要負責NM中管理所有Container生命周期，其主要包含啟動(dòng)Container、恢復Container、停止Container等功能。
主要功能由ContainerManagerImpl類(lèi)實(shí)現，具體代碼可以參考當前類(lèi)。

初始化

初始化主要分為兩部分：

ContainerManagerImpl實(shí)例的構造函數和serviceInit函數。

構造函數

當前函數為構造函數，主要初始化必須要的一些變量等。

• dispatcher ： 事件的中央調度器，主要用于管理單個(gè)Container的生命周期。在當前函數里面會(huì )通過(guò)dispatcher.register()注冊支持的事件。
• containersLauncher： 主要用于啟動(dòng)Container，可以通過(guò)配置項yarn.nodemanager.containers-launcher.class指定。默認為containersLauncher.class

serviceInit函數

主要是服務(wù)啟動(dòng)時(shí)的初始化函數，ContainerManager在NodeManager內部屬于一個(gè)服務(wù)。所以初始化的時(shí)候會(huì )調用這個(gè)函數初始化一些服務(wù)相關(guān)的東西。

在這個(gè)函數里面總結下來(lái)主要做了幾件事：

• 繼續初始化一些事件，主要包含LogHandlerEventType、SharedCacheUploadEventType。
• 初始化AMRMProxyService。
• 從本地的LevelDB恢復Container信息。

恢復當前NodeManager的所有作業(yè)信息

第一步：恢復Application 信息

首先是從LevelDB里面加載Application信息。循環(huán)加載。

RecoveredApplicationsState appsState = stateStore.loadApplicationsState();
 try (RecoveryIterator<ContainerManagerApplicationProto> rasIterator =
          appsState.getIterator()) {
   while (rasIterator.hasNext()) {
     ContainerManagerApplicationProto proto = rasIterator.next();
     LOG.debug("Recovering application with state: {}", proto);
     recoverApplication(proto);
   }
 }

加載Application的時(shí)候會(huì )將Application的上下文信息從LevelDB里面讀出來(lái)，通過(guò)上下文信息等初始化新的ApplicationImpl，并且觸發(fā)ApplicationInitEvent事件。
會(huì )根據當前作業(yè)上下文中實(shí)際的狀態(tài)等信息跳轉到實(shí)際的狀態(tài)。

ApplicationImpl app = new ApplicationImpl(dispatcher, p.getUser(), fc,
    appId, creds, context, p.getAppLogAggregationInitedTime());
context.getApplications().put(appId, app);
metrics.runningApplication();
app.handle(new ApplicationInitEvent(appId, acls, logAggregationContext));

第二步：恢復所有的Container信息

第二步是從LevelDB里面加載Container信息。循環(huán)加載。

try (RecoveryIterator<RecoveredContainerState> rcsIterator =
          stateStore.getContainerStateIterator()) {
   while (rcsIterator.hasNext()) {
     RecoveredContainerState rcs = rcsIterator.next();
     LOG.debug("Recovering container with state: {}", rcs);
     recoverContainer(rcs);
   }
 }

recoverContainer函數用于恢復單個(gè)Container信息。對于已經(jīng)存在的Application對應的Container會(huì )通過(guò)LevelDB里面加載到的信息初始化Container對象，
將其加到所有Container的列表里面并且觸發(fā)ApplicationContainerInitEvent，后續會(huì )根據實(shí)際狀態(tài)信息跳轉到指定狀態(tài)繼續處理。

Container container = new ContainerImpl(getConfig(), dispatcher,
    launchContext, credentials, metrics, token, context, rcs);
context.getContainers().put(token.getContainerID(), container);
containerScheduler.recoverActiveContainer(container, rcs);
app.handle(new ApplicationContainerInitEvent(container));

如果發(fā)現作業(yè)的狀態(tài)為KILL狀態(tài)，則會(huì )為當前Container重新觸發(fā)KILL事件，保證Container已經(jīng)停止。

對于A(yíng)pplication找不見(jiàn)的Container，認為作業(yè)已經(jīng)結束了，直接標記為已經(jīng)完成。

在恢復完成之后會(huì )觸發(fā)事件： ContainerSchedulerEventType.RECOVERY_COMPLETED

此狀態(tài)會(huì )重新拉起所有的Container。

啟動(dòng)Containers

獲取NMToken

在Container啟動(dòng)之前需要獲取NMToken,可以通過(guò)下面命令獲取，一般情況下獲取第一個(gè)NMTokenIdentifier類(lèi)型的Token。

Set<TokenIdentifier> tokenIdentifiers = remoteUgi.getTokenIdentifiers();

開(kāi)始啟動(dòng)Container

啟動(dòng)之前需要做的就是初始化ContainerImpl信息，方便后續啟動(dòng)Container。

Container container =
    new ContainerImpl(getConfig(), this.dispatcher,
        launchContext, credentials, metrics, containerTokenIdentifier,
        context, containerStartTime);

如果是第一次啟動(dòng)(滿(mǎn)足條件：!context.getApplications().containsKey(applicationID))，也就是AM，會(huì )通過(guò)下面命令觸發(fā)作業(yè)的啟動(dòng)：

context.getNMStateStore().storeApplication(applicationID,
    buildAppProto(applicationID, user, credentials, appAcls,
        logAggregationContext, flowContext));
dispatcher.getEventHandler().handle(new ApplicationInitEvent(
    applicationID, appAcls, logAggregationContext));

滿(mǎn)足下面條件則是恢復Application：

containerTokenIdentifier.getContainerType() == ContainerType.APPLICATION_MASTER && context.getApplications().containsKey(applicationID))

啟動(dòng)Container主要是觸發(fā)Container的Init事件：

this.context.getNMStateStore().storeContainer(containerId,
    containerTokenIdentifier.getVersion(), containerStartTime, request);
dispatcher.getEventHandler().handle(
  new ApplicationContainerInitEvent(container));

停止Container

獲取NMToken

在Container停止之前需要獲取NMToken,可以通過(guò)下面命令獲取，一般情況下獲取第一個(gè)NMTokenIdentifier類(lèi)型的Token。和啟動(dòng)時(shí)候的類(lèi)似。

Set<TokenIdentifier> tokenIdentifiers = remoteUgi.getTokenIdentifiers();

停止作業(yè)

停止作業(yè)的時(shí)候會(huì )優(yōu)先通過(guò)context.getContainers().get(containerID)獲取Container信息。

如下場(chǎng)景會(huì )拋出異常：

• 對于查詢(xún)不到的Container，如果不是最新停止的，則會(huì )拋出異常。
• 對于正在恢復的Container，不會(huì )接受停止，會(huì )拋出異常。

停止作業(yè)的核心邏輯如下,核心思想是觸發(fā)KILL事件。具體可以參見(jiàn)Container事件處理

context.getNMStateStore().storeContainerKilled(containerID);
container.sendKillEvent(ContainerExitStatus.KILLED_BY_APPMASTER,
    "Container killed by the ApplicationMaster.");

Container事件處理

Container啟動(dòng)的開(kāi)始是從事件InitContainerTransition。由當前事件通過(guò)狀態(tài)機轉到其他狀態(tài)。