From 8998c0f26632c34787a38ebacbc7563b82669449 Mon Sep 17 00:00:00 2001 From: ceilzcx <1758619238@qq.com> Date: Wed, 7 Jun 2023 15:59:21 +0800 Subject: [PATCH 1/6] implement rocketmq metrics data collector --- collector/pom.xml | 7 + .../collect/mq/RocketMQCollectData.java | 57 +++ .../collect/mq/RocketMQSingleCollectImpl.java | 332 ++++++++++++++++++ .../collector/dispatch/DispatchConstants.java | 4 + ...ertzbeat.collector.collect.AbstractCollect | 1 + .../hertzbeat/common/entity/job/Metrics.java | 4 + .../entity/job/protocol/RocketMQProtocol.java | 44 +++ .../main/resources/define/app-rocketmq.yml | 144 ++++++++ 8 files changed, 593 insertions(+) create mode 100644 collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java create mode 100644 collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java create mode 100644 common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/RocketMQProtocol.java create mode 100644 manager/src/main/resources/define/app-rocketmq.yml diff --git a/collector/pom.xml b/collector/pom.xml index 7cb2ac6d33b..5f6384ea97d 100644 --- a/collector/pom.xml +++ b/collector/pom.xml @@ -163,6 +163,13 @@ snmp4j 3.6.7 + + + + org.apache.rocketmq + rocketmq-tools + 4.9.4 + diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java new file mode 100644 index 00000000000..2c661915de4 --- /dev/null +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java @@ -0,0 +1,57 @@ +package org.dromara.hertzbeat.collector.collect.mq; + +import lombok.Data; + +import java.util.List; + +/** + * rocketmq采集数据实体类 + * + * @author ceilzcx + * @since 5/6/2023 + */ +@Data +public class RocketMQCollectData { + + private List clusterBrokerDataList; + + private List consumerInfoList; + + @Data + public static class ClusterBrokerData { + + private Long brokerId; + + private String address; + + private String version; + + private double producerMessageTPS; + + private double consumerMessageTPS; + + private long yesterdayProduceCount; + + private long todayProduceCount; + + private long yesterdayConsumeCount; + + private long todayConsumeCount; + } + + @Data + public static class ConsumerInfo { + + private String consumerGroup; + + private int clientQuantity; + + private String messageModel; + + private String consumeType; + + private double consumeTps; + + private long diffTotal; + } +} diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java new file mode 100644 index 00000000000..efe24cacff6 --- /dev/null +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java @@ -0,0 +1,332 @@ +package org.dromara.hertzbeat.collector.collect.mq; + +import com.alibaba.fastjson.JSONObject; +import com.google.common.collect.Lists; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.acl.common.AclClientRPCHook; +import org.apache.rocketmq.acl.common.SessionCredentials; +import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.admin.ConsumeStats; +import org.apache.rocketmq.common.protocol.body.ClusterInfo; +import org.apache.rocketmq.common.protocol.body.ConsumerConnection; +import org.apache.rocketmq.common.protocol.body.KVTable; +import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper; +import org.apache.rocketmq.common.protocol.route.BrokerData; +import org.apache.rocketmq.common.utils.ThreadUtils; +import org.apache.rocketmq.remoting.RPCHook; +import org.apache.rocketmq.tools.admin.DefaultMQAdminExt; +import org.dromara.hertzbeat.collector.collect.AbstractCollect; +import org.dromara.hertzbeat.collector.dispatch.DispatchConstants; +import org.dromara.hertzbeat.collector.util.JsonPathParser; +import org.dromara.hertzbeat.common.constants.CommonConstants; +import org.dromara.hertzbeat.common.entity.job.Metrics; +import org.dromara.hertzbeat.common.entity.job.protocol.RocketMQProtocol; +import org.dromara.hertzbeat.common.entity.message.CollectRep; +import org.dromara.hertzbeat.common.util.CommonUtil; +import org.jetbrains.annotations.NotNull; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.util.Assert; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +/** + * rocketmq采集实现类 + * + * @author ceilzcx + * @since 5/6/2023 + */ +@Slf4j +public class RocketMQSingleCollectImpl extends AbstractCollect implements DisposableBean { + + private static final Set SYSTEM_GROUP_SET = new HashSet<>(); + + private final ExecutorService executorService; + + static { + // system consumer group + SYSTEM_GROUP_SET.add(MixAll.TOOLS_CONSUMER_GROUP); + SYSTEM_GROUP_SET.add(MixAll.FILTERSRV_CONSUMER_GROUP); + SYSTEM_GROUP_SET.add(MixAll.SELF_TEST_CONSUMER_GROUP); + SYSTEM_GROUP_SET.add(MixAll.ONS_HTTP_PROXY_GROUP); + SYSTEM_GROUP_SET.add(MixAll.CID_ONSAPI_PULL_GROUP); + SYSTEM_GROUP_SET.add(MixAll.CID_ONSAPI_PERMISSION_GROUP); + SYSTEM_GROUP_SET.add(MixAll.CID_ONSAPI_OWNER_GROUP); + SYSTEM_GROUP_SET.add(MixAll.CID_SYS_RMQ_TRANS); + } + + public RocketMQSingleCollectImpl() { + Runtime runtime = Runtime.getRuntime(); + int corePoolSize = Math.max(8, runtime.availableProcessors()); + int maximumPoolSize = Math.max(16, runtime.availableProcessors()); + ThreadFactory threadFactory = new ThreadFactory() { + private final AtomicLong threadIndex = new AtomicLong(0); + + @Override + public Thread newThread(@NotNull Runnable r) { + return new Thread(r, "RocketMQCollectGroup_" + this.threadIndex.incrementAndGet()); + } + }; + this.executorService = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 60L, TimeUnit.SECONDS, + new LinkedBlockingQueue<>(5000), threadFactory, new ThreadPoolExecutor.DiscardOldestPolicy()); + } + + @Override + public void destroy() { + ThreadUtils.shutdownGracefully(this.executorService, 10L, TimeUnit.SECONDS); + } + + @Override + public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) { + try { + preCheck(metrics); + } catch (Exception e) { + builder.setCode(CollectRep.Code.FAIL); + builder.setMsg(e.getMessage()); + return; + } + DefaultMQAdminExt mqAdminExt = null; + try { + mqAdminExt = this.createMQAdminExt(metrics); + mqAdminExt.start(); + + RocketMQCollectData rocketMQCollectData = new RocketMQCollectData(); + this.collectData(mqAdminExt, rocketMQCollectData); + + this.fillBuilder(rocketMQCollectData, builder, metrics.getAliasFields(), metrics.getRocketmq().getParseScript()); + + } catch (Exception e) { + builder.setCode(CollectRep.Code.FAIL); + String message = CommonUtil.getMessageFromThrowable(e); + builder.setMsg(message); + } finally { + if (mqAdminExt != null) { + mqAdminExt.shutdown(); + } + } + } + + @Override + public String supportProtocol() { + return DispatchConstants.PROTOCOL_ROCKETMQ; + } + + /** + * 采集前置条件, 入参判断 + * @param metrics 数据指标 + */ + private void preCheck(Metrics metrics) { + if (metrics == null || metrics.getRocketmq() == null) { + throw new IllegalArgumentException("Mongodb collect must has rocketmq params"); + } + RocketMQProtocol rocketmq = metrics.getRocketmq(); + Assert.hasText(rocketmq.getNamesrvHost(), "Rocketmq Protocol namesrvHost is required."); + Assert.hasText(rocketmq.getNamesrvPort(), "Rocketmq Protocol namesrvPort is required."); + } + + /** + * 创建DefaultMQAdminExt实体类; 这里有个小问题, 是否需要每次都重新创建 + * @param metrics 数据指标 + * @return DefaultMQAdminExt + */ + private DefaultMQAdminExt createMQAdminExt(Metrics metrics) { + RocketMQProtocol rocketMQProtocol = metrics.getRocketmq(); + assert rocketMQProtocol != null; + RPCHook rpcHook = null; + if (StringUtils.isNotBlank(rocketMQProtocol.getAccessKey()) && StringUtils.isNotBlank(rocketMQProtocol.getSecretKey())) { + rpcHook = new AclClientRPCHook(new SessionCredentials(rocketMQProtocol.getAccessKey(), rocketMQProtocol.getSecretKey())); + } + DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook, 5000L); + defaultMQAdminExt.setNamesrvAddr(rocketMQProtocol.getNamesrvHost() + ":" + rocketMQProtocol.getNamesrvPort()); + defaultMQAdminExt.setInstanceName("admin-" + System.currentTimeMillis()); + return defaultMQAdminExt; + } + + /** + * 采集rocketmq数据 + * @param mqAdminExt rocketmq提供的远程调用类 + * @param rocketMQCollectData rocketmq数据采集类 + * @throws Exception 远程调用异常 + */ + private void collectData(DefaultMQAdminExt mqAdminExt, RocketMQCollectData rocketMQCollectData) throws Exception { + this.collectClusterData(mqAdminExt, rocketMQCollectData); + this.collectConsumerData(mqAdminExt, rocketMQCollectData); + } + + /** + * 采集rocketmq的集群数据 + * @param mqAdminExt rocketmq提供的远程调用类 + * @param rocketMQCollectData rocketmq数据采集类 + * @throws Exception 远程调用异常 + */ + private void collectClusterData(DefaultMQAdminExt mqAdminExt, RocketMQCollectData rocketMQCollectData) throws Exception { + try { + List clusterBrokerDataList = new ArrayList<>(); + rocketMQCollectData.setClusterBrokerDataList(clusterBrokerDataList); + + ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo(); + for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) { + + for (Map.Entry entry : brokerData.getBrokerAddrs().entrySet()) { + RocketMQCollectData.ClusterBrokerData clusterBrokerData = new RocketMQCollectData.ClusterBrokerData(); + clusterBrokerDataList.add(clusterBrokerData); + + clusterBrokerData.setBrokerId(entry.getKey()); + clusterBrokerData.setAddress(entry.getValue()); + + KVTable kvTable = mqAdminExt.fetchBrokerRuntimeStats(entry.getValue()); + clusterBrokerData.setVersion(kvTable.getTable().get("brokerVersionDesc")); + + String putTps = kvTable.getTable().get("putTps"); + if (StringUtils.isNotEmpty(putTps)) { + String[] putTpsArr = putTps.split(" "); + clusterBrokerData.setProducerMessageTPS(Double.parseDouble(putTpsArr[0])); + } + + String getTransferredTps = kvTable.getTable().get("getTransferedTps"); + if (StringUtils.isNotEmpty(getTransferredTps)) { + String[] getTransferredTpsArr = getTransferredTps.split(" "); + clusterBrokerData.setConsumerMessageTPS(Double.parseDouble(getTransferredTpsArr[0])); + } + + String msgPutTotalTodayMorning = kvTable.getTable().get("msgPutTotalTodayMorning"); + String msgPutTotalYesterdayMorning = kvTable.getTable().get("msgPutTotalYesterdayMorning"); + if (StringUtils.isNotEmpty(msgPutTotalTodayMorning) && StringUtils.isNotEmpty(msgPutTotalYesterdayMorning)) { + long yesterdayProduceCount = Long.parseLong(msgPutTotalTodayMorning) - Long.parseLong(msgPutTotalYesterdayMorning); + clusterBrokerData.setYesterdayProduceCount(yesterdayProduceCount); + } + + String msgGetTotalTodayMorning = kvTable.getTable().get("msgGetTotalTodayMorning"); + String msgGetTotalYesterdayMorning = kvTable.getTable().get("msgGetTotalYesterdayMorning"); + if (StringUtils.isNotEmpty(msgGetTotalTodayMorning) && StringUtils.isNotEmpty(msgGetTotalYesterdayMorning)) { + long yesterdayConsumerCount = Long.parseLong(msgGetTotalTodayMorning) - Long.parseLong(msgGetTotalYesterdayMorning); + clusterBrokerData.setYesterdayConsumeCount(yesterdayConsumerCount); + } + + String msgPutTotalTodayNow = kvTable.getTable().get("msgPutTotalTodayNow"); + if (StringUtils.isNotEmpty(msgPutTotalTodayNow) && StringUtils.isNotEmpty(msgPutTotalTodayMorning)) { + long todayProduceCount = Long.parseLong(msgPutTotalTodayNow) - Long.parseLong(msgPutTotalTodayMorning); + clusterBrokerData.setTodayProduceCount(todayProduceCount); + } + + String msgGetTotalTodayNow = kvTable.getTable().get("msgGetTotalTodayNow"); + if (StringUtils.isNotEmpty(msgGetTotalTodayNow) && StringUtils.isNotEmpty(msgGetTotalTodayMorning)) { + long todayConsumerCount = Long.parseLong(msgGetTotalTodayNow) - Long.parseLong(msgGetTotalTodayMorning); + clusterBrokerData.setTodayConsumeCount(todayConsumerCount); + } + } + + } + } catch (Exception e) { + log.warn("collect rocketmq cluster data error", e); + throw e; + } + } + + /** + * 采集rocketmq的消费者数据 + * @param mqAdminExt rocketmq提供的远程调用类 + * @param rocketMQCollectData rocketmq数据采集类 + * @throws Exception 远程调用异常 + */ + private void collectConsumerData(DefaultMQAdminExt mqAdminExt, RocketMQCollectData rocketMQCollectData) throws Exception { + Set consumerGroupSet = new HashSet<>(); + try { + // 获取consumerGroup集合 + ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo(); + for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) { + SubscriptionGroupWrapper subscriptionGroupWrapper = mqAdminExt.getAllSubscriptionGroup(brokerData.selectBrokerAddr(), 3000L); + consumerGroupSet.addAll(subscriptionGroupWrapper.getSubscriptionGroupTable().keySet()); + } + + List consumerInfoList = Collections.synchronizedList(Lists.newArrayList()); + rocketMQCollectData.setConsumerInfoList(consumerInfoList); + CountDownLatch countDownLatch = new CountDownLatch(consumerGroupSet.size()); + for (String consumerGroup : consumerGroupSet) { + if (SYSTEM_GROUP_SET.contains(consumerGroup)) { + continue; + } + executorService.submit(() -> { + RocketMQCollectData.ConsumerInfo consumerInfo = new RocketMQCollectData.ConsumerInfo(); + consumerInfoList.add(consumerInfo); + consumerInfo.setConsumerGroup(consumerGroup); + try { + ConsumeStats consumeStats = null; + try { + consumeStats = mqAdminExt.examineConsumeStats(consumerGroup); + } + catch (Exception e) { + log.warn("examineConsumeStats exception to consumerGroup {}, response [{}]", consumerGroup, e.getMessage()); + } + if (consumeStats != null) { + consumerInfo.setConsumeTps(consumeStats.getConsumeTps()); + consumerInfo.setDiffTotal(consumeStats.computeTotalDiff()); + } + + ConsumerConnection consumerConnection = null; + try { + consumerConnection = mqAdminExt.examineConsumerConnectionInfo(consumerGroup); + } + catch (Exception e) { + log.warn("examineConsumeStats exception to consumerGroup {}, response [{}]", consumerGroup, e.getMessage()); + } + if (consumerConnection != null) { + consumerInfo.setClientQuantity(consumerConnection.getConnectionSet().size()); + consumerInfo.setMessageModel(consumerConnection.getMessageModel().getModeCN()); + consumerInfo.setConsumeType(consumerConnection.getConsumeType().getTypeCN()); + } + } catch (Exception e) { + log.warn("examineConsumeStats or examineConsumerConnectionInfo error, {}", consumerGroup, e); + } finally { + countDownLatch.countDown(); + } + }); + } + + if (!countDownLatch.await(10, TimeUnit.SECONDS)) { + log.warn("examineConsumeStats or examineConsumerConnectionInfo timeout"); + } + } catch (Exception e) { + log.warn("collect rocketmq consume data error", e); + throw e; + } + } + + /** + * 采集数据填充到builder + * @param rocketMQCollectData rocketmq数据采集类 + * @param builder metrics data builder + * @param aliasFields 字段别名 + * @param parseScript JSON的base path + */ + private void fillBuilder(RocketMQCollectData rocketMQCollectData, CollectRep.MetricsData.Builder builder, List aliasFields, String parseScript) { + String dataJson = JSONObject.toJSONString(rocketMQCollectData); + List results = JsonPathParser.parseContentWithJsonPath(dataJson, parseScript); + for (int i = 0; i < results.size(); i++) { + CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder(); + for (String aliasField : aliasFields) { + List valueList = JsonPathParser.parseContentWithJsonPath(dataJson, parseScript + aliasField); + if (CollectionUtils.isNotEmpty(valueList) && valueList.size() > i) { + Object value = valueList.get(i); + valueRowBuilder.addColumns(value == null ? CommonConstants.NULL_VALUE : String.valueOf(value)); + } else { + valueRowBuilder.addColumns(CommonConstants.NULL_VALUE); + } + } + builder.addValues(valueRowBuilder.build()); + } + } +} diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/dispatch/DispatchConstants.java b/collector/src/main/java/org/dromara/hertzbeat/collector/dispatch/DispatchConstants.java index 13367129b75..c053d141162 100644 --- a/collector/src/main/java/org/dromara/hertzbeat/collector/dispatch/DispatchConstants.java +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/dispatch/DispatchConstants.java @@ -75,6 +75,10 @@ public interface DispatchConstants { * protocol ssl Certificate - custom */ String PROTOCOL_SSL_CERT = "ssl_cert"; + /** + * protocol rocketmq + */ + String PROTOCOL_ROCKETMQ = "rocketmq"; // Protocol type related - end // 协议类型相关 - end // diff --git a/collector/src/main/resources/META-INF/services/org.dromara.hertzbeat.collector.collect.AbstractCollect b/collector/src/main/resources/META-INF/services/org.dromara.hertzbeat.collector.collect.AbstractCollect index ed4a1cfbecb..f7dc59c9f9b 100644 --- a/collector/src/main/resources/META-INF/services/org.dromara.hertzbeat.collector.collect.AbstractCollect +++ b/collector/src/main/resources/META-INF/services/org.dromara.hertzbeat.collector.collect.AbstractCollect @@ -9,3 +9,4 @@ org.dromara.hertzbeat.collector.collect.snmp.SnmpCollectImpl org.dromara.hertzbeat.collector.collect.ssh.SshCollectImpl org.dromara.hertzbeat.collector.collect.telnet.TelnetCollectImpl org.dromara.hertzbeat.collector.collect.ftp.FtpCollectImpl +org.dromara.hertzbeat.collector.collect.mq.RocketMQSingleCollectImpl \ No newline at end of file diff --git a/common/src/main/java/org/dromara/hertzbeat/common/entity/job/Metrics.java b/common/src/main/java/org/dromara/hertzbeat/common/entity/job/Metrics.java index e59a915a0ed..54632bfe19f 100644 --- a/common/src/main/java/org/dromara/hertzbeat/common/entity/job/Metrics.java +++ b/common/src/main/java/org/dromara/hertzbeat/common/entity/job/Metrics.java @@ -151,6 +151,10 @@ public class Metrics { * 使用公共的ftp协议的监控配置信息 */ private FtpProtocol ftp; + /** + * Monitoring configuration information using the public rocketmq protocol 使用公共的rocketmq协议的监控配置信息 + */ + private RocketMQProtocol rocketmq; /** * collector use - Temporarily store subTask indicator group response data diff --git a/common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/RocketMQProtocol.java b/common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/RocketMQProtocol.java new file mode 100644 index 00000000000..aed8c737211 --- /dev/null +++ b/common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/RocketMQProtocol.java @@ -0,0 +1,44 @@ +package org.dromara.hertzbeat.common.entity.job.protocol; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * rocketmq protocol + * + * @author ceilzcx + * @since 5/6/2023 + */ +@Data +@Builder +@AllArgsConstructor +@NoArgsConstructor +public class RocketMQProtocol { + + /** + * rocketmq namesrv host + */ + private String namesrvHost; + + /** + * rocketmq namesrv port + */ + private String namesrvPort; + + /** + * accessKey + */ + private String accessKey; + + /** + * secretKey + */ + private String secretKey; + + /** + * jsonpath解析脚本 + */ + private String parseScript; +} diff --git a/manager/src/main/resources/define/app-rocketmq.yml b/manager/src/main/resources/define/app-rocketmq.yml new file mode 100644 index 00000000000..668c8571003 --- /dev/null +++ b/manager/src/main/resources/define/app-rocketmq.yml @@ -0,0 +1,144 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +category: mid +app: rocketmq +name: + zh-CN: RocketMQ + en-US: RocketMQ +params: + # field-param field key + # field-字段名称标识符 + - field: host + # name-param field display i18n name + # name-参数字段显示名称 + name: + zh-CN: 注册中心Host + en-US: Namesrv Host + # type-param field type(most mapping the html input type) + # type-字段类型,样式(大部分映射input标签type属性) + type: host + # required-true or false + # 是否是必输项 true-必填 false-可选 + required: true + - field: port + name: + zh-CN: 端口 + en-US: Port + type: number + # when type is number, range is required + # 当type为number时,用range表示范围 + range: '[0,65535]' + required: true + defaultValue: 9876 + - field: accessKey + name: + zh-CN: accessKey + en-US: accessKey + type: text + - field: secretKey + name: + zh-CN: secretKey + en-US: secretKey + type: text +metrics: + - name: cluster + priority: 0 + fields: + - field: BrokerId + type: 1 + - field: Address + type: 1 + instance: true + - field: Version + type: 1 + - field: Producer_Message_TPS + type: 0 + - field: Consumer_Message_TPS + type: 0 + - field: Yesterday_Produce_Count + type: 0 + - field: Yesterday_Consume_Count + type: 0 + - field: Today_Produce_Count + type: 0 + - field: Today_Consume_Count + type: 0 + aliasFields: + - brokerId + - address + - version + - producerMessageTPS + - consumerMessageTPS + - yesterdayProduceCount + - yesterdayConsumeCount + - todayProduceCount + - todayConsumeCount + calculates: + - BrokerId=brokerId + - Address=address + - Version=version + - Producer_Message_TPS=producerMessageTPS + - Consumer_Message_TPS=consumerMessageTPS + - Yesterday_Produce_Count=yesterdayProduceCount + - Yesterday_Consume_Count=yesterdayConsumeCount + - Today_Produce_Count=todayProduceCount + - Today_Consume_Count=todayConsumeCount + # 监控采集使用协议 eg: sql, ssh, http, telnet, wmi, snmp, sdk, rocketmq + protocol: rocketmq + # 当protocol为http协议时具体的采集配置 + rocketmq: + namesrvHost: ^_^host^_^ + namesrvPort: ^_^port^_^ + accessKey: ^_^accessKey^_^ + secretKey: ^_^secretKey^_^ + parseScript: $.clusterBrokerDataList.* + - name: consumer + priority: 1 + fields: + - field: Consumer_group + type: 1 + instance: true + - field: Client_quantity + type: 0 + - field: Message_model + type: 1 + - field: Consume_type + type: 1 + - field: Consume_tps + type: 0 + - field: Delay + type: 0 + aliasFields: + - consumerGroup + - clientQuantity + - messageModel + - consumeType + - consumeTps + - diffTotal + calculates: + - Consumer_group=consumerGroup + - Client_quantity=clientQuantity + - Message_model=messageModel + - Consume_type=consumeType + - Consume_tps=consumeTps + - Delay=diffTotal + # 监控采集使用协议 eg: sql, ssh, http, telnet, wmi, snmp, sdk, rocketmq + protocol: rocketmq + # 当protocol为http协议时具体的采集配置 + rocketmq: + namesrvHost: ^_^host^_^ + namesrvPort: ^_^port^_^ + parseScript: $.consumerInfoList.* \ No newline at end of file From 89f4f0627bec9de0487d189606d86ce91b767c23 Mon Sep 17 00:00:00 2001 From: ceilzcx <1758619238@qq.com> Date: Thu, 15 Jun 2023 13:57:00 +0800 Subject: [PATCH 2/6] add collect rocketmq topic data --- .../collect/mq/RocketMQCollectData.java | 17 +++++++++++ .../collect/mq/RocketMQSingleCollectImpl.java | 28 +++++++++++++++++++ 2 files changed, 45 insertions(+) diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java index 2c661915de4..c711d373aeb 100644 --- a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java @@ -3,6 +3,7 @@ import lombok.Data; import java.util.List; +import java.util.Map; /** * rocketmq采集数据实体类 @@ -17,6 +18,8 @@ public class RocketMQCollectData { private List consumerInfoList; + private List>> topicInfoList; + @Data public static class ClusterBrokerData { @@ -54,4 +57,18 @@ public static class ConsumerInfo { private long diffTotal; } + + @Data + public static class TopicQueueInfo { + + private String brokerName; + + private int queueId; + + private long minOffset; + + private long maxOffset; + + private long lastUpdateTimestamp; + } } diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java index efe24cacff6..c1f2c4e2f7b 100644 --- a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java @@ -13,6 +13,7 @@ import org.apache.rocketmq.common.protocol.body.ConsumerConnection; import org.apache.rocketmq.common.protocol.body.KVTable; import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper; +import org.apache.rocketmq.common.protocol.body.TopicList; import org.apache.rocketmq.common.protocol.route.BrokerData; import org.apache.rocketmq.common.utils.ThreadUtils; import org.apache.rocketmq.remoting.RPCHook; @@ -31,6 +32,7 @@ import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -42,6 +44,7 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; /** * rocketmq采集实现类 @@ -164,6 +167,7 @@ private DefaultMQAdminExt createMQAdminExt(Metrics metrics) { private void collectData(DefaultMQAdminExt mqAdminExt, RocketMQCollectData rocketMQCollectData) throws Exception { this.collectClusterData(mqAdminExt, rocketMQCollectData); this.collectConsumerData(mqAdminExt, rocketMQCollectData); + this.collectTopicData(mqAdminExt, rocketMQCollectData); } /** @@ -305,6 +309,30 @@ private void collectConsumerData(DefaultMQAdminExt mqAdminExt, RocketMQCollectDa } } + private void collectTopicData(DefaultMQAdminExt defaultMQAdminExt, RocketMQCollectData rocketMQCollectData) throws Exception { + try { + TopicList topicList = defaultMQAdminExt.fetchAllTopicList(); + Set topics = topicList.getTopicList() + .stream() + .filter(topic -> !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX))) + .collect(Collectors.toSet()); + List>> topicInfoList = new ArrayList<>(); + for (String topic : topics) { + Map> topicQueueInfoTable = new HashMap<>(); + List topicQueueInfoList = new ArrayList<>(); + + // todo 查询topic的queue信息需要for循环调用 mqAdminExt.examineTopicStats(), topic数量很大的情况, 调用极其频繁 + + topicQueueInfoTable.put(topic, topicQueueInfoList); + topicInfoList.add(topicQueueInfoTable); + rocketMQCollectData.setTopicInfoList(topicInfoList); + } + } catch (Exception e) { + log.warn("collect rocketmq topic data error", e); + throw e; + } + } + /** * 采集数据填充到builder * @param rocketMQCollectData rocketmq数据采集类 From 5c95e29051ebf1dce693574321919dd76556f14a Mon Sep 17 00:00:00 2001 From: ceilzcx <1758619238@qq.com> Date: Wed, 28 Jun 2023 10:56:42 +0800 Subject: [PATCH 3/6] check style, update rocketMQ -> rocketmq --- .../collect/mq/RocketMQSingleCollectImpl.java | 70 ++++++++++--------- ...lectData.java => RocketmqCollectData.java} | 2 +- .../hertzbeat/common/entity/job/Metrics.java | 2 +- ...tMQProtocol.java => RocketmqProtocol.java} | 2 +- 4 files changed, 41 insertions(+), 35 deletions(-) rename collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/{RocketMQCollectData.java => RocketmqCollectData.java} (97%) rename common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/{RocketMQProtocol.java => RocketmqProtocol.java} (95%) diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java index c1f2c4e2f7b..1ccd00054ff 100644 --- a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java @@ -23,7 +23,7 @@ import org.dromara.hertzbeat.collector.util.JsonPathParser; import org.dromara.hertzbeat.common.constants.CommonConstants; import org.dromara.hertzbeat.common.entity.job.Metrics; -import org.dromara.hertzbeat.common.entity.job.protocol.RocketMQProtocol; +import org.dromara.hertzbeat.common.entity.job.protocol.RocketmqProtocol; import org.dromara.hertzbeat.common.entity.message.CollectRep; import org.dromara.hertzbeat.common.util.CommonUtil; import org.jetbrains.annotations.NotNull; @@ -106,7 +106,7 @@ public void collect(CollectRep.MetricsData.Builder builder, long appId, String a mqAdminExt = this.createMQAdminExt(metrics); mqAdminExt.start(); - RocketMQCollectData rocketMQCollectData = new RocketMQCollectData(); + RocketmqCollectData rocketMQCollectData = new RocketmqCollectData(); this.collectData(mqAdminExt, rocketMQCollectData); this.fillBuilder(rocketMQCollectData, builder, metrics.getAliasFields(), metrics.getRocketmq().getParseScript()); @@ -135,7 +135,7 @@ private void preCheck(Metrics metrics) { if (metrics == null || metrics.getRocketmq() == null) { throw new IllegalArgumentException("Mongodb collect must has rocketmq params"); } - RocketMQProtocol rocketmq = metrics.getRocketmq(); + RocketmqProtocol rocketmq = metrics.getRocketmq(); Assert.hasText(rocketmq.getNamesrvHost(), "Rocketmq Protocol namesrvHost is required."); Assert.hasText(rocketmq.getNamesrvPort(), "Rocketmq Protocol namesrvPort is required."); } @@ -146,14 +146,14 @@ private void preCheck(Metrics metrics) { * @return DefaultMQAdminExt */ private DefaultMQAdminExt createMQAdminExt(Metrics metrics) { - RocketMQProtocol rocketMQProtocol = metrics.getRocketmq(); - assert rocketMQProtocol != null; + RocketmqProtocol rocketmqProtocol = metrics.getRocketmq(); + assert rocketmqProtocol != null; RPCHook rpcHook = null; - if (StringUtils.isNotBlank(rocketMQProtocol.getAccessKey()) && StringUtils.isNotBlank(rocketMQProtocol.getSecretKey())) { - rpcHook = new AclClientRPCHook(new SessionCredentials(rocketMQProtocol.getAccessKey(), rocketMQProtocol.getSecretKey())); + if (StringUtils.isNotBlank(rocketmqProtocol.getAccessKey()) && StringUtils.isNotBlank(rocketmqProtocol.getSecretKey())) { + rpcHook = new AclClientRPCHook(new SessionCredentials(rocketmqProtocol.getAccessKey(), rocketmqProtocol.getSecretKey())); } DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook, 5000L); - defaultMQAdminExt.setNamesrvAddr(rocketMQProtocol.getNamesrvHost() + ":" + rocketMQProtocol.getNamesrvPort()); + defaultMQAdminExt.setNamesrvAddr(rocketmqProtocol.getNamesrvHost() + ":" + rocketmqProtocol.getNamesrvPort()); defaultMQAdminExt.setInstanceName("admin-" + System.currentTimeMillis()); return defaultMQAdminExt; } @@ -161,31 +161,31 @@ private DefaultMQAdminExt createMQAdminExt(Metrics metrics) { /** * 采集rocketmq数据 * @param mqAdminExt rocketmq提供的远程调用类 - * @param rocketMQCollectData rocketmq数据采集类 + * @param rocketmqCollectData rocketmq数据采集类 * @throws Exception 远程调用异常 */ - private void collectData(DefaultMQAdminExt mqAdminExt, RocketMQCollectData rocketMQCollectData) throws Exception { - this.collectClusterData(mqAdminExt, rocketMQCollectData); - this.collectConsumerData(mqAdminExt, rocketMQCollectData); - this.collectTopicData(mqAdminExt, rocketMQCollectData); + private void collectData(DefaultMQAdminExt mqAdminExt, RocketmqCollectData rocketmqCollectData) throws Exception { + this.collectClusterData(mqAdminExt, rocketmqCollectData); + this.collectConsumerData(mqAdminExt, rocketmqCollectData); + this.collectTopicData(mqAdminExt, rocketmqCollectData); } /** * 采集rocketmq的集群数据 * @param mqAdminExt rocketmq提供的远程调用类 - * @param rocketMQCollectData rocketmq数据采集类 + * @param rocketmqCollectData rocketmq数据采集类 * @throws Exception 远程调用异常 */ - private void collectClusterData(DefaultMQAdminExt mqAdminExt, RocketMQCollectData rocketMQCollectData) throws Exception { + private void collectClusterData(DefaultMQAdminExt mqAdminExt, RocketmqCollectData rocketmqCollectData) throws Exception { try { - List clusterBrokerDataList = new ArrayList<>(); - rocketMQCollectData.setClusterBrokerDataList(clusterBrokerDataList); + List clusterBrokerDataList = new ArrayList<>(); + rocketmqCollectData.setClusterBrokerDataList(clusterBrokerDataList); ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo(); for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) { for (Map.Entry entry : brokerData.getBrokerAddrs().entrySet()) { - RocketMQCollectData.ClusterBrokerData clusterBrokerData = new RocketMQCollectData.ClusterBrokerData(); + RocketmqCollectData.ClusterBrokerData clusterBrokerData = new RocketmqCollectData.ClusterBrokerData(); clusterBrokerDataList.add(clusterBrokerData); clusterBrokerData.setBrokerId(entry.getKey()); @@ -243,10 +243,10 @@ private void collectClusterData(DefaultMQAdminExt mqAdminExt, RocketMQCollectDat /** * 采集rocketmq的消费者数据 * @param mqAdminExt rocketmq提供的远程调用类 - * @param rocketMQCollectData rocketmq数据采集类 + * @param rocketmqCollectData rocketmq数据采集类 * @throws Exception 远程调用异常 */ - private void collectConsumerData(DefaultMQAdminExt mqAdminExt, RocketMQCollectData rocketMQCollectData) throws Exception { + private void collectConsumerData(DefaultMQAdminExt mqAdminExt, RocketmqCollectData rocketmqCollectData) throws Exception { Set consumerGroupSet = new HashSet<>(); try { // 获取consumerGroup集合 @@ -256,15 +256,15 @@ private void collectConsumerData(DefaultMQAdminExt mqAdminExt, RocketMQCollectDa consumerGroupSet.addAll(subscriptionGroupWrapper.getSubscriptionGroupTable().keySet()); } - List consumerInfoList = Collections.synchronizedList(Lists.newArrayList()); - rocketMQCollectData.setConsumerInfoList(consumerInfoList); + List consumerInfoList = Collections.synchronizedList(Lists.newArrayList()); + rocketmqCollectData.setConsumerInfoList(consumerInfoList); CountDownLatch countDownLatch = new CountDownLatch(consumerGroupSet.size()); for (String consumerGroup : consumerGroupSet) { if (SYSTEM_GROUP_SET.contains(consumerGroup)) { continue; } executorService.submit(() -> { - RocketMQCollectData.ConsumerInfo consumerInfo = new RocketMQCollectData.ConsumerInfo(); + RocketmqCollectData.ConsumerInfo consumerInfo = new RocketmqCollectData.ConsumerInfo(); consumerInfoList.add(consumerInfo); consumerInfo.setConsumerGroup(consumerGroup); try { @@ -309,23 +309,29 @@ private void collectConsumerData(DefaultMQAdminExt mqAdminExt, RocketMQCollectDa } } - private void collectTopicData(DefaultMQAdminExt defaultMQAdminExt, RocketMQCollectData rocketMQCollectData) throws Exception { + /** + * + * @param defaultMQAdminExt rocketmq提供的远程调用类 + * @param rocketmqCollectData rocketmq数据采集类 + * @throws Exception 远程调用异常 + */ + private void collectTopicData(DefaultMQAdminExt defaultMQAdminExt, RocketmqCollectData rocketmqCollectData) throws Exception { try { TopicList topicList = defaultMQAdminExt.fetchAllTopicList(); Set topics = topicList.getTopicList() .stream() .filter(topic -> !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX))) .collect(Collectors.toSet()); - List>> topicInfoList = new ArrayList<>(); + List>> topicInfoList = new ArrayList<>(); for (String topic : topics) { - Map> topicQueueInfoTable = new HashMap<>(); - List topicQueueInfoList = new ArrayList<>(); + Map> topicQueueInfoTable = new HashMap<>(); + List topicQueueInfoList = new ArrayList<>(); - // todo 查询topic的queue信息需要for循环调用 mqAdminExt.examineTopicStats(), topic数量很大的情况, 调用极其频繁 + // todo 查询topic的queue信息需要for循环调用 mqAdminExt.examineTopicStats(), topic数量很大的情况, 调用次数也会很多 topicQueueInfoTable.put(topic, topicQueueInfoList); topicInfoList.add(topicQueueInfoTable); - rocketMQCollectData.setTopicInfoList(topicInfoList); + rocketmqCollectData.setTopicInfoList(topicInfoList); } } catch (Exception e) { log.warn("collect rocketmq topic data error", e); @@ -335,13 +341,13 @@ private void collectTopicData(DefaultMQAdminExt defaultMQAdminExt, RocketMQColle /** * 采集数据填充到builder - * @param rocketMQCollectData rocketmq数据采集类 + * @param rocketmqCollectData rocketmq数据采集类 * @param builder metrics data builder * @param aliasFields 字段别名 * @param parseScript JSON的base path */ - private void fillBuilder(RocketMQCollectData rocketMQCollectData, CollectRep.MetricsData.Builder builder, List aliasFields, String parseScript) { - String dataJson = JSONObject.toJSONString(rocketMQCollectData); + private void fillBuilder(RocketmqCollectData rocketmqCollectData, CollectRep.MetricsData.Builder builder, List aliasFields, String parseScript) { + String dataJson = JSONObject.toJSONString(rocketmqCollectData); List results = JsonPathParser.parseContentWithJsonPath(dataJson, parseScript); for (int i = 0; i < results.size(); i++) { CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder(); diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java similarity index 97% rename from collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java rename to collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java index c711d373aeb..094545c66a2 100644 --- a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQCollectData.java +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java @@ -12,7 +12,7 @@ * @since 5/6/2023 */ @Data -public class RocketMQCollectData { +public class RocketmqCollectData { private List clusterBrokerDataList; diff --git a/common/src/main/java/org/dromara/hertzbeat/common/entity/job/Metrics.java b/common/src/main/java/org/dromara/hertzbeat/common/entity/job/Metrics.java index 54632bfe19f..449d085ec09 100644 --- a/common/src/main/java/org/dromara/hertzbeat/common/entity/job/Metrics.java +++ b/common/src/main/java/org/dromara/hertzbeat/common/entity/job/Metrics.java @@ -154,7 +154,7 @@ public class Metrics { /** * Monitoring configuration information using the public rocketmq protocol 使用公共的rocketmq协议的监控配置信息 */ - private RocketMQProtocol rocketmq; + private RocketmqProtocol rocketmq; /** * collector use - Temporarily store subTask indicator group response data diff --git a/common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/RocketMQProtocol.java b/common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/RocketmqProtocol.java similarity index 95% rename from common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/RocketMQProtocol.java rename to common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/RocketmqProtocol.java index aed8c737211..c8d230e2d60 100644 --- a/common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/RocketMQProtocol.java +++ b/common/src/main/java/org/dromara/hertzbeat/common/entity/job/protocol/RocketmqProtocol.java @@ -15,7 +15,7 @@ @Builder @AllArgsConstructor @NoArgsConstructor -public class RocketMQProtocol { +public class RocketmqProtocol { /** * rocketmq namesrv host From fb001ba93858fd3a75b28d2bfe1f1ede536cd944 Mon Sep 17 00:00:00 2001 From: ceilzcx <1758619238@qq.com> Date: Wed, 28 Jun 2023 11:07:25 +0800 Subject: [PATCH 4/6] check style, update TPS -> Tps --- .../collector/collect/mq/RocketMQSingleCollectImpl.java | 4 ++-- .../collector/collect/mq/RocketmqCollectData.java | 4 ++-- manager/src/main/resources/define/app-rocketmq.yml | 8 ++++---- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java index 1ccd00054ff..d30eafe7acf 100644 --- a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java @@ -197,13 +197,13 @@ private void collectClusterData(DefaultMQAdminExt mqAdminExt, RocketmqCollectDat String putTps = kvTable.getTable().get("putTps"); if (StringUtils.isNotEmpty(putTps)) { String[] putTpsArr = putTps.split(" "); - clusterBrokerData.setProducerMessageTPS(Double.parseDouble(putTpsArr[0])); + clusterBrokerData.setProducerMessageTps(Double.parseDouble(putTpsArr[0])); } String getTransferredTps = kvTable.getTable().get("getTransferedTps"); if (StringUtils.isNotEmpty(getTransferredTps)) { String[] getTransferredTpsArr = getTransferredTps.split(" "); - clusterBrokerData.setConsumerMessageTPS(Double.parseDouble(getTransferredTpsArr[0])); + clusterBrokerData.setConsumerMessageTps(Double.parseDouble(getTransferredTpsArr[0])); } String msgPutTotalTodayMorning = kvTable.getTable().get("msgPutTotalTodayMorning"); diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java index 094545c66a2..c7c1be49076 100644 --- a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java @@ -29,9 +29,9 @@ public static class ClusterBrokerData { private String version; - private double producerMessageTPS; + private double producerMessageTps; - private double consumerMessageTPS; + private double consumerMessageTps; private long yesterdayProduceCount; diff --git a/manager/src/main/resources/define/app-rocketmq.yml b/manager/src/main/resources/define/app-rocketmq.yml index 668c8571003..4b3101af94e 100644 --- a/manager/src/main/resources/define/app-rocketmq.yml +++ b/manager/src/main/resources/define/app-rocketmq.yml @@ -80,8 +80,8 @@ metrics: - brokerId - address - version - - producerMessageTPS - - consumerMessageTPS + - producerMessageTps + - consumerMessageTps - yesterdayProduceCount - yesterdayConsumeCount - todayProduceCount @@ -90,8 +90,8 @@ metrics: - BrokerId=brokerId - Address=address - Version=version - - Producer_Message_TPS=producerMessageTPS - - Consumer_Message_TPS=consumerMessageTPS + - Producer_Message_TPS=producerMessageTps + - Consumer_Message_TPS=consumerMessageTps - Yesterday_Produce_Count=yesterdayProduceCount - Yesterday_Consume_Count=yesterdayConsumeCount - Today_Produce_Count=todayProduceCount From b628482bea83595050846c4424981104bbd8a156 Mon Sep 17 00:00:00 2001 From: ceilzcx <1758619238@qq.com> Date: Wed, 28 Jun 2023 14:29:37 +0800 Subject: [PATCH 5/6] add comment, update MQ to mq --- .../collect/mq/RocketmqCollectData.java | 72 ++++++++++++++++++- ...pl.java => RocketmqSingleCollectImpl.java} | 32 +++++---- ...ertzbeat.collector.collect.AbstractCollect | 2 +- 3 files changed, 89 insertions(+), 17 deletions(-) rename collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/{RocketMQSingleCollectImpl.java => RocketmqSingleCollectImpl.java} (94%) diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java index c7c1be49076..5bffcfff9c2 100644 --- a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqCollectData.java @@ -14,61 +14,131 @@ @Data public class RocketmqCollectData { + /** + * cluster broker info + */ private List clusterBrokerDataList; + /** + * consumer info + */ private List consumerInfoList; - private List>> topicInfoList; + /** + * topic info + * Map[key: TopicName, value: Topic Queue info List] + */ + private List>> topicInfoList; @Data public static class ClusterBrokerData { + /** + * broker id + */ private Long brokerId; + /** + * broker address + */ private String address; + /** + * mq version + */ private String version; + /** + * producer send message tps + */ private double producerMessageTps; + /** + * consumer receive message tps + */ private double consumerMessageTps; + /** + * yesterday producer send message count + */ private long yesterdayProduceCount; + /** + * today producer send message count + */ private long todayProduceCount; + /** + * yesterday consumer receive message count + */ private long yesterdayConsumeCount; + /** + * today consumer receive message count + */ private long todayConsumeCount; } @Data public static class ConsumerInfo { + /** + * consumer group + */ private String consumerGroup; + /** + * client num + */ private int clientQuantity; + /** + * message model + */ private String messageModel; + /** + * consume type + */ private String consumeType; + /** + * consume tps + */ private double consumeTps; + /** + * message delay + */ private long diffTotal; } @Data public static class TopicQueueInfo { + /** + * broker name + */ private String brokerName; + /** + * queue id + */ private int queueId; + /** + * message queue min offset + */ private long minOffset; + /** + * message queue max offset + */ private long maxOffset; + /** + * last update time(ms) + */ private long lastUpdateTimestamp; } } diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqSingleCollectImpl.java similarity index 94% rename from collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java rename to collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqSingleCollectImpl.java index d30eafe7acf..115d67a225f 100644 --- a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketMQSingleCollectImpl.java +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqSingleCollectImpl.java @@ -53,7 +53,9 @@ * @since 5/6/2023 */ @Slf4j -public class RocketMQSingleCollectImpl extends AbstractCollect implements DisposableBean { +public class RocketmqSingleCollectImpl extends AbstractCollect implements DisposableBean { + + private static final int WAIT_TIMEOUT = 10; private static final Set SYSTEM_GROUP_SET = new HashSet<>(); @@ -71,7 +73,7 @@ public class RocketMQSingleCollectImpl extends AbstractCollect implements Dispos SYSTEM_GROUP_SET.add(MixAll.CID_SYS_RMQ_TRANS); } - public RocketMQSingleCollectImpl() { + public RocketmqSingleCollectImpl() { Runtime runtime = Runtime.getRuntime(); int corePoolSize = Math.max(8, runtime.availableProcessors()); int maximumPoolSize = Math.max(16, runtime.availableProcessors()); @@ -103,13 +105,13 @@ public void collect(CollectRep.MetricsData.Builder builder, long appId, String a } DefaultMQAdminExt mqAdminExt = null; try { - mqAdminExt = this.createMQAdminExt(metrics); + mqAdminExt = this.createMqAdminExt(metrics); mqAdminExt.start(); - RocketmqCollectData rocketMQCollectData = new RocketmqCollectData(); - this.collectData(mqAdminExt, rocketMQCollectData); + RocketmqCollectData rocketmqCollectData = new RocketmqCollectData(); + this.collectData(mqAdminExt, rocketmqCollectData); - this.fillBuilder(rocketMQCollectData, builder, metrics.getAliasFields(), metrics.getRocketmq().getParseScript()); + this.fillBuilder(rocketmqCollectData, builder, metrics.getAliasFields(), metrics.getRocketmq().getParseScript()); } catch (Exception e) { builder.setCode(CollectRep.Code.FAIL); @@ -145,17 +147,17 @@ private void preCheck(Metrics metrics) { * @param metrics 数据指标 * @return DefaultMQAdminExt */ - private DefaultMQAdminExt createMQAdminExt(Metrics metrics) { + private DefaultMQAdminExt createMqAdminExt(Metrics metrics) { RocketmqProtocol rocketmqProtocol = metrics.getRocketmq(); assert rocketmqProtocol != null; RPCHook rpcHook = null; if (StringUtils.isNotBlank(rocketmqProtocol.getAccessKey()) && StringUtils.isNotBlank(rocketmqProtocol.getSecretKey())) { rpcHook = new AclClientRPCHook(new SessionCredentials(rocketmqProtocol.getAccessKey(), rocketmqProtocol.getSecretKey())); } - DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook, 5000L); - defaultMQAdminExt.setNamesrvAddr(rocketmqProtocol.getNamesrvHost() + ":" + rocketmqProtocol.getNamesrvPort()); - defaultMQAdminExt.setInstanceName("admin-" + System.currentTimeMillis()); - return defaultMQAdminExt; + DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt(rpcHook, 5000L); + mqAdminExt.setNamesrvAddr(rocketmqProtocol.getNamesrvHost() + ":" + rocketmqProtocol.getNamesrvPort()); + mqAdminExt.setInstanceName("admin-" + System.currentTimeMillis()); + return mqAdminExt; } /** @@ -300,7 +302,7 @@ private void collectConsumerData(DefaultMQAdminExt mqAdminExt, RocketmqCollectDa }); } - if (!countDownLatch.await(10, TimeUnit.SECONDS)) { + if (!countDownLatch.await(WAIT_TIMEOUT, TimeUnit.SECONDS)) { log.warn("examineConsumeStats or examineConsumerConnectionInfo timeout"); } } catch (Exception e) { @@ -311,13 +313,13 @@ private void collectConsumerData(DefaultMQAdminExt mqAdminExt, RocketmqCollectDa /** * - * @param defaultMQAdminExt rocketmq提供的远程调用类 + * @param mqAdminExt rocketmq提供的远程调用类 * @param rocketmqCollectData rocketmq数据采集类 * @throws Exception 远程调用异常 */ - private void collectTopicData(DefaultMQAdminExt defaultMQAdminExt, RocketmqCollectData rocketmqCollectData) throws Exception { + private void collectTopicData(DefaultMQAdminExt mqAdminExt, RocketmqCollectData rocketmqCollectData) throws Exception { try { - TopicList topicList = defaultMQAdminExt.fetchAllTopicList(); + TopicList topicList = mqAdminExt.fetchAllTopicList(); Set topics = topicList.getTopicList() .stream() .filter(topic -> !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX))) diff --git a/collector/src/main/resources/META-INF/services/org.dromara.hertzbeat.collector.collect.AbstractCollect b/collector/src/main/resources/META-INF/services/org.dromara.hertzbeat.collector.collect.AbstractCollect index f7dc59c9f9b..294550b1eab 100644 --- a/collector/src/main/resources/META-INF/services/org.dromara.hertzbeat.collector.collect.AbstractCollect +++ b/collector/src/main/resources/META-INF/services/org.dromara.hertzbeat.collector.collect.AbstractCollect @@ -9,4 +9,4 @@ org.dromara.hertzbeat.collector.collect.snmp.SnmpCollectImpl org.dromara.hertzbeat.collector.collect.ssh.SshCollectImpl org.dromara.hertzbeat.collector.collect.telnet.TelnetCollectImpl org.dromara.hertzbeat.collector.collect.ftp.FtpCollectImpl -org.dromara.hertzbeat.collector.collect.mq.RocketMQSingleCollectImpl \ No newline at end of file +org.dromara.hertzbeat.collector.collect.mq.RocketmqSingleCollectImpl \ No newline at end of file From 4e0bde4b54b813cd6e29d794b8ace54c1d67c6c8 Mon Sep 17 00:00:00 2001 From: ceilzcx <1758619238@qq.com> Date: Wed, 28 Jun 2023 14:45:47 +0800 Subject: [PATCH 6/6] hashmap init capacity --- .../collector/collect/mq/RocketmqSingleCollectImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqSingleCollectImpl.java b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqSingleCollectImpl.java index 115d67a225f..4d0ce18f5c6 100644 --- a/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqSingleCollectImpl.java +++ b/collector/src/main/java/org/dromara/hertzbeat/collector/collect/mq/RocketmqSingleCollectImpl.java @@ -326,7 +326,7 @@ private void collectTopicData(DefaultMQAdminExt mqAdminExt, RocketmqCollectData .collect(Collectors.toSet()); List>> topicInfoList = new ArrayList<>(); for (String topic : topics) { - Map> topicQueueInfoTable = new HashMap<>(); + Map> topicQueueInfoTable = new HashMap<>(32); List topicQueueInfoList = new ArrayList<>(); // todo 查询topic的queue信息需要for循环调用 mqAdminExt.examineTopicStats(), topic数量很大的情况, 调用次数也会很多