Files
dbx/apps/desktop/src/lib/mq/rocketmqTraceUtils.ts
T
2f9ca313cb fix(rocketmq): 优化消费组加载/连接并修复积压、FIFO 与 Docker remap
* feat(rocketmq): 优化 RocketMQ 消费者组管理和性能监控

- 添加并行订阅组配置收集功能,支持超时控制和并发限制
- 实现消费者滞后监控功能,支持按主题过滤的滞后统计
- 重构消费滞后结果构建逻辑,添加 Broker 客户端映射功能
- 优化 Admin 客户端构建,确保请求超时参数正确传递
- 为 RabbitMQ 面板组件添加 MQ 写入保护确认机制
- 统一 MQ 面板样式表引入,清理重复按钮样式定义
- 增强连接对话框中的生产环境保护逻辑,支持 MQ 类型范围
- 更新测试用例以验证新的消费者滞后计算和客户端映射功能
- 为 RocketMQ 订阅面板添加过滤计数功能和测试覆盖

* feat(rocketmq): 优化 RocketMQ 客户端连接和消费者组管理

- 移除硬编码的 5 秒收集超时,改用连接配置的请求超时
- 添加从第一个可用主节点收集消费者组配置的方法以提高速度
- 重构内部方法访问客户端 API 实现和超时配置
- 在主题过滤模式下保留一个主节点转储以恢复 FIFO 类型准确性
- 改进消费延迟计算,失败时不设置总延迟而是保持未定义
- 为消费者运行信息获取添加 Docker 地址映射支持
- 更新命名服务器探测逻辑以更好地处理高可用配置
- 增加消息队列保活机制的查询超时支持
- 在前端界面中显示部分积压加载失败的错误信息
- 添加卸载主题前的操作确认保护
- 改进空结果集的用户界面提示信息

* fix(rocketmq): 解决消费组延迟统计失败时显示为健康零延迟的问题

- 在 examineConsumeStatsRemapped 中添加成功计数和最后错误记录
- 添加 ensureConsumeStatsProbeSucceeded 方法确保没有主节点响应时抛出异常
- 修改 TopicConsumeDetail 接口使 delay 字段可选以表示探针失败
- 更新 UI 显示逻辑在延迟不可用时显示破折号而不是零
- 添加测试用例验证所有代理失败时的异常处理
- 在 Rust 代码中检查 totalLag 是否存在以避免将缺失值视为零
- 添加消息队列适配器的标识检查防止过期的保活连接被错误清理

* fix(rocketmq): 修复RocketMQ消费组详情加载和统计查询问题

- 为examineConsumeStats调用添加超时参数以支持Docker地址映射
- 更新ensureConsumeStatsProbeSucceeded方法以处理部分失败情况下的空合并结果
- 在消费者组对话框中添加序列号防止陈旧详情/配置响应覆盖新数据
- 修复消费延迟显示逻辑,将null延迟值显示为"-"
- 改进错误处理,确保仅在所有代理成功时才接受空偏移表作为有效结果

* fix(rocketmq): 修复消费者组配置收集和消息积压显示问题

- 修改 collectConsumerGroupConfigsForTopicFilter 方法以合并多个主节点的订阅配置
- 优化 Topic-filtered 列表的配置收集逻辑,确保 FIFO 组类型准确性
- 添加 totalLagFailed 标志以正确处理消息积压探测失败情况
- 更新前端组件以在探测失败时显示 "-" 而非 0
- 修复 Docker 环境下多代理地址映射冲突问题
- 添加 RFC1918 私有 IP 验证方法并改进主机可达性判断
- 为积压不可用状态添加数据模型支持

* fix(rocketmq): 修复RocketMQ消费者组配置收集中的FIFO组识别问题

- 重构订阅组配置合并逻辑,确保FIFO组不会被误识别为NORMAL组
- 添加mergeSubscriptionGroupConfigs方法处理重复和冲突的订阅组配置
- 修改collectConsumerGroupConfigsForTopicFilter方法使用新的合并策略
- 添加测试用例验证FIFO组优先级和后续组保持功能
- 修复中间代理返回重复转储时丢失FIFO-only组的问题

