LTS原理--Monitor任务监控和Admin管理（五）

女爷i 2022-05-10 06:20 269阅读 0赞

前几篇博客我们已经简单的介绍了LTS的JobClient、JobTracker和TaskTracker相互之间的任务发起、调度和执行的流程。

[LTS原理--JobClient提交任务过程（二）][LTS_--JobClient]

[LTS原理--JobTracker任务接收与分配（三）][LTS_--JobTracker]

[LTS原理--TaskTracker任务处理（四）][LTS_--TaskTracker]

接下来我们了解一下LTS的Monitor和Admin是如何采集和收集数据的，最终所有的数据是记录在数据库的表中的，LTS提供了以下表：

![70][]

1、表lts\_admin\_node\_onoffline\_log

作用：记录所有节点（JobClient、JobTracker、TaskTracker和Monitor节点的上下线情况）

实现机制：

lts-admin模块启动时BackendRegistrySr会去监听zk，监听节点的上下线情况然后将log日志添加到数据库中

/**
         * 记录节点上下线日志
         */
        private void addLog(NotifyEvent event, List<Node> nodes) {
            List<NodeOnOfflineLog> logs = new ArrayList<NodeOnOfflineLog>(nodes.size());
    
            for (Node node : nodes) {
                NodeOnOfflineLog log = new NodeOnOfflineLog();
                log.setLogTime(new Date());
                log.setEvent(event == NotifyEvent.ADD ? "ONLINE" : "OFFLINE");
    
                log.setClusterName(node.getClusterName());
                log.setCreateTime(node.getCreateTime());
                log.setGroup(node.getGroup());
                log.setHostName(node.getHostName());
                log.setIdentity(node.getIdentity());
                log.setIp(node.getIp());
                log.setPort(node.getPort());
                log.setThreads(node.getThreads());
                log.setNodeType(node.getNodeType());
                log.setHttpCmdPort(node.getHttpCmdPort());
    
                logs.add(log);
            }
    
            appContext.getBackendNodeOnOfflineLogAccess().insert(logs);
        }
    
        public void start() throws Exception {
    
            registry = RegistryFactory.getRegistry(appContext);
    
            notifyListener = new NotifyListener() {
                @Override
                public void notify(NotifyEvent event, List<Node> nodes) {
                    if (CollectionUtils.isEmpty(nodes)) {
                        return;
                    }
                    switch (event) {
                        case ADD:
                            appContext.getNodeMemCacheAccess().addNode(nodes);
                            LOGGER.info("ADD NODE " + nodes);
                            break;
                        case REMOVE:
                            appContext.getNodeMemCacheAccess().removeNode(nodes);
                            LOGGER.info("REMOVE NODE " + nodes);
                            break;
                    }
                    // 记录日志
                    addLog(event, nodes);
                }
            };
    
            subscribe();
        }

日志信息：

![70 1][]

2、Monitor表：

lts_admin_job_client_monitor_data、lts_admin_job_tracker_monitor_data、lts_admin_jvm_gc、lts_admin_jvm_memory、

lts_admin_jvm_thread、lts_admin_task_tracker_monitor_data。

以上几个表是Moniotr模块创建的表，在节点（JobClient、JobTracker和TaskTracker）启动时会从注册中心zk中获取Monitor的地址信息，会将自身的JVM相关信息和自身节点执行的情况信息发送到Monitor模块，Monitor会监听数据将数据存放到数据库中。

Monitor模块启动时会对外提供ServerSocket

public void start() {
    
            if (!start.compareAndSet(false, true)) {
                // 如果已经启动了,就不重复启动
                return;
            }
    
            if (thread == null) {
                this.thread = new NamedThreadFactory("HTTP-CMD-ACCEPTOR", true).newThread(
                        new Runnable() {
                            @Override
                            public void run() {
    
                                while (isStarted()) {
                                    try {
                                        Socket socket = serverSocket.accept();
                                        if (socket == null) {
                                            continue;
                                        }
                                        executorService.submit(new HttpCmdExecutor(context, socket));
    
                                    } catch (Throwable t) {
                                        LOGGER.error("Accept error ", t);
                                        try {
                                            Thread.sleep(1000); // 1s
                                        } catch (InterruptedException ignored) {
                                        }
                                    }
                                }
    
                            }
                        }
                );
            }
            // 启动线程
            thread.start();
    
            LOGGER.info("HttpCmdAcceptor start succeed ");
        }

