You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@eventmesh.apache.org by "SPUERSAIYAN (via GitHub)" <gi...@apache.org> on 2026/09/19 12:47:07 UTC
[PR] feat: 采集 RocketMQ 主题和消费者偏移量 [eventmesh-dashboard]
SPUERSAIYAN opened a new pull request, #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232
使用共享的主题列表（topic-list）请求，采集主题偏移量、消费者偏移量、消费积压量以及连接数。
使用异步回调，并在将结果交给现有的采集流水线之前等待其完成。
通过 MapStruct 映射和共享的异常处理机制，复用现有的指标模型。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "SPUERSAIYAN (via GitHub)" <gi...@apache.org>.
SPUERSAIYAN commented on PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232#issuecomment-6101125388
> <img alt="图像" width="826" height="340" src="https://private-user-images.githubusercontent.com/41445332/670515923-07ca20d4-a170-4ab6-82be-5f6a792371cd.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3OTE2NTk0MDQsIm5iZiI6MTc5MTY1OTEwNCwicGF0aCI6Ii80MTQ0NTMzMi82NzA1MTU5MjMtMDdjYTIwZDQtYTE3MC00YWI2LTgyYmUtNWY2YTc5MjM3MWNkLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNjEwMTAlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjYxMDEwVDE5MDUwNFomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPWRkM2EwY2Q5NmMxYjdiMzMxMTRlNTJhNGY2YjdmMGZhYTJjMDQ0NzFkZWY4NjE4NDMwMTE1MzI1Y2UwOGZkYTkmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JnJlc3BvbnNlLWNvbnRlbnQtdHlwZT1pbWFnZSUyRnBuZyJ9.1NWY-wBIQJdzURNAAakb1v1Ius20A-ZsCJzQ-pxxTSM">
> 未修复项目会在后续提交中修复
好的，会在后续迭代中补充修复
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "Pil0tXia (via GitHub)" <gi...@apache.org>.
Pil0tXia commented on code in PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232#discussion_r4232891105
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/Rocketmq2ProducerOffset.java:
##########
@@ -28,8 +29,21 @@
@Data
@EqualsAndHashCode(callSuper = true)
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_producer_offset",
- defaultViewType = ReportViewType.HISTOGRAM, tableName = "rocketmq_producer_offset",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_producer_offset",
comment = "topic offset")
public class Rocketmq2ProducerOffset extends RuntimeLongValue {
+ @ReportTag
+ private String topicName;
+
+ private String queueId;
Review Comment:
**[P1] 新维度导致报表查询直接异常**
新增 `queueId` 后，IoTDB 查询生成器自动添加 `_parameter.queueId != null`，但实际查询参数 `SingleGeneralReportDO` 及其父类 `GeneralReportDO` 没有此属性。实际渲染抛出 `BuilderException` → `NoSuchPropertyException: SingleGeneralReportDO.queueId`，生产位点查询无法执行。
新增消费位点、积压时间模型也受同一问题影响；客户端模型新增的 `clientId` 同样缺少对应查询参数。需要同步更新查询参数及生成逻辑。
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/iotdb/IotDBReportMetaHandler.java:
##########
@@ -155,7 +156,7 @@ public String createTable() {
if (Objects.equals(filedName, "time")) {
stringBuilder.append(" time");
- } else if (filedName.endsWith("Id")) {
+ } else if (filedName.endsWith("Id") || field.isAnnotationPresent(ReportTag.class)) {
Review Comment:
**[P2] 新标签未接入查询和分组**
`ReportTag` 目前仅被建表逻辑识别，`query()` 仍只处理 `Id` 字段和数值字段。实际 `rocketmq_topic_messages_in` 查询仅选择 `time,value,value_window_count`，丢失 `topicName` 和 `window`；查询参数也不支持这些名称标签，因此无法按 Topic 或分钟/小时/日窗口过滤、区分结果，聚合时还会混算不同窗口。
需要同步支持新标签的选择、过滤和分组，包括消费组、线程池和磁盘维度。
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/CollectManage.java:
##########
@@ -164,6 +164,7 @@ public Thread newThread(Runnable r) {
public void init() {
this.dataSyncHandler.setPaddingService(paddingService);
+ this.dataSyncHandler.setReportEngine(this.reportHandlerManage.getReportEngine());
Review Comment:
**[P1] 新采集链路仍无法完成真实入库**
此处接通的真实落库链路仍被已有的同步容器缺陷阻断：非空批次经 `DataSyncHandler.padding()` → `AbstractReportEngine.batchInsertByClass()` → `IotDBReportEngine.batchInsert()` 进入 MyBatis `<foreach>` 时，`DataSyncHandler.CollectList.iterator()` 把 `ArrayList` 强转为 `Iterator`，实际复现 `ClassCastException`，SQL 尚未发送就失败。多节点乱序完成还会因 `restoreDataList.add(index, data)` 触发 `IndexOutOfBoundsException`。
这些辅助类缺陷在基线已存在，但阻断了本 PR 新增采集功能，需要补齐配套修复。本 PR 自带的 `DataSyncHandlerTest` 已复现乱序和并发同步失败。
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/mapstruct/report/RocketMQCollectMapper.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.dashboard.console.mapstruct.report;
+
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqBrokerMessagesIn;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqBrokerMessagesOut;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerConnectionNumber;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerFailedTps;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerGroupNumber;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerLagLatency;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerOffset;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerProcessTime;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerSuccessTps;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqGroupMessagesOut;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqMessagesInTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqMessagesOutTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDiskFreeBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDiskUsage;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDispatchBehindBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageFlushBehindBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageMessageReserveTime;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqThreadPoolWartermark;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqThroughputInTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqTopicMessagesIn;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqTopicNumber;
+
+import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
+import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
+
+import org.mapstruct.BeanMapping;
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+import org.mapstruct.factory.Mappers;
+
+/** 将 Broker 采样转换为语义对应的报表模型，由模型注解指定目标表。 */
+@Mapper
+public interface RocketMQCollectMapper {
+ RocketMQCollectMapper INSTANCE = Mappers.getMapper(RocketMQCollectMapper.class);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "groupName", source = "group")
+ @Mapping(target = "valueConnectionCount", source = "count")
+ RocketmqConsumerConnectionNumber connections(String group, long count);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topic")
+ @Mapping(target = "queueId", source = "queue")
+ @Mapping(target = "value", source = "offset.maxOffset")
+ @Mapping(target = "valueMinOffset", source = "offset.minOffset")
+ @Mapping(target = "valueLastUpdateTime", source = "offset.lastUpdateTimestamp")
+ Rocketmq2ProducerOffset topicOffset(String topic, String queue, TopicOffset offset);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topic")
+ @Mapping(target = "queueId", constant = "")
+ @Mapping(target = "valueMinOffsetSum", source = "min")
+ @Mapping(target = "valueMaxOffsetSum", source = "max")
+ @Mapping(target = "valueLastUpdateTime", source = "lastUpdate")
+ Rocketmq2ProducerOffset aggregateOffset(String topic, long min, long max, long lastUpdate);
Review Comment:
**[P1] 生产位点记录生成非法 INSERT**
队列记录与汇总记录使用同一个 `Rocketmq2ProducerOffset` 模型，但分别留下未赋值的数值字段：队列记录没有 `valueMinOffsetSum/valueMaxOffsetSum`，汇总记录没有 `value/valueMinOffset`。现有 `IotDBReportMetaHandler.insert()` 使用 `${item.xxx}` 直接拼接数值，MyBatis 将 `null` 渲染为空字符串，实际 SQL 出现类似 `...,12,,,123,5,'0','topicA')` 的空表达式，因此生产位点批次无法入库。
需要让插入器正确渲染 SQL `NULL`，或调整记录模型与映射。
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java:
##########
@@ -19,15 +19,730 @@
import org.apache.eventmesh.dashboard.common.annotation.ClusterTypeMark;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
+import org.apache.eventmesh.dashboard.common.model.metadata.RuntimeMetadata;
import org.apache.eventmesh.dashboard.console.function.report.collect.AbstractCollect;
+import org.apache.eventmesh.dashboard.console.mapstruct.report.RocketMQCollectMapper;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKTypeEnum;
+import org.apache.eventmesh.dashboard.core.function.SDK.operation.rocketmq.RocketMQRemotingSDKOperation.DefaultRemotingClient;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.netty.ResponseFuture;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetTopicStatsInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryConsumeTimeSpanRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryTopicConsumeByWhoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ViewBrokerStatsDataRequestHeader;
+
+import java.io.StringReader;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import lombok.Setter;
@ClusterTypeMark(clusterType = {ClusterType.STORAGE_ROCKETMQ_CLUSTER, ClusterType.STORAGE_ROCKETMQ_BROKER,
ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE, ClusterType.STORAGE_ROCKETMQ_BROKER_RAFT})
public class RocketMQCollect extends AbstractCollect {
+ private static final RocketMQCollectMapper MAPPER = RocketMQCollectMapper.INSTANCE;
+ private static final Pattern BYTE_COUNT = Pattern.compile("([0-9]+(?:\\.[0-9]+)?)\\s*(B|[KMGTPE]iB)");
+
+ private static final Pattern DISK_CAPACITY = Pattern.compile("Total : (.+), Free : (.+)\\.");
+
+ private String runtimeUnique;
+
+ @Setter
+ private DefaultRemotingClient defaultRemotingClient;
+
+ @Override
+ public void setRuntimeMetadata(RuntimeMetadata runtimeMetadata) {
+ super.setRuntimeMetadata(runtimeMetadata);
+ this.runtimeUnique = runtimeMetadata == null ? null : runtimeMetadata.getUnique();
+ }
@Override
protected void doCollect() {
+ if (this.runtimeUnique == null) {
+ return;
+ }
+ this.defaultRemotingClient = SDKManage.getInstance().getClient(SDKTypeEnum.ADMIN, this.runtimeUnique);
+ new CollectRound().executeAndAwait();
+ }
+
+ /**
+ * 一轮采集流程：
+ * 1. 先注册 Topic 和 Broker 两个方向的根请求，再分别启动，统一管理请求计数。
+ * 2. TopicCollect 采集 Topic 总数、位点、积压时间、消息统计及客户端处理指标；死信 Topic 不查询消费组。
+ * 3. BrokerCollect 采集配置消费组总数、消息统计和运行状态；响应直接调用父类 setData。
+ * 4. 等待所有异步请求完成后返回，父类再统一交付本轮数据。
+ * 5. 超时或中断时保留已经采集的数据，关闭本轮，拒绝尚未开始处理的迟到回调。
+ */
+ private class CollectRound {
+ private final DefaultRemotingClient client = defaultRemotingClient;
+ private final AtomicInteger pending = new AtomicInteger();
+ private final CountDownLatch completed = new CountDownLatch(1);
+ private final long deadline = System.currentTimeMillis() + 4000;
+ private volatile boolean closed;
+
+ private void executeAndAwait() {
+ try {
+ // 先注册所有根请求，避免同步回调在后续请求注册前让计数归零。
+ TopicCollect topics = new TopicCollect();
+ BrokerCollect broker = new BrokerCollect();
+ topics.start();
+ broker.start();
+ // 所有请求完成时计数归零并唤醒；超时则保留已采集的数据。
+ this.completed.await(Math.max(1, this.deadline - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ this.close();
+ }
+ }
+
+ private void close() {
+ this.closed = true;
+ }
Review Comment:
**[P2] 超时未隔离正在执行的回调**
`close()` 只设置 `closed`，`operationComplete()` 也只在入口检查一次。若回调在截止前进入、处理过程中超过四秒，采集线程会开始 `sync(current)` 并清理父类共享的 `current`，但已经开始的回调仍会继续向它写入，可能在同步期间修改数据、因 `current == null` 丢失样本，或写入下一轮的数据容器。
现有超时测试仅覆盖关闭后才开始的回调，需要协调关闭与在途写入，并固定本轮数据容器及写入归属。
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java:
##########
@@ -19,15 +19,730 @@
import org.apache.eventmesh.dashboard.common.annotation.ClusterTypeMark;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
+import org.apache.eventmesh.dashboard.common.model.metadata.RuntimeMetadata;
import org.apache.eventmesh.dashboard.console.function.report.collect.AbstractCollect;
+import org.apache.eventmesh.dashboard.console.mapstruct.report.RocketMQCollectMapper;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKTypeEnum;
+import org.apache.eventmesh.dashboard.core.function.SDK.operation.rocketmq.RocketMQRemotingSDKOperation.DefaultRemotingClient;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.netty.ResponseFuture;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetTopicStatsInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryConsumeTimeSpanRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryTopicConsumeByWhoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ViewBrokerStatsDataRequestHeader;
+
+import java.io.StringReader;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import lombok.Setter;
@ClusterTypeMark(clusterType = {ClusterType.STORAGE_ROCKETMQ_CLUSTER, ClusterType.STORAGE_ROCKETMQ_BROKER,
ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE, ClusterType.STORAGE_ROCKETMQ_BROKER_RAFT})
public class RocketMQCollect extends AbstractCollect {
+ private static final RocketMQCollectMapper MAPPER = RocketMQCollectMapper.INSTANCE;
+ private static final Pattern BYTE_COUNT = Pattern.compile("([0-9]+(?:\\.[0-9]+)?)\\s*(B|[KMGTPE]iB)");
+
+ private static final Pattern DISK_CAPACITY = Pattern.compile("Total : (.+), Free : (.+)\\.");
+
+ private String runtimeUnique;
+
+ @Setter
+ private DefaultRemotingClient defaultRemotingClient;
+
+ @Override
+ public void setRuntimeMetadata(RuntimeMetadata runtimeMetadata) {
+ super.setRuntimeMetadata(runtimeMetadata);
+ this.runtimeUnique = runtimeMetadata == null ? null : runtimeMetadata.getUnique();
+ }
@Override
protected void doCollect() {
+ if (this.runtimeUnique == null) {
+ return;
+ }
+ this.defaultRemotingClient = SDKManage.getInstance().getClient(SDKTypeEnum.ADMIN, this.runtimeUnique);
+ new CollectRound().executeAndAwait();
+ }
+
+ /**
+ * 一轮采集流程：
+ * 1. 先注册 Topic 和 Broker 两个方向的根请求，再分别启动，统一管理请求计数。
+ * 2. TopicCollect 采集 Topic 总数、位点、积压时间、消息统计及客户端处理指标；死信 Topic 不查询消费组。
+ * 3. BrokerCollect 采集配置消费组总数、消息统计和运行状态；响应直接调用父类 setData。
+ * 4. 等待所有异步请求完成后返回，父类再统一交付本轮数据。
+ * 5. 超时或中断时保留已经采集的数据，关闭本轮，拒绝尚未开始处理的迟到回调。
+ */
+ private class CollectRound {
+ private final DefaultRemotingClient client = defaultRemotingClient;
+ private final AtomicInteger pending = new AtomicInteger();
+ private final CountDownLatch completed = new CountDownLatch(1);
+ private final long deadline = System.currentTimeMillis() + 4000;
+ private volatile boolean closed;
+
+ private void executeAndAwait() {
+ try {
+ // 先注册所有根请求，避免同步回调在后续请求注册前让计数归零。
+ TopicCollect topics = new TopicCollect();
+ BrokerCollect broker = new BrokerCollect();
+ topics.start();
+ broker.start();
+ // 所有请求完成时计数归零并唤醒；超时则保留已采集的数据。
+ this.completed.await(Math.max(1, this.deadline - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ this.close();
+ }
+ }
+
+ private void close() {
+ this.closed = true;
+ }
+
+ /** 采集 Topic 总数、生产和消费位点、积压量及时间、消息统计、消费组连接数和客户端处理指标。 */
+ private class TopicCollect {
+ private final Set<String> connectedGroups = ConcurrentHashMap.newKeySet();
+ private final TopicsCallback topics =
+ new TopicsCallback(RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null));
+
+ private void start() {
+ this.topics.execute();
+ }
+
+ private void collectTopic(String topic) {
+ this.collectTopicStats(topic);
+ this.collectMessages(topic, null);
+ if (!topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
+ this.collectGroups(topic);
+ }
+ }
+
+ /** 原生统计键：写入按 Topic，取出按 Topic@消费组。 */
+ private void collectMessages(String topic, String group) {
+ ViewBrokerStatsDataRequestHeader header = new ViewBrokerStatsDataRequestHeader();
+ header.setStatsName(group == null ? "TOPIC_PUT_NUMS" : "GROUP_GET_NUMS");
+ header.setStatsKey(group == null ? topic : topic + "@" + group);
+ new MessageStatsCallback(RemotingCommand.createRequestCommand(RequestCode.VIEW_BROKER_STATS_DATA, header),
+ topic, group).execute();
+ }
+
+ private void collectTopicStats(String topic) {
+ GetTopicStatsInfoRequestHeader topicHeader = new GetTopicStatsInfoRequestHeader();
+ topicHeader.setTopic(topic);
+ RemotingCommand topicRequest = RemotingCommand.createRequestCommand(RequestCode.GET_TOPIC_STATS_INFO, topicHeader);
+ new TopicStatsCallback(topicRequest, topic).execute();
+ }
+
+ private void collectGroups(String topic) {
+ QueryTopicConsumeByWhoRequestHeader groupHeader = new QueryTopicConsumeByWhoRequestHeader();
+ groupHeader.setTopic(topic);
+ RemotingCommand groupRequest = RemotingCommand.createRequestCommand(RequestCode.QUERY_TOPIC_CONSUME_BY_WHO, groupHeader);
+ new GroupsCallback(groupRequest, topic).execute();
+ }
+
+ private void collectConnections(String group) {
+ GetConsumerConnectionListRequestHeader connectionHeader = new GetConsumerConnectionListRequestHeader();
+ connectionHeader.setConsumerGroup(group);
+ RemotingCommand connectionRequest = RemotingCommand.createRequestCommand(
+ RequestCode.GET_CONSUMER_CONNECTION_LIST, connectionHeader);
+ new ConnectionsCallback(connectionRequest, group).execute();
+ }
+
+ private void collectConsumeStats(String topic, String group) {
+ GetConsumeStatsRequestHeader consumeHeader = new GetConsumeStatsRequestHeader();
+ consumeHeader.setTopic(topic);
+ consumeHeader.setConsumerGroup(group);
+ RemotingCommand consumeRequest = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUME_STATS, consumeHeader);
+ new ConsumeStatsCallback(consumeRequest, topic, group).execute();
+ }
+
+ /** 步骤一：解析 Broker 的 Topic 列表，逐个启动采集。 */
+ private class TopicsCallback extends CollectCallback {
+ private TopicsCallback(RemotingCommand request) {
+ super(request);
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ TopicConfigSerializeWrapper topics =
+ RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), TopicConfigSerializeWrapper.class);
+ // 缺少列表不是零；仅对成功返回的完整配置表计数。
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body == null || !(body.get("topicConfigTable") instanceof Map<?, ?>)) {
+ return;
+ }
+ RocketMQCollect.this.setData(MAPPER.topicNumber((long) topics.getTopicConfigTable().size()));
+ for (String topic : topics.getTopicConfigTable().keySet()) {
+ collectTopic(topic);
+ }
+ }
+ }
+
+ private class MessageStatsCallback extends WindowStatsCallback {
+ private final String topic;
+ private final String group;
+
+ private MessageStatsCallback(RemotingCommand request, String topic, String group) {
+ super(request);
+ this.topic = topic;
+ this.group = group;
+ }
+
+ @Override
+ protected void recordWindow(String window, long count, float rate) {
+ RocketMQCollect.this.setData(this.group == null
+ ? MAPPER.topicMessages(this.topic, window, count, rate)
+ : MAPPER.groupMessages(this.topic, this.group, window, count, rate));
+ }
+ }
+
+ /** 步骤二：采集队列位点，并计算 Topic 汇总位点。 */
+ private class TopicStatsCallback extends CollectCallback {
+ private final String topic;
+
+ private TopicStatsCallback(RemotingCommand request, String topic) {
+ super(request);
+ this.topic = topic;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ TopicStatsTable stats =
+ RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), TopicStatsTable.class);
+ stats.getOffsetTable().forEach((queue, offset) -> {
+ String queueId = String.valueOf(queue.getQueueId());
+
+ RocketMQCollect.this.setData(MAPPER.topicOffset(topic, queueId, offset));
+ });
+ long min = stats.getOffsetTable().values().stream().mapToLong(offset -> offset.getMinOffset()).sum();
+ long max = stats.getOffsetTable().values().stream().mapToLong(offset -> offset.getMaxOffset()).sum();
+ long lastUpdate = stats.getOffsetTable().values().stream()
+ .mapToLong(offset -> offset.getLastUpdateTimestamp()).max().orElse(0L);
+
+ RocketMQCollect.this.setData(MAPPER.aggregateOffset(topic, min, max, lastUpdate));
+ }
+ }
+
+ /** 步骤三：发现 Topic 对应的消费组，启动连接与消费进度采集。 */
+ private class GroupsCallback extends CollectCallback {
+ private final String topic;
+
+ private GroupsCallback(RemotingCommand request, String topic) {
+ super(request);
+ this.topic = topic;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ GroupList groups = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), GroupList.class);
+ for (String group : groups.getGroupList()) {
+ if (TopicCollect.this.connectedGroups.add(group)) {
+ collectConnections(group);
+ }
+ collectConsumeStats(topic, group);
+ collectMessages(topic, group);
+ }
+ }
+ }
+
+ /** 步骤四：采集消费组连接数；同一消费组每轮只查询一次。 */
+ private class ConnectionsCallback extends CollectCallback {
+ private final String group;
+
+ private ConnectionsCallback(RemotingCommand request, String group) {
+ super(request);
+ this.group = group;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ ConsumerConnection connections = RemotingSerializable.decode(
+ responseFuture.getResponseCommand().getBody(), ConsumerConnection.class);
+
+ RocketMQCollect.this.setData(MAPPER.connections(group, connections.getConnectionSet().size()));
+ // 每个在线客户端分别采样，不把 Broker 取消息 TPS 当成处理成功 TPS。
+ Set<String> clients = ConcurrentHashMap.newKeySet();
+ connections.getConnectionSet().forEach(connection -> {
+ String clientId = connection.getClientId();
+ if (clientId != null && !clientId.isBlank() && clients.add(clientId)) {
+ GetConsumerRunningInfoRequestHeader header = new GetConsumerRunningInfoRequestHeader();
+ header.setConsumerGroup(this.group);
+ header.setClientId(clientId);
+ header.setJstackEnable(false);
+ new ClientStatsCallback(RemotingCommand.createRequestCommand(
+ RequestCode.GET_CONSUMER_RUNNING_INFO, header), this.group, clientId).execute();
+ }
+ });
+ }
+ }
+
+ /** 按 Topic、消费组和客户端采集消费成功 TPS、失败 TPS 和平均处理耗时。 */
+ private class ClientStatsCallback extends CollectCallback {
+ private final String group;
+ private final String clientId;
+
+ private ClientStatsCallback(RemotingCommand request, String group, String clientId) {
+ super(request);
+ this.group = group;
+ this.clientId = clientId;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body == null || !(body.get("statusTable") instanceof Map<?, ?> statuses)) {
+ return;
+ }
+ statuses.forEach((key, value) -> {
+ if (!(key instanceof String topic) || !(value instanceof Map<?, ?> status)) {
+ return;
+ }
+ Float success = nonnegativeFloatOrZero(status.get("consumeOKTPS"));
+ Float failed = nonnegativeFloatOrZero(status.get("consumeFailedTPS"));
+ Float elapsed = nonnegativeFloatOrZero(status.get("consumeRT"));
+ RocketMQCollect.this.setData(MAPPER.consumerSuccess(topic, this.group, this.clientId, success));
+ RocketMQCollect.this.setData(MAPPER.consumerFailed(topic, this.group, this.clientId, failed));
+ RocketMQCollect.this.setData(MAPPER.consumerProcessTime(topic, this.group, this.clientId, elapsed));
+ });
+ }
+ }
+
+ /** 采集各 Topic、消费组和队列中最早未消费消息的积压时间。 */
+ private class LagCallback extends CollectCallback {
+ private final String topic;
+ private final String group;
+
+ private LagCallback(RemotingCommand request, String topic, String group) {
+ super(request);
+ this.topic = topic;
+ this.group = group;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body == null || !(body.get("consumeTimeSpanSet") instanceof List<?> spans)) {
+ return;
+ }
+ for (Object item : spans) {
+ if (!(item instanceof Map<?, ?> span) || !(span.get("messageQueue") instanceof Map<?, ?> queue)
+ || !this.topic.equals(queue.get("topic"))) {
+ continue;
+ }
+ Long queueId = nonnegativeLong(queue.get("queueId"));
+ Long delay = nonnegativeLong(span.get("delayTime"));
+ Long earliest = nonnegativeLong(span.get("minTimeStamp"));
+ // 位点已过期或未初始化时，原生接口可能用 -1 时间戳计算出近似当前纪元的延迟。
+ if (queueId != null && delay != null && earliest != null && earliest > 0
+ && delay <= System.currentTimeMillis() - earliest) {
+ RocketMQCollect.this.setData(MAPPER.consumerLag(this.topic, this.group, queueId.toString(), delay));
+ }
+ }
+ }
+ }
+
+ /** 步骤五：采集 Topic 与消费组对应的队列位点和积压量。 */
+ private class ConsumeStatsCallback extends CollectCallback {
+ private final String topic;
+ private final String group;
+
+ private ConsumeStatsCallback(RemotingCommand request, String topic, String group) {
+ super(request);
+ this.topic = topic;
+ this.group = group;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ ConsumeStats stats = RemotingSerializable.decode(
+ responseFuture.getResponseCommand().getBody(), ConsumeStats.class);
+ stats.getOffsetTable().forEach((queue, offset) -> {
+ if (!topic.equals(queue.getTopic())) {
+ return;
+ }
+ String queueId = String.valueOf(queue.getQueueId());
+
+ RocketMQCollect.this.setData(MAPPER.consumerOffset(topic, group, queueId, offset));
+ });
+ if (!stats.getOffsetTable().isEmpty()) {
+ QueryConsumeTimeSpanRequestHeader header = new QueryConsumeTimeSpanRequestHeader();
+ header.setTopic(this.topic);
+ header.setGroup(this.group);
+ new LagCallback(RemotingCommand.createRequestCommand(RequestCode.QUERY_CONSUME_TIME_SPAN, header),
+ this.topic, this.group).execute();
+ }
+ }
+ }
+ }
+
+ /** 采集消费组总数、Broker 消息统计、存储积压、消息保留时间、磁盘使用率及剩余空间、线程池队列长度。 */
+ private class BrokerCollect {
+ private final BrokerConfigCallback config = new BrokerConfigCallback();
+ private final BrokerRuntimeCallback runtime = new BrokerRuntimeCallback();
+ private final GroupCountCallback groups = new GroupCountCallback();
+
+ private void start() {
+ this.config.execute();
+ this.runtime.execute();
+ this.groups.execute();
+ }
+
+ /** 采集 Broker 配置的消费组总数。 */
+ private class GroupCountCallback extends CollectCallback {
+ private GroupCountCallback() {
+ super(RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG,
+ new GetAllSubscriptionGroupRequestHeader()));
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body != null && body.get("subscriptionGroupTable") instanceof Map<?, ?> groups) {
+ RocketMQCollect.this.setData(MAPPER.groupNumber((long) groups.size()));
+ }
+ }
+ }
+
+ private void collectStats(String cluster, boolean incoming) {
+ ViewBrokerStatsDataRequestHeader header = new ViewBrokerStatsDataRequestHeader();
+ header.setStatsName(incoming ? "BROKER_PUT_NUMS" : "BROKER_GET_NUMS");
+ header.setStatsKey(cluster);
+ new BrokerStatsCallback(RemotingCommand.createRequestCommand(RequestCode.VIEW_BROKER_STATS_DATA, header), incoming).execute();
+ }
+
+ /** 步骤一：获取 Broker 所属集群名，启动消息写入和取出统计采集。 */
+ private class BrokerConfigCallback extends CollectCallback {
+ private BrokerConfigCallback() {
+ super(RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CONFIG, null));
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) throws Exception {
+ Properties properties = new Properties();
+ properties.load(new StringReader(new String(responseFuture.getResponseCommand().getBody(), StandardCharsets.UTF_8)));
+ String cluster = properties.getProperty("brokerClusterName");
+ if (cluster == null || cluster.isBlank()) {
+ return;
+ }
+ collectStats(cluster, true);
+ collectStats(cluster, false);
+ }
+ }
+
+ /** 步骤二：采集分钟、小时、日窗口的消息写入和取出数量及 TPS。 */
+ private class BrokerStatsCallback extends WindowStatsCallback {
+ private final boolean incoming;
+
+ private BrokerStatsCallback(RemotingCommand request, boolean incoming) {
+ super(request);
+ this.incoming = incoming;
+ }
+
+ @Override
+ protected void recordWindow(String window, long count, float rate) {
+ RocketMQCollect.this.setData(this.incoming
+ ? MAPPER.brokerMessagesIn(window, count, rate) : MAPPER.brokerMessagesOut(window, count, rate));
+ }
+ }
+
+ /** 步骤三：采集存储分发和刷盘积压、消息保留时间、磁盘使用率及剩余空间、线程池队列长度。 */
+ private class BrokerRuntimeCallback extends CollectCallback {
+ private BrokerRuntimeCallback() {
+ super(RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_RUNTIME_INFO, null));
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> response = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (response == null || !(response.get("table") instanceof Map<?, ?> table)) {
+ return;
+ }
+ Long messagesIn = nonnegativeLong(table.get("msgPutTotalTodayNow"));
+ if (messagesIn != null) {
+ RocketMQCollect.this.setData(MAPPER.messagesInTotal(messagesIn));
+ }
+ Long bytesIn = nonnegativeLong(table.get("putMessageSizeTotal"));
+ if (bytesIn != null) {
+ RocketMQCollect.this.setData(MAPPER.throughputInTotal(bytesIn));
+ }
+ Long messagesOut = nonnegativeLong(table.get("msgGetTotalTodayNow"));
+ if (messagesOut != null) {
+ RocketMQCollect.this.setData(MAPPER.messagesOutTotal(messagesOut));
+ }
+ Long dispatch = nonnegativeLong(table.get("dispatchBehindBytes"));
+ if (dispatch != null) {
+ RocketMQCollect.this.setData(MAPPER.dispatchBytes(dispatch));
+ }
+ this.collectFlush(table);
+ this.collectDisk(table);
+ Long earliest = nonnegativeLong(table.get("earliestMessageTimeStamp"));
+ long now = System.currentTimeMillis();
+ if (earliest != null && earliest > 0 && earliest <= now) {
+ RocketMQCollect.this.setData(MAPPER.reserveTime(now - earliest));
+ }
+ for (String pool : new String[] {"send", "pull", "litePull", "query", "ack"}) {
+ this.collectThreadPool(table, pool, pool + "ThreadPoolQueueSize");
+ }
+ this.collectThreadPool(table, "endTransaction", "EndTransactionQueueSize");
+ }
+
+ private void collectDisk(Map<?, ?> table) {
+ table.forEach((key, value) -> {
+ if (!(key instanceof String name)) {
+ return;
+ }
+ String type;
+ String path = "";
+ if (name.equals("commitLogDiskRatio")) {
+ type = "commitlog_min";
+ } else if (name.startsWith("commitLogDiskRatio_")) {
+ type = "commitlog";
+ path = name.substring("commitLogDiskRatio_".length());
+ } else if (name.equals("consumeQueueDiskRatio")) {
+ type = "consumequeue";
+ } else {
+ return;
+ }
+ Float ratio = nonnegativeFloat(value);
+ if (ratio != null && ratio <= 1) {
+ RocketMQCollect.this.setData(MAPPER.diskUsage(type, path, ratio));
+ }
+ });
+ Object capacity = table.get("commitLogDirCapacity");
+ if (capacity instanceof String text) {
+ Matcher matcher = DISK_CAPACITY.matcher(text.trim());
+ if (matcher.matches()) {
+ Long total = byteCount(matcher.group(1));
+ Long free = byteCount(matcher.group(2));
+ if (total != null && total > 0 && free != null && free <= total) {
+ RocketMQCollect.this.setData(MAPPER.diskFreeBytes(free));
+ }
+ }
+ }
+ }
+
+ private void collectFlush(Map<?, ?> table) {
+ Long flush = byteCount(table.get("remainHowManyDataToFlush"));
+ // 启用暂存池时，原生刷盘积压还包括待提交字节；字段缺席表示未启用该分支。
+ Long commit = table.containsKey("remainHowManyDataToCommit")
+ ? byteCount(table.get("remainHowManyDataToCommit")) : Long.valueOf(0L);
+ if (flush != null && commit != null && flush <= Long.MAX_VALUE - commit) {
+ RocketMQCollect.this.setData(MAPPER.flushBytes(flush + commit));
+ }
+ }
+
+ private void collectThreadPool(Map<?, ?> table, String pool, String key) {
+ Long size = nonnegativeLong(table.get(key));
+ if (size != null) {
+ RocketMQCollect.this.setData(MAPPER.threadPool(pool, size));
+ }
+ }
+ }
+
+ private Long byteCount(Object value) {
+ if (!(value instanceof String text)) {
+ return null;
+ }
+ Matcher matcher = BYTE_COUNT.matcher(text.trim());
+ if (!matcher.matches()) {
+ return null;
+ }
+ String unit = matcher.group(2);
+ int exponent = "B".equals(unit) ? 0 : "KMGTPE".indexOf(unit.charAt(0)) + 1;
+ try {
+ // Broker 已将二进制字节数格式化并舍入，此处只能还原近似值，不能声称字节级精确。
+ return new BigDecimal(matcher.group(1)).multiply(BigDecimal.valueOf(1024).pow(exponent))
+ .setScale(0, RoundingMode.HALF_UP).longValueExact();
+ } catch (ArithmeticException e) {
+ return null;
+ }
+ }
+ }
+
+ private Long nonnegativeLong(Object value) {
+ if (value == null) {
+ return null;
+ }
+ try {
+ long number = Long.parseLong(value.toString());
+ return number >= 0 ? number : null;
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ private Float nonnegativeFloatOrZero(Object value) {
+ Float result = this.nonnegativeFloat(value);
+ return result == null ? 0f : result;
+ }
+
+ private Float nonnegativeFloat(Object value) {
+ if (value == null) {
+ return null;
+ }
+ try {
+ double number = Double.parseDouble(value.toString());
+ float result = (float) number;
+ return number >= 0 && Float.isFinite(result) ? result : null;
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ /** 解析 Broker、Topic 和消费组的分钟、小时、日消息统计窗口。 */
+ private abstract class WindowStatsCallback extends CollectCallback {
+ private WindowStatsCallback(RemotingCommand request) {
+ super(request);
+ }
+
+ @Override
+ protected final void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> stats = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ this.collectWindow(stats, "statsMinute", "minute");
+ this.collectWindow(stats, "statsHour", "hour");
+ this.collectWindow(stats, "statsDay", "day");
+ }
+
+ private void collectWindow(Map<?, ?> stats, String key, String window) {
+ if (stats == null || !(stats.get(key) instanceof Map<?, ?> sample)
+ || !(sample.get("sum") instanceof Number count) || !(sample.get("tps") instanceof Number rate)) {
+ return;
+ }
+ try {
+ long value = Long.parseLong(count.toString());
+ float tps = rate.floatValue();
+ if (value >= 0 && Float.isFinite(tps) && rate.doubleValue() >= 0) {
+ this.recordWindow(window, value, tps);
+ }
+ } catch (NumberFormatException e) {
+ // 小数、溢出或格式无效的消息数不参与采集。
+ }
+ }
+
+ protected abstract void recordWindow(String window, long count, float rate);
+ }
+
+ /** 统一处理请求发送、响应校验、异常和完成计数，具体响应由各采集回调解析。 */
+ private abstract class CollectCallback implements InvokeCallback {
+ private final RemotingCommand request;
+ private final AtomicBoolean finished = new AtomicBoolean();
+
+ private CollectCallback(RemotingCommand request) {
+ this.request = request;
+ // 请求发送前 +1；子请求先计数，父请求处理结束后才 -1。
+ CollectRound.this.pending.incrementAndGet();
+ }
+
+ // 发送异常也进入失败回调，每个请求只扣减一次。
+ protected final void execute() {
+ executeSafely(this,
+ () -> client.invokeAsync(this.request, this.timeoutMillis(), this));
+ }
+
+ private long timeoutMillis() throws TimeoutException {
+ long remaining = CollectRound.this.deadline - System.currentTimeMillis();
+ if (CollectRound.this.closed || remaining <= 0) {
+ throw new TimeoutException("RocketMQ collection deadline reached");
+ }
+ return Math.min(3000, remaining);
+ }
+
+ @Override
+ public final void operationComplete(ResponseFuture responseFuture) {
+ if (!this.finished.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ if (CollectRound.this.closed) {
+ return;
+ }
+ RemotingCommand response = responseFuture.getResponseCommand();
+ if (response == null) {
+ throw new IllegalStateException("Missing RocketMQ response");
+ }
+ if (response.getCode() != ResponseCode.SUCCESS || response.getBody() == null || response.getBody().length == 0) {
+ return;
+ }
+ this.handleResponse(responseFuture);
+ } catch (Exception e) {
+ this.restoreInterrupt(e);
Review Comment:
**[P1] 并发回调会丢失采集样本**
RocketMQ 默认使用多线程执行响应回调，而 `handleResponse()` 最终经父类 `setData()` 写入 `RestoreData` 中普通 `HashMap` 和 `ArrayList`，没有同步保护。多个 Topic/窗口响应同时写入时，`computeIfAbsent()` 可抛出 `ConcurrentModificationException`，`ArrayList.add()` 可丢样本并产生 `null` 行；这里又直接吞掉异常，最终仍将该请求视为完成。
使用实际 `RestoreData.setData()` 的并发探针已复现样本丢失和异常。`pending` 计数只能等待完成，不能保证写入安全，需要串行化写入或采用安全的数据容器。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "SPUERSAIYAN (via GitHub)" <gi...@apache.org>.
SPUERSAIYAN commented on code in PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232#discussion_r4237706301
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/Rocketmq2ProducerOffset.java:
##########
@@ -28,8 +29,21 @@
@Data
@EqualsAndHashCode(callSuper = true)
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_producer_offset",
- defaultViewType = ReportViewType.HISTOGRAM, tableName = "rocketmq_producer_offset",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_producer_offset",
comment = "topic offset")
public class Rocketmq2ProducerOffset extends RuntimeLongValue {
+ @ReportTag
+ private String topicName;
+
+ private String queueId;
Review Comment:
已确认该问题。这个PR 仅负责指标采集和数据交付，报表查询适配不在本次范围内，后续异步写库的时候会进行同步修复
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "SPUERSAIYAN (via GitHub)" <gi...@apache.org>.
SPUERSAIYAN commented on code in PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232#discussion_r4237815645
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java:
##########
@@ -19,15 +19,730 @@
import org.apache.eventmesh.dashboard.common.annotation.ClusterTypeMark;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
+import org.apache.eventmesh.dashboard.common.model.metadata.RuntimeMetadata;
import org.apache.eventmesh.dashboard.console.function.report.collect.AbstractCollect;
+import org.apache.eventmesh.dashboard.console.mapstruct.report.RocketMQCollectMapper;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKTypeEnum;
+import org.apache.eventmesh.dashboard.core.function.SDK.operation.rocketmq.RocketMQRemotingSDKOperation.DefaultRemotingClient;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.netty.ResponseFuture;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetTopicStatsInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryConsumeTimeSpanRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryTopicConsumeByWhoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ViewBrokerStatsDataRequestHeader;
+
+import java.io.StringReader;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import lombok.Setter;
@ClusterTypeMark(clusterType = {ClusterType.STORAGE_ROCKETMQ_CLUSTER, ClusterType.STORAGE_ROCKETMQ_BROKER,
ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE, ClusterType.STORAGE_ROCKETMQ_BROKER_RAFT})
public class RocketMQCollect extends AbstractCollect {
+ private static final RocketMQCollectMapper MAPPER = RocketMQCollectMapper.INSTANCE;
+ private static final Pattern BYTE_COUNT = Pattern.compile("([0-9]+(?:\\.[0-9]+)?)\\s*(B|[KMGTPE]iB)");
+
+ private static final Pattern DISK_CAPACITY = Pattern.compile("Total : (.+), Free : (.+)\\.");
+
+ private String runtimeUnique;
+
+ @Setter
+ private DefaultRemotingClient defaultRemotingClient;
+
+ @Override
+ public void setRuntimeMetadata(RuntimeMetadata runtimeMetadata) {
+ super.setRuntimeMetadata(runtimeMetadata);
+ this.runtimeUnique = runtimeMetadata == null ? null : runtimeMetadata.getUnique();
+ }
@Override
protected void doCollect() {
+ if (this.runtimeUnique == null) {
+ return;
+ }
+ this.defaultRemotingClient = SDKManage.getInstance().getClient(SDKTypeEnum.ADMIN, this.runtimeUnique);
+ new CollectRound().executeAndAwait();
+ }
+
+ /**
+ * 一轮采集流程：
+ * 1. 先注册 Topic 和 Broker 两个方向的根请求，再分别启动，统一管理请求计数。
+ * 2. TopicCollect 采集 Topic 总数、位点、积压时间、消息统计及客户端处理指标；死信 Topic 不查询消费组。
+ * 3. BrokerCollect 采集配置消费组总数、消息统计和运行状态；响应直接调用父类 setData。
+ * 4. 等待所有异步请求完成后返回，父类再统一交付本轮数据。
+ * 5. 超时或中断时保留已经采集的数据，关闭本轮，拒绝尚未开始处理的迟到回调。
+ */
+ private class CollectRound {
+ private final DefaultRemotingClient client = defaultRemotingClient;
+ private final AtomicInteger pending = new AtomicInteger();
+ private final CountDownLatch completed = new CountDownLatch(1);
+ private final long deadline = System.currentTimeMillis() + 4000;
+ private volatile boolean closed;
+
+ private void executeAndAwait() {
+ try {
+ // 先注册所有根请求，避免同步回调在后续请求注册前让计数归零。
+ TopicCollect topics = new TopicCollect();
+ BrokerCollect broker = new BrokerCollect();
+ topics.start();
+ broker.start();
+ // 所有请求完成时计数归零并唤醒；超时则保留已采集的数据。
+ this.completed.await(Math.max(1, this.deadline - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ this.close();
+ }
+ }
+
+ private void close() {
+ this.closed = true;
+ }
+
+ /** 采集 Topic 总数、生产和消费位点、积压量及时间、消息统计、消费组连接数和客户端处理指标。 */
+ private class TopicCollect {
+ private final Set<String> connectedGroups = ConcurrentHashMap.newKeySet();
+ private final TopicsCallback topics =
+ new TopicsCallback(RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null));
+
+ private void start() {
+ this.topics.execute();
+ }
+
+ private void collectTopic(String topic) {
+ this.collectTopicStats(topic);
+ this.collectMessages(topic, null);
+ if (!topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
+ this.collectGroups(topic);
+ }
+ }
+
+ /** 原生统计键：写入按 Topic，取出按 Topic@消费组。 */
+ private void collectMessages(String topic, String group) {
+ ViewBrokerStatsDataRequestHeader header = new ViewBrokerStatsDataRequestHeader();
+ header.setStatsName(group == null ? "TOPIC_PUT_NUMS" : "GROUP_GET_NUMS");
+ header.setStatsKey(group == null ? topic : topic + "@" + group);
+ new MessageStatsCallback(RemotingCommand.createRequestCommand(RequestCode.VIEW_BROKER_STATS_DATA, header),
+ topic, group).execute();
+ }
+
+ private void collectTopicStats(String topic) {
+ GetTopicStatsInfoRequestHeader topicHeader = new GetTopicStatsInfoRequestHeader();
+ topicHeader.setTopic(topic);
+ RemotingCommand topicRequest = RemotingCommand.createRequestCommand(RequestCode.GET_TOPIC_STATS_INFO, topicHeader);
+ new TopicStatsCallback(topicRequest, topic).execute();
+ }
+
+ private void collectGroups(String topic) {
+ QueryTopicConsumeByWhoRequestHeader groupHeader = new QueryTopicConsumeByWhoRequestHeader();
+ groupHeader.setTopic(topic);
+ RemotingCommand groupRequest = RemotingCommand.createRequestCommand(RequestCode.QUERY_TOPIC_CONSUME_BY_WHO, groupHeader);
+ new GroupsCallback(groupRequest, topic).execute();
+ }
+
+ private void collectConnections(String group) {
+ GetConsumerConnectionListRequestHeader connectionHeader = new GetConsumerConnectionListRequestHeader();
+ connectionHeader.setConsumerGroup(group);
+ RemotingCommand connectionRequest = RemotingCommand.createRequestCommand(
+ RequestCode.GET_CONSUMER_CONNECTION_LIST, connectionHeader);
+ new ConnectionsCallback(connectionRequest, group).execute();
+ }
+
+ private void collectConsumeStats(String topic, String group) {
+ GetConsumeStatsRequestHeader consumeHeader = new GetConsumeStatsRequestHeader();
+ consumeHeader.setTopic(topic);
+ consumeHeader.setConsumerGroup(group);
+ RemotingCommand consumeRequest = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUME_STATS, consumeHeader);
+ new ConsumeStatsCallback(consumeRequest, topic, group).execute();
+ }
+
+ /** 步骤一：解析 Broker 的 Topic 列表，逐个启动采集。 */
+ private class TopicsCallback extends CollectCallback {
+ private TopicsCallback(RemotingCommand request) {
+ super(request);
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ TopicConfigSerializeWrapper topics =
+ RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), TopicConfigSerializeWrapper.class);
+ // 缺少列表不是零；仅对成功返回的完整配置表计数。
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body == null || !(body.get("topicConfigTable") instanceof Map<?, ?>)) {
+ return;
+ }
+ RocketMQCollect.this.setData(MAPPER.topicNumber((long) topics.getTopicConfigTable().size()));
+ for (String topic : topics.getTopicConfigTable().keySet()) {
+ collectTopic(topic);
+ }
+ }
+ }
+
+ private class MessageStatsCallback extends WindowStatsCallback {
+ private final String topic;
+ private final String group;
+
+ private MessageStatsCallback(RemotingCommand request, String topic, String group) {
+ super(request);
+ this.topic = topic;
+ this.group = group;
+ }
+
+ @Override
+ protected void recordWindow(String window, long count, float rate) {
+ RocketMQCollect.this.setData(this.group == null
+ ? MAPPER.topicMessages(this.topic, window, count, rate)
+ : MAPPER.groupMessages(this.topic, this.group, window, count, rate));
+ }
+ }
+
+ /** 步骤二：采集队列位点，并计算 Topic 汇总位点。 */
+ private class TopicStatsCallback extends CollectCallback {
+ private final String topic;
+
+ private TopicStatsCallback(RemotingCommand request, String topic) {
+ super(request);
+ this.topic = topic;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ TopicStatsTable stats =
+ RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), TopicStatsTable.class);
+ stats.getOffsetTable().forEach((queue, offset) -> {
+ String queueId = String.valueOf(queue.getQueueId());
+
+ RocketMQCollect.this.setData(MAPPER.topicOffset(topic, queueId, offset));
+ });
+ long min = stats.getOffsetTable().values().stream().mapToLong(offset -> offset.getMinOffset()).sum();
+ long max = stats.getOffsetTable().values().stream().mapToLong(offset -> offset.getMaxOffset()).sum();
+ long lastUpdate = stats.getOffsetTable().values().stream()
+ .mapToLong(offset -> offset.getLastUpdateTimestamp()).max().orElse(0L);
+
+ RocketMQCollect.this.setData(MAPPER.aggregateOffset(topic, min, max, lastUpdate));
+ }
+ }
+
+ /** 步骤三：发现 Topic 对应的消费组，启动连接与消费进度采集。 */
+ private class GroupsCallback extends CollectCallback {
+ private final String topic;
+
+ private GroupsCallback(RemotingCommand request, String topic) {
+ super(request);
+ this.topic = topic;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ GroupList groups = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), GroupList.class);
+ for (String group : groups.getGroupList()) {
+ if (TopicCollect.this.connectedGroups.add(group)) {
+ collectConnections(group);
+ }
+ collectConsumeStats(topic, group);
+ collectMessages(topic, group);
+ }
+ }
+ }
+
+ /** 步骤四：采集消费组连接数；同一消费组每轮只查询一次。 */
+ private class ConnectionsCallback extends CollectCallback {
+ private final String group;
+
+ private ConnectionsCallback(RemotingCommand request, String group) {
+ super(request);
+ this.group = group;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ ConsumerConnection connections = RemotingSerializable.decode(
+ responseFuture.getResponseCommand().getBody(), ConsumerConnection.class);
+
+ RocketMQCollect.this.setData(MAPPER.connections(group, connections.getConnectionSet().size()));
+ // 每个在线客户端分别采样，不把 Broker 取消息 TPS 当成处理成功 TPS。
+ Set<String> clients = ConcurrentHashMap.newKeySet();
+ connections.getConnectionSet().forEach(connection -> {
+ String clientId = connection.getClientId();
+ if (clientId != null && !clientId.isBlank() && clients.add(clientId)) {
+ GetConsumerRunningInfoRequestHeader header = new GetConsumerRunningInfoRequestHeader();
+ header.setConsumerGroup(this.group);
+ header.setClientId(clientId);
+ header.setJstackEnable(false);
+ new ClientStatsCallback(RemotingCommand.createRequestCommand(
+ RequestCode.GET_CONSUMER_RUNNING_INFO, header), this.group, clientId).execute();
+ }
+ });
+ }
+ }
+
+ /** 按 Topic、消费组和客户端采集消费成功 TPS、失败 TPS 和平均处理耗时。 */
+ private class ClientStatsCallback extends CollectCallback {
+ private final String group;
+ private final String clientId;
+
+ private ClientStatsCallback(RemotingCommand request, String group, String clientId) {
+ super(request);
+ this.group = group;
+ this.clientId = clientId;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body == null || !(body.get("statusTable") instanceof Map<?, ?> statuses)) {
+ return;
+ }
+ statuses.forEach((key, value) -> {
+ if (!(key instanceof String topic) || !(value instanceof Map<?, ?> status)) {
+ return;
+ }
+ Float success = nonnegativeFloatOrZero(status.get("consumeOKTPS"));
+ Float failed = nonnegativeFloatOrZero(status.get("consumeFailedTPS"));
+ Float elapsed = nonnegativeFloatOrZero(status.get("consumeRT"));
+ RocketMQCollect.this.setData(MAPPER.consumerSuccess(topic, this.group, this.clientId, success));
+ RocketMQCollect.this.setData(MAPPER.consumerFailed(topic, this.group, this.clientId, failed));
+ RocketMQCollect.this.setData(MAPPER.consumerProcessTime(topic, this.group, this.clientId, elapsed));
+ });
+ }
+ }
+
+ /** 采集各 Topic、消费组和队列中最早未消费消息的积压时间。 */
+ private class LagCallback extends CollectCallback {
+ private final String topic;
+ private final String group;
+
+ private LagCallback(RemotingCommand request, String topic, String group) {
+ super(request);
+ this.topic = topic;
+ this.group = group;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body == null || !(body.get("consumeTimeSpanSet") instanceof List<?> spans)) {
+ return;
+ }
+ for (Object item : spans) {
+ if (!(item instanceof Map<?, ?> span) || !(span.get("messageQueue") instanceof Map<?, ?> queue)
+ || !this.topic.equals(queue.get("topic"))) {
+ continue;
+ }
+ Long queueId = nonnegativeLong(queue.get("queueId"));
+ Long delay = nonnegativeLong(span.get("delayTime"));
+ Long earliest = nonnegativeLong(span.get("minTimeStamp"));
+ // 位点已过期或未初始化时，原生接口可能用 -1 时间戳计算出近似当前纪元的延迟。
+ if (queueId != null && delay != null && earliest != null && earliest > 0
+ && delay <= System.currentTimeMillis() - earliest) {
+ RocketMQCollect.this.setData(MAPPER.consumerLag(this.topic, this.group, queueId.toString(), delay));
+ }
+ }
+ }
+ }
+
+ /** 步骤五：采集 Topic 与消费组对应的队列位点和积压量。 */
+ private class ConsumeStatsCallback extends CollectCallback {
+ private final String topic;
+ private final String group;
+
+ private ConsumeStatsCallback(RemotingCommand request, String topic, String group) {
+ super(request);
+ this.topic = topic;
+ this.group = group;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ ConsumeStats stats = RemotingSerializable.decode(
+ responseFuture.getResponseCommand().getBody(), ConsumeStats.class);
+ stats.getOffsetTable().forEach((queue, offset) -> {
+ if (!topic.equals(queue.getTopic())) {
+ return;
+ }
+ String queueId = String.valueOf(queue.getQueueId());
+
+ RocketMQCollect.this.setData(MAPPER.consumerOffset(topic, group, queueId, offset));
+ });
+ if (!stats.getOffsetTable().isEmpty()) {
+ QueryConsumeTimeSpanRequestHeader header = new QueryConsumeTimeSpanRequestHeader();
+ header.setTopic(this.topic);
+ header.setGroup(this.group);
+ new LagCallback(RemotingCommand.createRequestCommand(RequestCode.QUERY_CONSUME_TIME_SPAN, header),
+ this.topic, this.group).execute();
+ }
+ }
+ }
+ }
+
+ /** 采集消费组总数、Broker 消息统计、存储积压、消息保留时间、磁盘使用率及剩余空间、线程池队列长度。 */
+ private class BrokerCollect {
+ private final BrokerConfigCallback config = new BrokerConfigCallback();
+ private final BrokerRuntimeCallback runtime = new BrokerRuntimeCallback();
+ private final GroupCountCallback groups = new GroupCountCallback();
+
+ private void start() {
+ this.config.execute();
+ this.runtime.execute();
+ this.groups.execute();
+ }
+
+ /** 采集 Broker 配置的消费组总数。 */
+ private class GroupCountCallback extends CollectCallback {
+ private GroupCountCallback() {
+ super(RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG,
+ new GetAllSubscriptionGroupRequestHeader()));
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body != null && body.get("subscriptionGroupTable") instanceof Map<?, ?> groups) {
+ RocketMQCollect.this.setData(MAPPER.groupNumber((long) groups.size()));
+ }
+ }
+ }
+
+ private void collectStats(String cluster, boolean incoming) {
+ ViewBrokerStatsDataRequestHeader header = new ViewBrokerStatsDataRequestHeader();
+ header.setStatsName(incoming ? "BROKER_PUT_NUMS" : "BROKER_GET_NUMS");
+ header.setStatsKey(cluster);
+ new BrokerStatsCallback(RemotingCommand.createRequestCommand(RequestCode.VIEW_BROKER_STATS_DATA, header), incoming).execute();
+ }
+
+ /** 步骤一：获取 Broker 所属集群名，启动消息写入和取出统计采集。 */
+ private class BrokerConfigCallback extends CollectCallback {
+ private BrokerConfigCallback() {
+ super(RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CONFIG, null));
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) throws Exception {
+ Properties properties = new Properties();
+ properties.load(new StringReader(new String(responseFuture.getResponseCommand().getBody(), StandardCharsets.UTF_8)));
+ String cluster = properties.getProperty("brokerClusterName");
+ if (cluster == null || cluster.isBlank()) {
+ return;
+ }
+ collectStats(cluster, true);
+ collectStats(cluster, false);
+ }
+ }
+
+ /** 步骤二：采集分钟、小时、日窗口的消息写入和取出数量及 TPS。 */
+ private class BrokerStatsCallback extends WindowStatsCallback {
+ private final boolean incoming;
+
+ private BrokerStatsCallback(RemotingCommand request, boolean incoming) {
+ super(request);
+ this.incoming = incoming;
+ }
+
+ @Override
+ protected void recordWindow(String window, long count, float rate) {
+ RocketMQCollect.this.setData(this.incoming
+ ? MAPPER.brokerMessagesIn(window, count, rate) : MAPPER.brokerMessagesOut(window, count, rate));
+ }
+ }
+
+ /** 步骤三：采集存储分发和刷盘积压、消息保留时间、磁盘使用率及剩余空间、线程池队列长度。 */
+ private class BrokerRuntimeCallback extends CollectCallback {
+ private BrokerRuntimeCallback() {
+ super(RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_RUNTIME_INFO, null));
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> response = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (response == null || !(response.get("table") instanceof Map<?, ?> table)) {
+ return;
+ }
+ Long messagesIn = nonnegativeLong(table.get("msgPutTotalTodayNow"));
+ if (messagesIn != null) {
+ RocketMQCollect.this.setData(MAPPER.messagesInTotal(messagesIn));
+ }
+ Long bytesIn = nonnegativeLong(table.get("putMessageSizeTotal"));
+ if (bytesIn != null) {
+ RocketMQCollect.this.setData(MAPPER.throughputInTotal(bytesIn));
+ }
+ Long messagesOut = nonnegativeLong(table.get("msgGetTotalTodayNow"));
+ if (messagesOut != null) {
+ RocketMQCollect.this.setData(MAPPER.messagesOutTotal(messagesOut));
+ }
+ Long dispatch = nonnegativeLong(table.get("dispatchBehindBytes"));
+ if (dispatch != null) {
+ RocketMQCollect.this.setData(MAPPER.dispatchBytes(dispatch));
+ }
+ this.collectFlush(table);
+ this.collectDisk(table);
+ Long earliest = nonnegativeLong(table.get("earliestMessageTimeStamp"));
+ long now = System.currentTimeMillis();
+ if (earliest != null && earliest > 0 && earliest <= now) {
+ RocketMQCollect.this.setData(MAPPER.reserveTime(now - earliest));
+ }
+ for (String pool : new String[] {"send", "pull", "litePull", "query", "ack"}) {
+ this.collectThreadPool(table, pool, pool + "ThreadPoolQueueSize");
+ }
+ this.collectThreadPool(table, "endTransaction", "EndTransactionQueueSize");
+ }
+
+ private void collectDisk(Map<?, ?> table) {
+ table.forEach((key, value) -> {
+ if (!(key instanceof String name)) {
+ return;
+ }
+ String type;
+ String path = "";
+ if (name.equals("commitLogDiskRatio")) {
+ type = "commitlog_min";
+ } else if (name.startsWith("commitLogDiskRatio_")) {
+ type = "commitlog";
+ path = name.substring("commitLogDiskRatio_".length());
+ } else if (name.equals("consumeQueueDiskRatio")) {
+ type = "consumequeue";
+ } else {
+ return;
+ }
+ Float ratio = nonnegativeFloat(value);
+ if (ratio != null && ratio <= 1) {
+ RocketMQCollect.this.setData(MAPPER.diskUsage(type, path, ratio));
+ }
+ });
+ Object capacity = table.get("commitLogDirCapacity");
+ if (capacity instanceof String text) {
+ Matcher matcher = DISK_CAPACITY.matcher(text.trim());
+ if (matcher.matches()) {
+ Long total = byteCount(matcher.group(1));
+ Long free = byteCount(matcher.group(2));
+ if (total != null && total > 0 && free != null && free <= total) {
+ RocketMQCollect.this.setData(MAPPER.diskFreeBytes(free));
+ }
+ }
+ }
+ }
+
+ private void collectFlush(Map<?, ?> table) {
+ Long flush = byteCount(table.get("remainHowManyDataToFlush"));
+ // 启用暂存池时，原生刷盘积压还包括待提交字节；字段缺席表示未启用该分支。
+ Long commit = table.containsKey("remainHowManyDataToCommit")
+ ? byteCount(table.get("remainHowManyDataToCommit")) : Long.valueOf(0L);
+ if (flush != null && commit != null && flush <= Long.MAX_VALUE - commit) {
+ RocketMQCollect.this.setData(MAPPER.flushBytes(flush + commit));
+ }
+ }
+
+ private void collectThreadPool(Map<?, ?> table, String pool, String key) {
+ Long size = nonnegativeLong(table.get(key));
+ if (size != null) {
+ RocketMQCollect.this.setData(MAPPER.threadPool(pool, size));
+ }
+ }
+ }
+
+ private Long byteCount(Object value) {
+ if (!(value instanceof String text)) {
+ return null;
+ }
+ Matcher matcher = BYTE_COUNT.matcher(text.trim());
+ if (!matcher.matches()) {
+ return null;
+ }
+ String unit = matcher.group(2);
+ int exponent = "B".equals(unit) ? 0 : "KMGTPE".indexOf(unit.charAt(0)) + 1;
+ try {
+ // Broker 已将二进制字节数格式化并舍入，此处只能还原近似值，不能声称字节级精确。
+ return new BigDecimal(matcher.group(1)).multiply(BigDecimal.valueOf(1024).pow(exponent))
+ .setScale(0, RoundingMode.HALF_UP).longValueExact();
+ } catch (ArithmeticException e) {
+ return null;
+ }
+ }
+ }
+
+ private Long nonnegativeLong(Object value) {
+ if (value == null) {
+ return null;
+ }
+ try {
+ long number = Long.parseLong(value.toString());
+ return number >= 0 ? number : null;
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ private Float nonnegativeFloatOrZero(Object value) {
+ Float result = this.nonnegativeFloat(value);
+ return result == null ? 0f : result;
+ }
+
+ private Float nonnegativeFloat(Object value) {
+ if (value == null) {
+ return null;
+ }
+ try {
+ double number = Double.parseDouble(value.toString());
+ float result = (float) number;
+ return number >= 0 && Float.isFinite(result) ? result : null;
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ /** 解析 Broker、Topic 和消费组的分钟、小时、日消息统计窗口。 */
+ private abstract class WindowStatsCallback extends CollectCallback {
+ private WindowStatsCallback(RemotingCommand request) {
+ super(request);
+ }
+
+ @Override
+ protected final void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> stats = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ this.collectWindow(stats, "statsMinute", "minute");
+ this.collectWindow(stats, "statsHour", "hour");
+ this.collectWindow(stats, "statsDay", "day");
+ }
+
+ private void collectWindow(Map<?, ?> stats, String key, String window) {
+ if (stats == null || !(stats.get(key) instanceof Map<?, ?> sample)
+ || !(sample.get("sum") instanceof Number count) || !(sample.get("tps") instanceof Number rate)) {
+ return;
+ }
+ try {
+ long value = Long.parseLong(count.toString());
+ float tps = rate.floatValue();
+ if (value >= 0 && Float.isFinite(tps) && rate.doubleValue() >= 0) {
+ this.recordWindow(window, value, tps);
+ }
+ } catch (NumberFormatException e) {
+ // 小数、溢出或格式无效的消息数不参与采集。
+ }
+ }
+
+ protected abstract void recordWindow(String window, long count, float rate);
+ }
+
+ /** 统一处理请求发送、响应校验、异常和完成计数，具体响应由各采集回调解析。 */
+ private abstract class CollectCallback implements InvokeCallback {
+ private final RemotingCommand request;
+ private final AtomicBoolean finished = new AtomicBoolean();
+
+ private CollectCallback(RemotingCommand request) {
+ this.request = request;
+ // 请求发送前 +1；子请求先计数，父请求处理结束后才 -1。
+ CollectRound.this.pending.incrementAndGet();
+ }
+
+ // 发送异常也进入失败回调，每个请求只扣减一次。
+ protected final void execute() {
+ executeSafely(this,
+ () -> client.invokeAsync(this.request, this.timeoutMillis(), this));
+ }
+
+ private long timeoutMillis() throws TimeoutException {
+ long remaining = CollectRound.this.deadline - System.currentTimeMillis();
+ if (CollectRound.this.closed || remaining <= 0) {
+ throw new TimeoutException("RocketMQ collection deadline reached");
+ }
+ return Math.min(3000, remaining);
+ }
+
+ @Override
+ public final void operationComplete(ResponseFuture responseFuture) {
+ if (!this.finished.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ if (CollectRound.this.closed) {
+ return;
+ }
+ RemotingCommand response = responseFuture.getResponseCommand();
+ if (response == null) {
+ throw new IllegalStateException("Missing RocketMQ response");
+ }
+ if (response.getCode() != ResponseCode.SUCCESS || response.getBody() == null || response.getBody().length == 0) {
+ return;
+ }
+ this.handleResponse(responseFuture);
+ } catch (Exception e) {
+ this.restoreInterrupt(e);
Review Comment:
已将采集缓存改为 ConcurrentHashMap 和线程安全的 List，保留原有异步回调方式
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "SPUERSAIYAN (via GitHub)" <gi...@apache.org>.
SPUERSAIYAN commented on code in PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232#discussion_r4238159685
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java:
##########
@@ -19,15 +19,730 @@
import org.apache.eventmesh.dashboard.common.annotation.ClusterTypeMark;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
+import org.apache.eventmesh.dashboard.common.model.metadata.RuntimeMetadata;
import org.apache.eventmesh.dashboard.console.function.report.collect.AbstractCollect;
+import org.apache.eventmesh.dashboard.console.mapstruct.report.RocketMQCollectMapper;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKTypeEnum;
+import org.apache.eventmesh.dashboard.core.function.SDK.operation.rocketmq.RocketMQRemotingSDKOperation.DefaultRemotingClient;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.netty.ResponseFuture;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetTopicStatsInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryConsumeTimeSpanRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryTopicConsumeByWhoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ViewBrokerStatsDataRequestHeader;
+
+import java.io.StringReader;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import lombok.Setter;
@ClusterTypeMark(clusterType = {ClusterType.STORAGE_ROCKETMQ_CLUSTER, ClusterType.STORAGE_ROCKETMQ_BROKER,
ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE, ClusterType.STORAGE_ROCKETMQ_BROKER_RAFT})
public class RocketMQCollect extends AbstractCollect {
+ private static final RocketMQCollectMapper MAPPER = RocketMQCollectMapper.INSTANCE;
+ private static final Pattern BYTE_COUNT = Pattern.compile("([0-9]+(?:\\.[0-9]+)?)\\s*(B|[KMGTPE]iB)");
+
+ private static final Pattern DISK_CAPACITY = Pattern.compile("Total : (.+), Free : (.+)\\.");
+
+ private String runtimeUnique;
+
+ @Setter
+ private DefaultRemotingClient defaultRemotingClient;
+
+ @Override
+ public void setRuntimeMetadata(RuntimeMetadata runtimeMetadata) {
+ super.setRuntimeMetadata(runtimeMetadata);
+ this.runtimeUnique = runtimeMetadata == null ? null : runtimeMetadata.getUnique();
+ }
@Override
protected void doCollect() {
+ if (this.runtimeUnique == null) {
+ return;
+ }
+ this.defaultRemotingClient = SDKManage.getInstance().getClient(SDKTypeEnum.ADMIN, this.runtimeUnique);
+ new CollectRound().executeAndAwait();
+ }
+
+ /**
+ * 一轮采集流程：
+ * 1. 先注册 Topic 和 Broker 两个方向的根请求，再分别启动，统一管理请求计数。
+ * 2. TopicCollect 采集 Topic 总数、位点、积压时间、消息统计及客户端处理指标；死信 Topic 不查询消费组。
+ * 3. BrokerCollect 采集配置消费组总数、消息统计和运行状态；响应直接调用父类 setData。
+ * 4. 等待所有异步请求完成后返回，父类再统一交付本轮数据。
+ * 5. 超时或中断时保留已经采集的数据，关闭本轮，拒绝尚未开始处理的迟到回调。
+ */
+ private class CollectRound {
+ private final DefaultRemotingClient client = defaultRemotingClient;
+ private final AtomicInteger pending = new AtomicInteger();
+ private final CountDownLatch completed = new CountDownLatch(1);
+ private final long deadline = System.currentTimeMillis() + 4000;
+ private volatile boolean closed;
+
+ private void executeAndAwait() {
+ try {
+ // 先注册所有根请求，避免同步回调在后续请求注册前让计数归零。
+ TopicCollect topics = new TopicCollect();
+ BrokerCollect broker = new BrokerCollect();
+ topics.start();
+ broker.start();
+ // 所有请求完成时计数归零并唤醒；超时则保留已采集的数据。
+ this.completed.await(Math.max(1, this.deadline - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ this.close();
+ }
+ }
+
+ private void close() {
+ this.closed = true;
+ }
Review Comment:
已在 723685f 中补充每次写入前的关闭检查：所有采样写入统一经过 `CollectRound.setData()`，发现本轮 `closed` 后直接返回，再由原父类方法执行正常写入。API 发送前原有的 `timeoutMillis()` 关闭检查保持不变。
补充了“回调已进入解析 → 本轮超时 → 下一轮开始 → 旧回调恢复”的模拟回归测试，验证旧回调在写入前发现关闭后，不调用父类写入、不发送子请求，下一轮正常采集。该测试在修改前复现一次旧回调写入，修改后通过。完整采集测试共 30 项：28 项通过，2 项真实环境测试跳过；编译和 Checkstyle 通过。
本次采用简单状态检查，未增加锁、CAS 或缓存隔离，因此仍存在“检查通过后才发生关闭”的并发间隙，尚未完整解决评审指出的关闭与在途写入协调问题。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "SPUERSAIYAN (via GitHub)" <gi...@apache.org>.
SPUERSAIYAN commented on code in PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232#discussion_r4237838662
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/iotdb/IotDBReportMetaHandler.java:
##########
@@ -155,7 +156,7 @@ public String createTable() {
if (Objects.equals(filedName, "time")) {
stringBuilder.append(" time");
- } else if (filedName.endsWith("Id")) {
+ } else if (filedName.endsWith("Id") || field.isAnnotationPresent(ReportTag.class)) {
Review Comment:
新增标签已保留在采集模型中，查询与分组适配由报表模块后续再提交pr完成
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "SPUERSAIYAN (via GitHub)" <gi...@apache.org>.
SPUERSAIYAN commented on code in PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232#discussion_r4233107980
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/mapstruct/report/RocketMQCollectMapper.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.dashboard.console.mapstruct.report;
+
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqBrokerMessagesIn;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqBrokerMessagesOut;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerConnectionNumber;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerFailedTps;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerGroupNumber;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerLagLatency;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerOffset;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerProcessTime;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerSuccessTps;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqGroupMessagesOut;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqMessagesInTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqMessagesOutTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDiskFreeBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDiskUsage;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDispatchBehindBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageFlushBehindBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageMessageReserveTime;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqThreadPoolWartermark;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqThroughputInTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqTopicMessagesIn;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqTopicNumber;
+
+import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
+import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
+
+import org.mapstruct.BeanMapping;
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+import org.mapstruct.factory.Mappers;
+
+/** 将 Broker 采样转换为语义对应的报表模型，由模型注解指定目标表。 */
+@Mapper
+public interface RocketMQCollectMapper {
+ RocketMQCollectMapper INSTANCE = Mappers.getMapper(RocketMQCollectMapper.class);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "groupName", source = "group")
+ @Mapping(target = "valueConnectionCount", source = "count")
+ RocketmqConsumerConnectionNumber connections(String group, long count);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topic")
+ @Mapping(target = "queueId", source = "queue")
+ @Mapping(target = "value", source = "offset.maxOffset")
+ @Mapping(target = "valueMinOffset", source = "offset.minOffset")
+ @Mapping(target = "valueLastUpdateTime", source = "offset.lastUpdateTimestamp")
+ Rocketmq2ProducerOffset topicOffset(String topic, String queue, TopicOffset offset);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topic")
+ @Mapping(target = "queueId", constant = "")
+ @Mapping(target = "valueMinOffsetSum", source = "min")
+ @Mapping(target = "valueMaxOffsetSum", source = "max")
+ @Mapping(target = "valueLastUpdateTime", source = "lastUpdate")
+ Rocketmq2ProducerOffset aggregateOffset(String topic, long min, long max, long lastUpdate);
Review Comment:
数值 null 的异步写库处理会后续进行修复，我这边维持现有队列和汇总数据的含义
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "SPUERSAIYAN (via GitHub)" <gi...@apache.org>.
SPUERSAIYAN commented on code in PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232#discussion_r4238163408
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java:
##########
@@ -19,15 +19,730 @@
import org.apache.eventmesh.dashboard.common.annotation.ClusterTypeMark;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
+import org.apache.eventmesh.dashboard.common.model.metadata.RuntimeMetadata;
import org.apache.eventmesh.dashboard.console.function.report.collect.AbstractCollect;
+import org.apache.eventmesh.dashboard.console.mapstruct.report.RocketMQCollectMapper;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKTypeEnum;
+import org.apache.eventmesh.dashboard.core.function.SDK.operation.rocketmq.RocketMQRemotingSDKOperation.DefaultRemotingClient;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.netty.ResponseFuture;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetTopicStatsInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryConsumeTimeSpanRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryTopicConsumeByWhoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ViewBrokerStatsDataRequestHeader;
+
+import java.io.StringReader;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import lombok.Setter;
@ClusterTypeMark(clusterType = {ClusterType.STORAGE_ROCKETMQ_CLUSTER, ClusterType.STORAGE_ROCKETMQ_BROKER,
ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE, ClusterType.STORAGE_ROCKETMQ_BROKER_RAFT})
public class RocketMQCollect extends AbstractCollect {
+ private static final RocketMQCollectMapper MAPPER = RocketMQCollectMapper.INSTANCE;
+ private static final Pattern BYTE_COUNT = Pattern.compile("([0-9]+(?:\\.[0-9]+)?)\\s*(B|[KMGTPE]iB)");
+
+ private static final Pattern DISK_CAPACITY = Pattern.compile("Total : (.+), Free : (.+)\\.");
+
+ private String runtimeUnique;
+
+ @Setter
+ private DefaultRemotingClient defaultRemotingClient;
+
+ @Override
+ public void setRuntimeMetadata(RuntimeMetadata runtimeMetadata) {
+ super.setRuntimeMetadata(runtimeMetadata);
+ this.runtimeUnique = runtimeMetadata == null ? null : runtimeMetadata.getUnique();
+ }
@Override
protected void doCollect() {
+ if (this.runtimeUnique == null) {
+ return;
+ }
+ this.defaultRemotingClient = SDKManage.getInstance().getClient(SDKTypeEnum.ADMIN, this.runtimeUnique);
+ new CollectRound().executeAndAwait();
+ }
+
+ /**
+ * 一轮采集流程：
+ * 1. 先注册 Topic 和 Broker 两个方向的根请求，再分别启动，统一管理请求计数。
+ * 2. TopicCollect 采集 Topic 总数、位点、积压时间、消息统计及客户端处理指标；死信 Topic 不查询消费组。
+ * 3. BrokerCollect 采集配置消费组总数、消息统计和运行状态；响应直接调用父类 setData。
+ * 4. 等待所有异步请求完成后返回，父类再统一交付本轮数据。
+ * 5. 超时或中断时保留已经采集的数据，关闭本轮，拒绝尚未开始处理的迟到回调。
+ */
+ private class CollectRound {
+ private final DefaultRemotingClient client = defaultRemotingClient;
+ private final AtomicInteger pending = new AtomicInteger();
+ private final CountDownLatch completed = new CountDownLatch(1);
+ private final long deadline = System.currentTimeMillis() + 4000;
+ private volatile boolean closed;
+
+ private void executeAndAwait() {
+ try {
+ // 先注册所有根请求，避免同步回调在后续请求注册前让计数归零。
+ TopicCollect topics = new TopicCollect();
+ BrokerCollect broker = new BrokerCollect();
+ topics.start();
+ broker.start();
+ // 所有请求完成时计数归零并唤醒；超时则保留已采集的数据。
+ this.completed.await(Math.max(1, this.deadline - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ this.close();
+ }
+ }
+
+ private void close() {
+ this.closed = true;
+ }
Review Comment:
已在 723685f 中补充每次 setData() 前的 closed 检查，发现本轮已关闭就放弃写入，API 发送前保留原有关闭检查。超时情况下就不要进行数据交互了。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "SPUERSAIYAN (via GitHub)" <gi...@apache.org>.
SPUERSAIYAN commented on code in PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232#discussion_r4233070286
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/CollectManage.java:
##########
@@ -164,6 +164,7 @@ public Thread newThread(Runnable r) {
public void init() {
this.dataSyncHandler.setPaddingService(paddingService);
+ this.dataSyncHandler.setReportEngine(this.reportHandlerManage.getReportEngine());
Review Comment:
后续异步写库，会重新补充pr
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "SPUERSAIYAN (via GitHub)" <gi...@apache.org>.
SPUERSAIYAN commented on code in PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232#discussion_r4238159685
##########
eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java:
##########
@@ -19,15 +19,730 @@
import org.apache.eventmesh.dashboard.common.annotation.ClusterTypeMark;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
+import org.apache.eventmesh.dashboard.common.model.metadata.RuntimeMetadata;
import org.apache.eventmesh.dashboard.console.function.report.collect.AbstractCollect;
+import org.apache.eventmesh.dashboard.console.mapstruct.report.RocketMQCollectMapper;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKTypeEnum;
+import org.apache.eventmesh.dashboard.core.function.SDK.operation.rocketmq.RocketMQRemotingSDKOperation.DefaultRemotingClient;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.netty.ResponseFuture;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetTopicStatsInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryConsumeTimeSpanRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryTopicConsumeByWhoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ViewBrokerStatsDataRequestHeader;
+
+import java.io.StringReader;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import lombok.Setter;
@ClusterTypeMark(clusterType = {ClusterType.STORAGE_ROCKETMQ_CLUSTER, ClusterType.STORAGE_ROCKETMQ_BROKER,
ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE, ClusterType.STORAGE_ROCKETMQ_BROKER_RAFT})
public class RocketMQCollect extends AbstractCollect {
+ private static final RocketMQCollectMapper MAPPER = RocketMQCollectMapper.INSTANCE;
+ private static final Pattern BYTE_COUNT = Pattern.compile("([0-9]+(?:\\.[0-9]+)?)\\s*(B|[KMGTPE]iB)");
+
+ private static final Pattern DISK_CAPACITY = Pattern.compile("Total : (.+), Free : (.+)\\.");
+
+ private String runtimeUnique;
+
+ @Setter
+ private DefaultRemotingClient defaultRemotingClient;
+
+ @Override
+ public void setRuntimeMetadata(RuntimeMetadata runtimeMetadata) {
+ super.setRuntimeMetadata(runtimeMetadata);
+ this.runtimeUnique = runtimeMetadata == null ? null : runtimeMetadata.getUnique();
+ }
@Override
protected void doCollect() {
+ if (this.runtimeUnique == null) {
+ return;
+ }
+ this.defaultRemotingClient = SDKManage.getInstance().getClient(SDKTypeEnum.ADMIN, this.runtimeUnique);
+ new CollectRound().executeAndAwait();
+ }
+
+ /**
+ * 一轮采集流程：
+ * 1. 先注册 Topic 和 Broker 两个方向的根请求，再分别启动，统一管理请求计数。
+ * 2. TopicCollect 采集 Topic 总数、位点、积压时间、消息统计及客户端处理指标；死信 Topic 不查询消费组。
+ * 3. BrokerCollect 采集配置消费组总数、消息统计和运行状态；响应直接调用父类 setData。
+ * 4. 等待所有异步请求完成后返回，父类再统一交付本轮数据。
+ * 5. 超时或中断时保留已经采集的数据，关闭本轮，拒绝尚未开始处理的迟到回调。
+ */
+ private class CollectRound {
+ private final DefaultRemotingClient client = defaultRemotingClient;
+ private final AtomicInteger pending = new AtomicInteger();
+ private final CountDownLatch completed = new CountDownLatch(1);
+ private final long deadline = System.currentTimeMillis() + 4000;
+ private volatile boolean closed;
+
+ private void executeAndAwait() {
+ try {
+ // 先注册所有根请求，避免同步回调在后续请求注册前让计数归零。
+ TopicCollect topics = new TopicCollect();
+ BrokerCollect broker = new BrokerCollect();
+ topics.start();
+ broker.start();
+ // 所有请求完成时计数归零并唤醒；超时则保留已采集的数据。
+ this.completed.await(Math.max(1, this.deadline - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ this.close();
+ }
+ }
+
+ private void close() {
+ this.closed = true;
+ }
Review Comment:
已在 723685f 中补充每次写入前的关闭检查：所有采样写入统一经过 `CollectRound.setData()`，发现本轮 `closed` 后直接返回，再由原父类方法执行正常写入。API 发送前原有的 `timeoutMillis()` 关闭检查保持不变。
补充了“回调已进入解析 → 本轮超时 → 下一轮开始 → 旧回调恢复”的模拟回归测试，验证旧回调在写入前发现关闭后，不调用父类写入、不发送子请求，下一轮正常采集。该测试在修改前复现一次旧回调写入，修改后通过。完整采集测试共 30 项：28 项通过，2 项真实环境测试跳过；编译和 Checkstyle 通过。
本次采用简单状态检查，未增加锁、CAS 或缓存隔离，因此仍存在“检查通过后才发生关闭”的并发间隙，尚未完整解决评审指出的关闭与在途写入协调问题。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] feat: 采集 RocketMQ [eventmesh-dashboard]
Posted by "Pil0tXia (via GitHub)" <gi...@apache.org>.
Pil0tXia merged PR #232:
URL: https://github.com/apache/eventmesh-dashboard/pull/232
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org