* refactor(rocketmq): 优化消费组统计和名称服务器连接逻辑

- 引入 MasterBrokerAddrPlan 类来管理重映射和冲突回退地址
- 将 examineConsumeStatsRemapped 重构为 examineConsumeStatsOnMasters 方法
- 实现对 Docker 环境中重映射地址的更好支持和故障转移处理
- 在连接前预先探测 NameServer 可达性以避免 JVM 启动超时
- 优化消费者组在线成员数显示,未知状态时显示为 '-'
- 添加相关单元测试验证地址解析和冲突处理逻辑

* fix(rocketmq): 修复Docker重映射碰撞和消费者组配置问题

- 修复订阅组配置查找逻辑,支持FIFO合并避免顺序消息被隐藏
- 添加超时参数到订阅组查询方法
- 实现Docker重映射碰撞检测和失败关闭机制
- 添加测试验证碰撞空合并失败关闭逻辑
- 修复RocketMQ跟踪详情对话框加载序列控制问题
- 更新消费延迟标签为积压(backlog)
- 添加连接缺失和只读写拒绝的国际化消息提示
- 修复CSS样式中按钮主题颜色应用问题
- 优化NameServer探测预算计算逻辑

* fix(rocketmq): 修复消费者组类型分类和成员计数显示问题

- 当订阅配置缺失时返回 UNKNOWN 类型而不是默认 NORMAL
- 在删除和更新消费者组时对所有主节点执行操作确保一致性
- 成员计数查询失败时保持字段为空避免显示错误的离线状态
- 添加 UNKNOWN 类型的国际化支持和UI样式适配
- 修改测试用例验证缺失配置时的类型分类逻辑

* style(mq): 移除MQ面板中的主按钮样式覆盖

- 删除了 .mq-btn-primary 和 .btn-primary 的填充颜色样式定义
- 保留了悬停效果样式以维持用户体验一致性
- 更新注释说明MQ Chrome在浅色主题中保持页面表面样式
- 避免与工具栏和表格操作按钮产生视觉冲突

* test(composables): 为 useMqMutationGuard 添加测试并更新实现

- 添加了测试用例验证MQ面板使用的getter连接ID解析
- 将 unref 替换为 toValue 以正确处理getter参数
- 更新类型导入使用 MaybeRefOrGetter 替代 MaybeRef
- 添加注释说明使用 toValue 的原因

* fix(rocketmq): 修复消费者连接检查和订阅组变更的容错机制

- 重构 examineConsumerConnectionInfoRemapped 方法以正确处理无主节点响应的情况
- 添加 requireConsumerConnectionProbeResult 方法确保连接探针结果的准确性
- 修改 mutateSubscriptionGroupOnMasters 方法实现全地址覆盖要求防止部分更新
- 新增 ensureSubscriptionGroupMutationSucceeded 和 shouldFailClosedOnCollisionPartialMutation
  方法增强订阅组变更的错误处理
- 更新 rocketmq_subscription_from_group 函数默认类型为 UNKNOWN 而非 NORMAL
- 添加多个单元测试验证连接探针和订阅组变更的各种异常场景

* test(rocketmq): 为消费者连接探测测试方法添加异常声明

- 在requireConsumerConnectionProbeResultFailsClosedWhenNoMasterAnswers测试方法中添加throws Exception声明
- 确保测试方法能够正确处理可能抛出的异常情况

* fix(i18n): MQ 生产确认 source 本地化并清理 fieldStatus 死键

---------

Co-authored-by: AiLi <aili@aili.com>
Co-authored-by: t8y2 <t8y2@users.noreply.github.com>
Co-authored-by: skyler <1156263951@qq.com>
2026-08-06 01:12:31 +08:00

197 lines
8.0 KiB
TypeScript