在HttpCmdExecutor类中会解析节点（JobClient、JobTracker和TaskTracker）发送过来的信息，然后调用不同的处理器将相关监控信息入库

@Override
        public void run() {
    
            try {
                // 解析请求
                HttpCmdRequest request = parseRequest();
    
                Assert.notNull(request, "Request Error");
    
                Assert.hasText(request.getCommand(), "Command is blank");
    
                Assert.hasText(request.getNodeIdentity(), "nodeIdentity is blank");
    
                HttpCmdProc httpCmdProc = context.getCmdProcessor(request.getNodeIdentity(), request.getCommand());
    
                Assert.notNull(httpCmdProc, "Can not find the command:[" + request.getCommand() + "]");
    
                sendResponse(HTTP_OK, JSON.toJSONString(httpCmdProc.execute(request)));
    
            } catch (HttpCMDErrorException ignored) {
                // 忽略
            } catch (IllegalArgumentException e) {
                sendError(HTTP_BADREQUEST, JSON.toJSONString(HttpCmdResponse.newResponse(false, e.getMessage())), false);
            } catch (Throwable t) {
                LOGGER.error("Error When Execute Command", t);
                sendError(HTTP_INTERNALERROR, JSON.toJSONString(HttpCmdResponse.newResponse(false, "Error:" + t.getMessage())), false);
            }
        }
    
        private HttpCmdRequest parseRequest() throws Exception {
    
            BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8"));
    
            StringTokenizer st = new StringTokenizer(in.readLine());
            if (!st.hasMoreTokens())
                sendError(HTTP_BADREQUEST, "BAD REQUEST: Syntax error");
    
            String method = st.nextToken();
    
            if (!st.hasMoreTokens())
                sendError(HTTP_BADREQUEST, "BAD REQUEST: Missing URI");
    
            String uri = st.nextToken();
    
            Properties params = new Properties();
            assert uri != null;
            int qmi = uri.indexOf('?');
            if (qmi >= 0) {
                decodeParams(uri.substring(qmi + 1), params);
                uri = uri.substring(0, qmi);
            }
    
            Properties header = new Properties();
            if (st.hasMoreTokens()) {
                String line = in.readLine();
                while (line.trim().length() > 0) {
                    int p = line.indexOf(':');
                    header.put(line.substring(0, p).trim().toLowerCase(), line.substring(p + 1).trim());
                    line = in.readLine();
                }
            }
    
            if (method.equalsIgnoreCase("POST")) {
                long size = 0x7FFFFFFFFFFFFFFFL;
                String contentLength = header.getProperty("Content-Length");
                if (contentLength == null) {
                    contentLength = header.getProperty("content-length");
                }
                if (contentLength != null) {
                    size = Integer.parseInt(contentLength);
                }
                String postLine = "";
                char buf[] = new char[512];
                int read = in.read(buf);
                while (read >= 0 && size > 0 && !postLine.endsWith("\r\n")) {
                    size -= read;
                    postLine += String.valueOf(buf, 0, read);
                    if (size > 0)
                        read = in.read(buf);
                }
                postLine = postLine.trim();
                decodeParams(postLine, params);
            }
            return resolveRequest(uri, params);
        }
    
    
        protected static HttpCmdRequest resolveRequest(String uri, Properties params) {
    
            HttpCmdRequest request = new HttpCmdRequest();
            String[] pathNode = uri.substring(1, uri.length()).split("/");
            String nodeIdentity = pathNode[0];
            String command = pathNode[1];
            ;
            request.setCommand(command);
            request.setNodeIdentity(nodeIdentity);
    
            for (Map.Entry<Object, Object> entry : params.entrySet()) {
                request.addParam(String.valueOf(entry.getKey()), String.valueOf(entry.getValue()));
            }
            return request;
        }

