【Hadoop仿真】如何在仿真NodeManager中实现心跳

原创

zeekling

发布于 2025-04-13 13:51:27

470

背景

Hadoop自带的hadoop-sls只能用于压测调度器，可在实际中影响ResourceManager性能的因素比较多，不能只看调度器。当前项目可构造海量的Fake NM节点，用于模拟线上RM的巨大压力场景，进行优化。首先需要对NM进行仿真。

仿真NodeManager

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

向RM注册NM。
开启NM的RPC服务和HTTP服务。
向RM发送心跳信息，以及处理NM中的Container信息。

注册NM

向RM注册NM，主要是调用ResourceTracker的registerNodeManager函数，主要需要下面信息：

http端口。
当前NM节点的资源信息以及物理机资源信息。
当前NM的节点ID，主要由当前节点的主机名以及rpc端口组成，格式为主机名:rpc端口。
当前节点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服务

根据当前Fake的主机名和rpc端口初始化rpc服务，主要实现ContainerManagementProtocol协议。

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服务，目前只显示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信息：
- 节点id。
- NodeUtilization：当前节点的资源信息
- ContainersUtilization：Container的资源信息。
- ContainersStatuses：所有Container的状态信息。
- 当前节点的状态信息：
  - 是否健康。
  - 健康状态，在健康模式下一般是Healthy，非健康模式下就是当前节点的异常信息。
  - 心跳的时间戳。
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);