import { formatError } from "@/lib/backend/errorUtils";
import { formatRocketMqTimestamp } from "@/lib/mq/rocketmqMessageUtils";
/** RocketMQ CODE 17: trace topic has no route on NameServer (trace not enabled or wrong topic). */
export function isRocketMqTraceTopicRouteMissingError(error: unknown): boolean {
const message = formatError(error);
return /CODE:\s*17/i.test(message) && /not matched route info|No topic route info/i.test(message);
}
export function formatRocketMqTraceError(error: unknown, traceTopicMissingHint: string): string {
if (isRocketMqTraceTopicRouteMissingError(error)) {
return traceTopicMissingHint;
}
return formatError(error);
}
/** Content / field splitters from Apache RocketMQ TraceConstants. */
const CONTENT_SPLITTER = "\u0001";
const FIELD_SPLITTER = "\u0002";
export type RocketMqTraceType = "Pub" | "SubBefore" | "SubAfter" | "EndTransaction" | "Unknown";
/** Stable field keys mapped to mqTrace.field* i18n labels in the dialog. */
export type RocketMqTraceFieldKey = "regionId" | "group" | "topic" | "msgId" | "tags" | "keys" | "storeHost" | "clientHost" | "bodyLength" | "costTime" | "msgType" | "offsetMsgId" | "requestId" | "retryTimes" | "contextCode";
export interface RocketMqTraceField {
key: RocketMqTraceFieldKey;
value: string;
}
export interface RocketMqTraceRecord {
type: RocketMqTraceType;
timestamp?: number;
/** Parsed success flag when present (Pub / SubAfter). */
success?: boolean;
fields: RocketMqTraceField[];
raw: string;
}
function emptyDisplay(value: string | undefined): string {
if (value == null) return "-";
const trimmed = value.trim();
if (!trimmed || trimmed === "null") return "-";
return trimmed;
}
function parseTimestamp(raw: string | undefined): number | undefined {
if (!raw?.trim()) return undefined;
const numeric = Number(raw);
return Number.isFinite(numeric) ? numeric : undefined;
}
function parseSuccess(raw: string | undefined): boolean | undefined {
if (raw == null || !raw.trim()) return undefined;
const normalized = raw.trim().toLowerCase();
if (normalized === "true") return true;
if (normalized === "false") return false;
return undefined;
}
function formatCostTime(raw: string | undefined): string {
const value = emptyDisplay(raw);
if (value === "-") return value;
const numeric = Number(value);
if (!Number.isFinite(numeric)) return value;
return `${numeric} ms`;
}
function pushField(fields: RocketMqTraceField[], key: RocketMqTraceFieldKey, value: string | undefined): void {
const display = key === "costTime" ? formatCostTime(value) : emptyDisplay(value);
// Keep tags/keys/retry visible even when empty so Pub/Sub cards stay scannable.
if (display === "-" && key !== "tags" && key !== "keys" && key !== "retryTimes") return;
fields.push({ key, value: display });
}
/** Normalize rare "Type@fields" prefix into official Type\\u0001fields form. */
function normalizeTraceSegment(segment: string): string {
return segment.replace(/^(Pub|SubBefore|SubAfter|EndTransaction)@/, `$1${CONTENT_SPLITTER}`);
}
function parsePub(parts: string[], raw: string): RocketMqTraceRecord {
// Pub: type, ts, region, group, topic, msgId, tags, keys, storeHost, bodyLength, cost, msgType, offsetMsgId, success[, clientHost]
const timestamp = parseTimestamp(parts[1]);
const success = parseSuccess(parts[13]);
const fields: RocketMqTraceField[] = [];
pushField(fields, "regionId", parts[2]);
pushField(fields, "group", parts[3]);
pushField(fields, "topic", parts[4]);
pushField(fields, "msgId", parts[5]);
pushField(fields, "tags", parts[6]);
pushField(fields, "keys", parts[7]);
pushField(fields, "storeHost", parts[8]);
pushField(fields, "bodyLength", parts[9]);
pushField(fields, "costTime", parts[10]);
pushField(fields, "msgType", parts[11]);
pushField(fields, "offsetMsgId", parts[12]);
if (parts.length >= 15) pushField(fields, "clientHost", parts[14]);
return { type: "Pub", timestamp, success, fields, raw };
}
function parseSubBefore(parts: string[], raw: string): RocketMqTraceRecord {
// SubBefore: type, ts, region, group, requestId, msgId, retryTimes, keys[, clientHost]
// Some encoders also append storeHost before clientHost (length >= 9/10).
const timestamp = parseTimestamp(parts[1]);
const fields: RocketMqTraceField[] = [];
pushField(fields, "regionId", parts[2]);
pushField(fields, "group", parts[3]);
pushField(fields, "requestId", parts[4]);
pushField(fields, "msgId", parts[5]);
pushField(fields, "retryTimes", parts[6]);
pushField(fields, "keys", parts[7]);
if (parts.length >= 10) {
pushField(fields, "storeHost", parts[8]);
pushField(fields, "clientHost", parts[9]);
} else if (parts.length >= 9) {
pushField(fields, "clientHost", parts[8]);
}
return { type: "SubBefore", timestamp, fields, raw };
}
function parseSubAfter(parts: string[], raw: string): RocketMqTraceRecord {
// SubAfter: type, requestId, msgId, cost, success, keys, contextCode[, timestamp, groupName]
const success = parseSuccess(parts[4]);
const timestamp = parts.length >= 8 ? parseTimestamp(parts[7]) : undefined;
const fields: RocketMqTraceField[] = [];
pushField(fields, "requestId", parts[1]);
pushField(fields, "msgId", parts[2]);
pushField(fields, "costTime", parts[3]);
pushField(fields, "keys", parts[5]);
pushField(fields, "contextCode", parts[6]);
if (parts.length >= 9) pushField(fields, "group", parts[8]);
return { type: "SubAfter", timestamp, success, fields, raw };
}
function parseEndTransaction(parts: string[], raw: string): RocketMqTraceRecord {
// EndTransaction: type, ts, region, group, topic, msgId, tags, keys, storeHost, msgType, clientHost, ...
const timestamp = parseTimestamp(parts[1]);
const fields: RocketMqTraceField[] = [];
pushField(fields, "regionId", parts[2]);
pushField(fields, "group", parts[3]);
pushField(fields, "topic", parts[4]);
pushField(fields, "msgId", parts[5]);
pushField(fields, "tags", parts[6]);
pushField(fields, "keys", parts[7]);
pushField(fields, "storeHost", parts[8]);
pushField(fields, "msgType", parts[9]);
if (parts.length >= 11) pushField(fields, "clientHost", parts[10]);
return { type: "EndTransaction", timestamp, fields, raw };
}
function parseTraceSegment(segment: string): RocketMqTraceRecord {
const raw = segment;
const normalized = normalizeTraceSegment(segment.trim());
if (!normalized) {
return { type: "Unknown", fields: [], raw };
}
const parts = normalized.split(CONTENT_SPLITTER);
const type = parts[0]?.trim();
if (type === "Pub" && parts.length >= 14) return parsePub(parts, raw);
if (type === "SubBefore" && parts.length >= 8) return parseSubBefore(parts, raw);
if (type === "SubAfter" && parts.length >= 7) return parseSubAfter(parts, raw);
if (type === "EndTransaction" && parts.length >= 9) return parseEndTransaction(parts, raw);
return { type: "Unknown", fields: [], raw };
}
/**
* Parse RocketMQ trace message body into structured records.
* Official format: Type\\u0001field...\\u0002Type\\u0001...
*/
export function parseRocketMqTracePayload(payload: string): RocketMqTraceRecord[] {
if (!payload?.trim()) return [];
const segments = payload
.split(FIELD_SPLITTER)
.map((part) => part.trim())
.filter(Boolean);
// Body may be a single record without trailing STX.
const units = segments.length ? segments : [payload.trim()];
return units.map(parseTraceSegment);
}
export function isParsedRocketMqTraceRecord(record: RocketMqTraceRecord): boolean {
return record.type !== "Unknown" && record.fields.length > 0;
}
/** Split KEYS header / keys field into display chips. */
export function splitRocketMqTraceKeys(value: string | undefined): string[] {
if (!value?.trim() || value.trim() === "-" || value.trim() === "null") return [];
return value
.split(/[\s,]+/)
.map((part) => part.trim())
.filter(Boolean);
}
export function rocketMqTraceRecordTimeLabel(record: RocketMqTraceRecord, fallbackTimestamp?: number): string {
return formatRocketMqTimestamp(record.timestamp ?? fallbackTimestamp);
}