【深入淺出 Yarn 架構與實現】4-6 RM 行為探究 - 申請與分配 Container

本小節介紹應用程式的 ApplicationMaster 在 NodeManager 成功啟動並向 ResourceManager 註冊後，向 ResourceManager 請求資源（Container）到獲取到資源的整個過程，以及 ResourceManager 內部涉及的主要工作流程。 ...

本小節介紹應用程式的 ApplicationMaster 在 NodeManager 成功啟動並向 ResourceManager 註冊後，向 ResourceManager 請求資源（Container）到獲取到資源的整個過程，以及 ResourceManager 內部涉及的主要工作流程。

一、整體流程

整個過程可看做以下兩個階段的送代迴圈：

階段1 ApplicationMaster 彙報資源需求並領取已經分配到的資源；
階段2 NodeManager 向 ResourceManager 彙報各個 Container 運行狀態，如果 ResourceManager 發現它上面有空閑的資源，則進行一次資源分配，並將分配的資源保存到對應的應用程式數據結構中，等待下次 ApplicationMaster 發送心跳信息時獲取（即階段1)。

一）AM 彙報心跳

1、ApplicationMaster 通過 RPC 函數 ApplicationMasterProtocol#allocate 向 ResourceManager 彙報資源需求（由於該函數被周期性調用，我們通常也稱之為“心跳”），包括新的資源需求描述、待釋放的 Container 列表、請求加入黑名單的節點列表、請求移除黑名單的節點列表等。

public AllocateResponse allocate(AllocateRequest request) {
	// Send the status update to the appAttempt.
    // 發送 RMAppAttemptEventType.STATUS_UPDATE 事件
	this.rmContext.getDispatcher().getEventHandler().handle(
	    new RMAppAttemptStatusupdateEvent(appAttemptId, request.getProgress()));
    
    // 從 am 心跳 AllocateRequest 中取出新的資源需求描述、待釋放的 Container 列表、黑名單列表
    List<ResourceRequest> ask = request.getAskList();
    List<ContainerId> release = request.getReleaseList();
    ResourceBlacklistRequest blacklistRequest = request.getResourceBlacklistRequest();

	// 接下來會做一些檢查（資源申請量、label、blacklist 等）

	// 將資源申請分割（動態調整 container 資源量）
    // Split Update Resource Requests into increase and decrease.
    // No Exceptions are thrown here. All update errors are aggregated
    // and returned to the AM.
    List<UpdateContainerRequest> increaseResourceReqs = new ArrayList<>();
    List<UpdateContainerRequest> decreaseResourceReqs = new ArrayList<>();
    List<UpdateContainerError> updateContainerErrors =
        RMServerUtils.validateAndSplitUpdateResourceRequests(rmContext,
            request, maximumCapacity, increaseResourceReqs,
            decreaseResourceReqs);

	// 調用 ResourceScheduler#allocate 函數，將該 AM 資源需求彙報給 ResourceScheduler
    // （實際是 Capacity、Fair、Fifo 等實際指定的 Scheduler 處理）
    allocation =
        this.rScheduler.allocate(appAttemptId, ask, release,
            blacklistAdditions, blacklistRemovals,
            increaseResourceReqs, decreaseResourceReqs);
}

2、ResourceManager 中的 ApplicationMasterService#allocate 負責處理來自 AM 的心跳請求，收到該請求後，會發送一個 RMAppAttemptEventType.STATUS_UPDATE 事件，RMAppAttemptImpl 收到該事件後，將更新應用程式執行進度和 AMLivenessMonitor 中記錄的應用程式最近更新時間。
3、調用 ResourceScheduler#allocate 函數，將該 AM 資源需求彙報給 ResourceScheduler，實際是 Capacity、Fair、Fifo 等實際指定的 Scheduler 處理。
以 CapacityScheduler#allocate 實現為例：

// CapacityScheduler#allocate
public Allocation allocate(ApplicationAttemptId applicationAttemptId,
    List<ResourceRequest> ask, List<ContainerId> release,
    List<String> blacklistAdditions, List<String> blacklistRemovals,
    List<UpdateContainerRequest> increaseRequests,
    List<UpdateContainerRequest> decreaseRequests) {

    // Release containers
	// 發送 RMContainerEventType.RELEASED
    releaseContainers(release, application);

    // update increase requests
    LeafQueue updateDemandForQueue =
        updateIncreaseRequests(increaseRequests, application);

    // Decrease containers
    decreaseContainers(decreaseRequests, application);

    // Sanity check for new allocation requests
    // 會將資源請求進行規範化，限制到最小和最大區間內，並且規範到最小增長量上
    SchedulerUtils.normalizeRequests(
        ask, getResourceCalculator(), getClusterResource(),
        getMinimumResourceCapability(), getMaximumResourceCapability());

    // Update application requests
    // 將新的資源需求更新到對應的數據結構中
    if (application.updateResourceRequests(ask)
        && (updateDemandForQueue == null)) {
      updateDemandForQueue = (LeafQueue) application.getQueue();
    }

    // 獲取已經為該應用程式分配的資源
    allocation = application.getAllocation(getResourceCalculator(),
                   clusterResource, getMinimumResourceCapability());
        
    return allocation;
}