在接口HttpCmdProc的实现类MDataAddHttpCmd对发送过来的监控信息进行处理入库

@Override
        public HttpCmdResponse execute(HttpCmdRequest request) throws Exception {
    
            String mNodeJson = request.getParam(HttpCmdParamNames.M_NODE);
            if (StringUtils.isEmpty(mNodeJson)) {
                return HttpCmdResponse.newResponse(false, "mData is empty");
            }
            MNode mNode = JSON.parse(mNodeJson, new TypeReference<MNode>() {
            }.getType());
    
            HttpCmdResponse response = paramCheck(mNode);
            if (response != null) {
                return response;
            }
    
            String mDataJson = request.getParam(HttpCmdParamNames.M_DATA);
            if (StringUtils.isEmpty(mDataJson)) {
                return HttpCmdResponse.newResponse(false, "mData is empty");
            }
            try {
                assert mNode != null;
                List<MData> mDatas = getMDataList(mNode.getNodeType(), mDataJson);
                appContext.getMDataSrv().addMDatas(mNode, mDatas);
            } catch (Exception e) {
                LOGGER.error("Add Monitor Data error: " + JSON.toJSONString(request), e);
                return HttpCmdResponse.newResponse(false, "Add Monitor Data error: " + e.getMessage());
            }
    
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Add Monitor Data success, mNode=" + mNodeJson + ", mData=" + mDataJson);
            }
    
            return HttpCmdResponse.newResponse(true, "Add Monitor Data success");
        }
    
    
        private List<MData> getMDataList(NodeType nodeType, String mDataJson) {
            List<MData> mDatas = null;
            if (NodeType.TASK_TRACKER == nodeType) {
                mDatas = JSON.parse(mDataJson, new TypeReference<List<TaskTrackerMData>>() {
                }.getType());
            } else if (NodeType.JOB_TRACKER == nodeType) {
                mDatas = JSON.parse(mDataJson, new TypeReference<List<JobTrackerMData>>() {
                }.getType());
            } else if (NodeType.JOB_CLIENT == nodeType) {
                mDatas = JSON.parse(mDataJson, new TypeReference<List<JobClientMData>>() {
                }.getType());
            }
            return mDatas;
        }

在MDataSrv类中会根据节点类型分别进行信息入库

