背景

Hadoop自帶的hadoop-sls只能用于壓測調度器，可在實(shí)際中影響ResourceManager性能的因素比較多，不能只看調度器。當前項目可構造海量的Fake NM節點(diǎn)，用于模擬線(xiàn)上RM的巨大壓力場(chǎng)景，進(jìn)行優(yōu)化。首先需要對NM進(jìn)行仿真。

仿真NodeManager

仿真NodeManager里面的主要功能如下：

• 向RM注冊NM。
• 開(kāi)啟NM的RPC服務(wù)和HTTP服務(wù)。
• 向RM發(fā)送心跳信息，以及處理NM中的Container信息。

注冊NM

向RM注冊NM，主要是調用ResourceTracker的registerNodeManager函數，主要需要下面信息：

• http端口。
• 當前NM節點(diǎn)的資源信息以及物理機資源信息。
• 當前NM的節點(diǎn)ID，主要由當前節點(diǎn)的主機名以及rpc端口組成，格式為主機名:rpc端口。
• 當前節點(diǎn)NM的版本信息。

核心代碼如下：

private void registerNodeManager() throws YarnException, IOException {
        RegisterNodeManagerRequest request = recordFactory.newRecordInstance(RegisterNodeManagerRequest.class);
        request.setHttpPort(httpPort);
        request.setResource(capability);
        request.setPhysicalResource(capability);
        request.setNodeId(this.nodeId);
        request.setNMVersion(YarnVersionInfo.getVersion());
        LOG.info("begin register NodeManager {} capacity={}, available={}, used={}", nodeId, this.capability, this.available, this.used);
        RegisterNodeManagerResponse response = resourceTracker.registerNodeManager(request);
        nmTokenMasterKey = response.getNMTokenMasterKey();
        LOG.info("Register NodeManager {}  success", nodeId);
    }

初始化RPC和http服務(wù)

根據當前Fake的主機名和rpc端口初始化rpc服務(wù)，主要實(shí)現ContainerManagementProtocol協(xié)議。

private void initRpcServer(YarnConfiguration config, int port, String hostName) {
        YarnRPC rpc = YarnRPC.create(config);
        InetSocketAddress addr = NetUtils.createSocketAddr(hostName + ":" + port);
        NMTokenSecretManagerInNM tokenSecretManager = new NMTokenSecretManagerInNM();
        tokenSecretManager.setMasterKey(nmTokenMasterKey);
        Server server = rpc.getServer(ContainerManagementProtocol.class,
                this, addr, config, tokenSecretManager, 10);
        server.start();
        LOG.info("Init rpc {}:{} success", hostName, port);
    }

根據當前Fake的主機名和http端口初始化http服務(wù)，目前只顯示NM的資源信息。

private void initHttpServer(int port, String hostName) throws IOException {
        InetSocketAddress addr = NetUtils.createSocketAddr(hostName + ":" + port);
        HttpServer httpServer = HttpServer.create(addr, 0);
        httpServer.createContext("/", new NMHttpHandler(this));
        httpServer.setExecutor(Executors.newFixedThreadPool(2));
        httpServer.start();
    }

心跳

NM和RM之間的心跳是NM的核心功能，主要是調用ResourceTracker的nodeHeartbeat函數，主要需要下面參數：

• NodeStatus信息：
  • 節點(diǎn)id。
  • NodeUtilization：當前節點(diǎn)的資源信息
  • ContainersUtilization：Container的資源信息。
  • ContainersStatuses：所有Container的狀態(tài)信息。
  • 當前節點(diǎn)的狀態(tài)信息：
    • 是否健康。
    • 健康狀態(tài)，在健康模式下一般是Healthy，非健康模式下就是當前節點(diǎn)的異常信息。
    • 心跳的時(shí)間戳。
• tokenSequenceNo：上次心跳返回的tokenSequenceNo
• LastKnownNMTokenMasterKey：上次心跳返回的LastKnownNMTokenMasterKey
• LastKnownContainerTokenMasterKey：上次心跳返回的LastKnownContainerTokenMasterKey

核心代碼如下：

NodeStatus nodeStatus = createNodeStatus(nodeId, getContainerStatuses(containers));
nodeStatus.setResponseId(responseID);
NodeHeartbeatRequest request = NodeHeartbeatRequest.newInstance(nodeStatus, nmTokenMasterKey, nmTokenMasterKey,
                    CommonNodeLabelsManager.EMPTY_NODELABEL_SET, null, null);
request.setTokenSequenceNo(tokenSequenceNo);
request.setLastKnownNMTokenMasterKey(nmTokenMasterKey);
request.setLastKnownContainerTokenMasterKey(nmTokenMasterKey);
NodeHeartbeatResponse response = resourceTracker.nodeHeartbeat(request);