4、ResourceScheduler 首先讀取待釋放 Container 列表，向對應的 RMContainerImpl 發送 RMContainerEventType.RELEASED 類型事件，殺死正在運行的 Container；然後將新的資源需求更新到對應的數據結構中，之後獲取已經為該應用程式分配的資源，並返回給 ApplicationMasterService。

二）NM 彙報心跳

1、NodeManager 將當前節點各種信息（container 狀況、節點利用率、健康情況等）封裝到 nodeStatus 中，再將標識節點的信息一起封裝到 request 中，之後通過RPC 函數 ResourceTracker#nodeHeartbeat 向 ResourceManager 彙報這些狀態。

// NodeStatusUpdaterImpl#startStatusUpdater
  protected void startStatusUpdater() {

    statusUpdaterRunnable = new Runnable() {
      @Override
      @SuppressWarnings("unchecked")
      public void run() {
        // ...
        Set<NodeLabel> nodeLabelsForHeartbeat =
                nodeLabelsHandler.getNodeLabelsForHeartbeat();
        NodeStatus nodeStatus = getNodeStatus(lastHeartbeatID);

        NodeHeartbeatRequest request =
            NodeHeartbeatRequest.newInstance(nodeStatus,
                NodeStatusUpdaterImpl.this.context
                    .getContainerTokenSecretManager().getCurrentKey(),
                NodeStatusUpdaterImpl.this.context
                    .getNMTokenSecretManager().getCurrentKey(),
                nodeLabelsForHeartbeat);
          
        // 發送 nm 的心跳
        response = resourceTracker.nodeHeartbeat(request);

2、ResourceManager 中的 ResourceTrackerService 負責處理來自 NodeManager 的請求，一旦收到該請求，會向 RMNodeImpl 發送一個 RMNodeEventType.STATUS_UPDATE 類型事件，而 RMNodelmpl 收到該事件後，將更新各個 Container 的運行狀態，併進一步向 ResoutceScheduler 發送一個 SchedulerEventType.NODE_UPDATE 類型事件。

// ResourceTrackerService#nodeHeartbeat
  public NodeHeartbeatResponse nodeHeartbeat(NodeHeartbeatRequest request)
      throws YarnException, IOException {

    NodeStatus remoteNodeStatus = request.getNodeStatus();
    /**
     * Here is the node heartbeat sequence...
     * 1. Check if it's a valid (i.e. not excluded) node
     * 2. Check if it's a registered node
     * 3. Check if it's a 'fresh' heartbeat i.e. not duplicate heartbeat
     * 4. Send healthStatus to RMNode
     * 5. Update node's labels if distributed Node Labels configuration is enabled
     */
      
    // 前 3 步都是各種檢查，後面才是重點的邏輯
    // Heartbeat response
    NodeHeartbeatResponse nodeHeartBeatResponse =
        YarnServerBuilderUtils.newNodeHeartbeatResponse(
            getNextResponseId(lastNodeHeartbeatResponse.getResponseId()),
            NodeAction.NORMAL, null, null, null, null, nextHeartBeatInterval);
    // 這裡會 set 待釋放的 container、application 列表
    // 思考：為何只有待釋放的列表呢？分配的資源不返回麽？ - 分配的資源是和 AM 進行交互的
    rmNode.setAndUpdateNodeHeartbeatResponse(nodeHeartBeatResponse);

    populateKeys(request, nodeHeartBeatResponse);

    ConcurrentMap<ApplicationId, ByteBuffer> systemCredentials =
        rmContext.getSystemCredentialsForApps();
    if (!systemCredentials.isEmpty()) {
      nodeHeartBeatResponse.setSystemCredentialsForApps(systemCredentials);
    }

    // 4. Send status to RMNode, saving the latest response.
    // 發送 RMNodeEventType.STATUS_UPDATE 事件
    RMNodeStatusEvent nodeStatusEvent =
        new RMNodeStatusEvent(nodeId, remoteNodeStatus);
    if (request.getLogAggregationReportsForApps() != null
        && !request.getLogAggregationReportsForApps().isEmpty()) {
      nodeStatusEvent.setLogAggregationReportsForApps(request
        .getLogAggregationReportsForApps());
    }
    this.rmContext.getDispatcher().getEventHandler().handle(nodeStatusEvent);

3、ResourceScheduler 收到事件後，如果該節點上有可分配的空閑資源，則會將這些資源分配給各個應用程式，而分配後的資源僅是記錄到對應的數據結構中，等待 ApplicationMaster 下次通過心跳機制來領取。（資源分配的具體邏輯，將在後面介紹 Scheduler 的文章中詳細講解）。

三、總結

本篇分析了申請與分配 Container 的流程，主要分為兩個階段。
第一階段由 AM 發起，通過心跳向 RM 發起資源請求。
第二階段由 NM 發起，通過心跳向 RM 彙報資源使用情況。
之後就是，RM 根據 AM 資源請求以及 NM 剩餘資源進行一次資源分配（具體分配邏輯將在後續文章中介紹），並將分配的資源通過下一次 AM 心跳返回給 AM。