public void addMDatas(MNode mNode, List<MData> mDatas) {
            if (CollectionUtils.isEmpty(mDatas)) {
                return;
            }
    
            switch (mNode.getNodeType()) {
                case JOB_CLIENT:
                    addJobClientMData(mNode, mDatas);
                    break;
                case JOB_TRACKER:
                    addJobTrackerMData(mNode, mDatas);
                    break;
                case TASK_TRACKER:
                    addTaskTrackerMData(mNode, mDatas);
                    break;
                default:
                    throw new LtsRuntimeException("Unsupport nodeType:" + mNode.getNodeType());
            }
        }
    
        private void addJobClientMData(MNode mNode, List<MData> mDatas) {
            List<JobClientMDataPo> pos = new ArrayList<JobClientMDataPo>(mDatas.size());
            for (MData mData : mDatas) {
                JobClientMDataPo po = new JobClientMDataPo();
                jobClientMDataBeanCopier.copyProps((JobClientMData) mData, po);
                po.setNodeType(mNode.getNodeType());
                po.setIdentity(mNode.getIdentity());
                po.setNodeGroup(mNode.getNodeGroup());
                po.setGmtCreated(SystemClock.now());
                pos.add(po);
            }
    
            appContext.getJobClientMAccess().insert(pos);
    
            // 添加jvm监控数据
            addJvmMData(mNode, mDatas);
        }
    
        private void addJobTrackerMData(MNode mNode, List<MData> mDatas) {
    
            List<JobTrackerMDataPo> pos = new ArrayList<JobTrackerMDataPo>(mDatas.size());
            for (MData mData : mDatas) {
                JobTrackerMDataPo po = new JobTrackerMDataPo();
                jobTrackerMDataBeanCopier.copyProps((JobTrackerMData) mData, po);
                po.setNodeType(mNode.getNodeType());
                po.setIdentity(mNode.getIdentity());
                po.setNodeGroup(mNode.getNodeGroup());
                po.setGmtCreated(SystemClock.now());
                pos.add(po);
            }
    
            appContext.getJobTrackerMAccess().insert(pos);
    
            // 添加jvm监控数据
            addJvmMData(mNode, mDatas);
        }
    
        private void addTaskTrackerMData(MNode mNode, List<MData> mDatas) {
    
            List<TaskTrackerMDataPo> pos = new ArrayList<TaskTrackerMDataPo>(mDatas.size());
            for (MData mData : mDatas) {
                TaskTrackerMDataPo po = new TaskTrackerMDataPo();
                taskTrackerMDataBeanCopier.copyProps((TaskTrackerMData) mData, po);
                po.setNodeType(mNode.getNodeType());
                po.setIdentity(mNode.getIdentity());
                po.setNodeGroup(mNode.getNodeGroup());
                po.setGmtCreated(SystemClock.now());
                pos.add(po);
            }
            appContext.getTaskTrackerMAccess().insert(pos);
    
            // 添加jvm监控数据
            addJvmMData(mNode, mDatas);
        }
    
        public void addJvmMData(MNode mNode, List<MData> mDatas) {
    
            int size = mDatas.size();
            List<JVMGCDataPo> jvmGCDataPos = new ArrayList<JVMGCDataPo>(size);
            List<JVMMemoryDataPo> jvmMemoryDataPos = new ArrayList<JVMMemoryDataPo>(size);
            List<JVMThreadDataPo> jvmThreadDataPos = new ArrayList<JVMThreadDataPo>(size);
    
            for (MData mData : mDatas) {
    
                JvmMData JVMMData = mData.getJvmMData();
                Long timestamp = mData.getTimestamp();
                // gc
                JVMGCDataPo jvmgcDataPo = getDataPo(JVMMData.getGcMap(), JVMGCDataPo.class, mNode, timestamp);
                jvmGCDataPos.add(jvmgcDataPo);
                // memory
                JVMMemoryDataPo jvmMemoryDataPo = getDataPo(JVMMData.getMemoryMap(), JVMMemoryDataPo.class, mNode, timestamp);
                jvmMemoryDataPos.add(jvmMemoryDataPo);
                // thread
                JVMThreadDataPo jvmThreadDataPo = getDataPo(JVMMData.getThreadMap(), JVMThreadDataPo.class, mNode, timestamp);
                jvmThreadDataPos.add(jvmThreadDataPo);
            }
    
            appContext.getJvmGCAccess().insert(jvmGCDataPos);
            appContext.getJvmMemoryAccess().insert(jvmMemoryDataPos);
            appContext.getJvmThreadAccess().insert(jvmThreadDataPos);
        }

这样Monitor模块完成了数据的入库操作。

在LTS-Admin模块中直接通过查询数据库数据来展示在页面中。

![70 2][]

[LTS_--JobClient]: https://blog.csdn.net/qq924862077/article/details/82813165
[LTS_--JobTracker]: https://blog.csdn.net/qq924862077/article/details/82813985
[LTS_--TaskTracker]: https://blog.csdn.net/qq924862077/article/details/82825708
[70]: /images/20220510/66f7cff7be3a40f68e2164f8e95681ac.png
[70 1]: /images/20220510/40f10afe5b964eb49e7759e3cabd0bca.png
[70 2]: /images/20220510/1a0d0a15e8124af0b2cc01514a0b0166.png