This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 8ba7a872e feat(clients): add client connection diagnostics (#3028)
8ba7a872e is described below
commit 8ba7a872e4025c8712138f943f00556341424ded
Author: coder999o <[email protected]>
AuthorDate: Fri Sep 4 15:00:50 2026 +0800
feat(clients): add client connection diagnostics (#3028)
---
web/src/i18n/translations.ts | 25 +
.../pages/cluster/__tests__/ClientsPage.test.tsx | 51 +-
web/src/pages/cluster/clients.tsx | 296 ++++++++++
web/src/utils/clientConnectionDiagnostics.test.ts | 208 +++++++
web/src/utils/clientConnectionDiagnostics.ts | 607 +++++++++++++++++++++
5 files changed, 1184 insertions(+), 3 deletions(-)
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index b010c0dd2..0352c2f46 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -248,6 +248,31 @@ const translations: Record<string, Record<Lang, string>> =
{
'clients.allClusters': { zh: '全部集群', en: 'All Clusters' },
'clients.searchPlaceholder': { zh: '搜索 Client ID 或地址', en: 'Search Client ID
or address' },
'clients.detailTitle': { zh: '客户端详情 - {id}', en: 'Client Detail - {id}' },
+ 'clients.diagnostics': { zh: '客户端连接诊断', en: 'Client Connection Diagnostics'
},
+ 'clients.diagnosticHealthyMessage': {
+ zh: '未发现客户端连接风险',
+ en: 'No client connection risk found',
+ },
+ 'clients.diagnosticIssues': { zh: '风险项', en: 'Risks' },
+ 'clients.diagnosticRecommendations': { zh: '处理建议', en: 'Recommendations' },
+ 'clients.diagnosticResources': { zh: '资源数', en: 'Resources' },
+ 'clients.diagnosticMixedProtocols': { zh: '协议混用', en: 'Mixed Protocols' },
+ 'clients.diagnosticVersionSkews': { zh: '版本漂移', en: 'Version Skews' },
+ 'clients.diagnosticSingleConsumers': { zh: '单实例 Group', en: 'Single Groups'
},
+ 'clients.diagnosticUniqueClients': { zh: '唯一 Client', en: 'Unique Clients' },
+ 'clients.diagnosticUniqueAddresses': { zh: '唯一地址', en: 'Unique Addresses' },
+ 'clients.diagnosticClients': { zh: '客户端实例', en: 'Client Instances' },
+ 'clients.diagnosticClientUnit': { zh: '客户端', en: 'clients' },
+ 'clients.diagnosticAddressUnit': { zh: '地址', en: 'addresses' },
+ 'clients.diagnosticSeverity': { zh: '级别', en: 'Severity' },
+ 'clients.diagnosticIssue': { zh: '风险项', en: 'Risk' },
+ 'clients.diagnosticEvidence': { zh: '依据', en: 'Evidence' },
+ 'clients.healthStatusHealthy': { zh: '健康', en: 'Healthy' },
+ 'clients.healthStatusWarning': { zh: '关注', en: 'Warning' },
+ 'clients.healthStatusCritical': { zh: '高风险', en: 'Critical' },
+ 'clients.issueSeverityCritical': { zh: '高风险', en: 'Critical' },
+ 'clients.issueSeverityWarning': { zh: '关注', en: 'Warning' },
+ 'clients.issueSeverityInfo': { zh: '提示', en: 'Info' },
// ─── Alert Rules ───
'alerts.title': { zh: '集群告警', en: 'Cluster Alerts' },
diff --git a/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
b/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
index e8b5bae57..277762e4e 100644
--- a/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
+++ b/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
@@ -198,6 +198,53 @@ describe('Clients page', () => {
expect(within(languageVersions).getByText('C++ 4.9.8:
1')).toBeInTheDocument();
});
+ it('renders client connection diagnostics for risky inventories', async ()
=> {
+ vi.mocked(connectionsService.listConnections).mockResolvedValue([
+ {
+ ...connection,
+ clientId: 'shared-client',
+ groupOrTopic: 'order-create',
+ address: '10.0.1.12:49152',
+ },
+ {
+ ...connection,
+ clientId: 'shared-client',
+ groupOrTopic: 'order-create',
+ address: '10.0.1.13:49152',
+ },
+ {
+ ...connection,
+ clientId: 'consumer-a',
+ type: 'Consumer',
+ groupOrTopic: 'cg-order',
+ address: '10.0.2.10:49152',
+ protocol: 'gRPC',
+ version: '5.0.7',
+ },
+ {
+ ...connection,
+ clientId: 'consumer-b',
+ type: 'Consumer',
+ groupOrTopic: 'cg-order',
+ address: '10.0.2.11:49152',
+ protocol: 'Remoting',
+ version: '4.9.8',
+ },
+ ]);
+ renderWithProviders(<ClientsPage />);
+
+ const diagnostics = await
screen.findByTestId('client-connection-diagnostics');
+ expect(within(diagnostics).getByText('客户端连接诊断')).toBeInTheDocument();
+ expect(within(diagnostics).getByText('客户端连接存在高风险')).toBeInTheDocument();
+ expect(within(diagnostics).getByText('Client ID
连接到多个地址')).toBeInTheDocument();
+ expect(within(diagnostics).getByText('同一资源存在多协议连接')).toBeInTheDocument();
+ expect(
+ within(diagnostics).getByText(
+ '确认该资源是否处于协议迁移期，并分别检查 Proxy 与 Broker 侧连接状态。',
+ ),
+ ).toBeInTheDocument();
+ });
+
it('updates statistics when the selected cluster filter changes', async ()
=> {
const user = userEvent.setup();
vi.mocked(connectionsService.listConnections).mockResolvedValue(connections);
@@ -447,9 +494,7 @@ describe('Clients page', () => {
expect(await screen.findByText('Unable to load registry
clusters')).toBeInTheDocument();
await user.click(screen.getByRole('button', { name: /重\s*试/ }));
- await waitFor(() =>
- expect(clusterService.listRegistryClusters).toHaveBeenNthCalledWith(2),
- );
+ await waitFor(() =>
expect(clusterService.listRegistryClusters).toHaveBeenNthCalledWith(2));
await act(async () => {
stale.resolve([]);
diff --git a/web/src/pages/cluster/clients.tsx
b/web/src/pages/cluster/clients.tsx
index d696550de..98eefcea0 100644
--- a/web/src/pages/cluster/clients.tsx
+++ b/web/src/pages/cluster/clients.tsx
@@ -24,6 +24,7 @@ import {
Flex,
Input,
Modal,
+ Progress,
Select,
Space,
Statistic,
@@ -45,6 +46,11 @@ import type { ClusterInfo } from '../../api/cluster';
import { formatDateTime } from '../../utils/format';
import { buildCsv, downloadCsv, type CsvColumn } from '../../utils/download';
import { tableScrollX } from '../../utils/table';
+import {
+ analyzeClientConnections,
+ type ClientConnectionIssue,
+ type ClientResourceSummary,
+} from '../../utils/clientConnectionDiagnostics';
const { Text } = Typography;
const DEFAULT_LOAD_ERROR = '客户端连接加载失败，请稍后重试';
@@ -61,6 +67,30 @@ const protocolConfig: Record<string, { color: string; label:
string }> = {
Remoting: { color: 'blue', label: 'Remoting' },
};
+const healthStatusColor: Record<ClientResourceSummary['status'], string> = {
+ healthy: 'green',
+ warning: 'gold',
+ critical: 'red',
+};
+
+const healthStatusTextKey: Record<ClientResourceSummary['status'], string> = {
+ healthy: 'clients.healthStatusHealthy',
+ warning: 'clients.healthStatusWarning',
+ critical: 'clients.healthStatusCritical',
+};
+
+const issueSeverityColor: Record<ClientConnectionIssue['severity'], string> = {
+ critical: 'red',
+ warning: 'gold',
+ info: 'blue',
+};
+
+const issueSeverityTextKey: Record<ClientConnectionIssue['severity'], string>
= {
+ critical: 'clients.issueSeverityCritical',
+ warning: 'clients.issueSeverityWarning',
+ info: 'clients.issueSeverityInfo',
+};
+
const languageConfig: Record<string, { color: string; label: string }> = {
Java: { color: 'default', label: 'Java' },
Go: { color: 'cyan', label: 'Go' },
@@ -265,6 +295,48 @@ const ClientsPage = () => {
};
}, [clusterConnections]);
+ const clientDiagnostics = useMemo(
+ () => analyzeClientConnections(clusterConnections),
+ [clusterConnections],
+ );
+
+ const diagnosticProgressStatus =
+ clientDiagnostics.status === 'critical'
+ ? 'exception'
+ : clientDiagnostics.status === 'healthy'
+ ? 'success'
+ : 'normal';
+
+ const diagnosticStrokeColor =
+ clientDiagnostics.status === 'critical'
+ ? '#ff4d4f'
+ : clientDiagnostics.status === 'warning'
+ ? '#faad14'
+ : '#52c41a';
+
+ const diagnosticSummaryItems = [
+ {
+ key: 'resources',
+ title: t('clients.diagnosticResources'),
+ value: clientDiagnostics.summary.resourceCount,
+ },
+ {
+ key: 'mixedProtocol',
+ title: t('clients.diagnosticMixedProtocols'),
+ value: clientDiagnostics.summary.mixedProtocolResourceCount,
+ },
+ {
+ key: 'mixedVersion',
+ title: t('clients.diagnosticVersionSkews'),
+ value: clientDiagnostics.summary.mixedVersionResourceCount,
+ },
+ {
+ key: 'singleConsumer',
+ title: t('clients.diagnosticSingleConsumers'),
+ value: clientDiagnostics.summary.singleConsumerGroupCount,
+ },
+ ];
+
/* ─── Filtered data (search + cluster only, table handles column filters)
─── */
const filtered = useMemo(() => {
const normalizedSearch = search.toLowerCase();
@@ -434,6 +506,146 @@ const ClientsPage = () => {
},
];
+ const resourceColumns: ColumnsType<ClientResourceSummary> = [
+ {
+ title: t('common.status'),
+ dataIndex: 'status',
+ key: 'status',
+ width: 90,
+ render: (status: ClientResourceSummary['status']) => (
+ <Tag
color={healthStatusColor[status]}>{t(healthStatusTextKey[status])}</Tag>
+ ),
+ },
+ {
+ title: t('clients.groupOrTopic'),
+ key: 'resource',
+ width: 220,
+ render: (_: unknown, record) => (
+ <Space direction="vertical" size={2}>
+ <Text strong>{record.resource}</Text>
+ <Text type="secondary">{record.type}</Text>
+ </Space>
+ ),
+ },
+ {
+ title: t('clients.diagnosticClients'),
+ key: 'clients',
+ width: 170,
+ render: (_: unknown, record) => (
+ <Space size={4} wrap>
+ <Tag>
+ {record.uniqueClientCount} {t('clients.diagnosticClientUnit')}
+ </Tag>
+ <Tag>
+ {record.uniqueAddressCount} {t('clients.diagnosticAddressUnit')}
+ </Tag>
+ </Space>
+ ),
+ },
+ {
+ title: t('clients.protocol'),
+ dataIndex: 'protocols',
+ key: 'protocols',
+ width: 160,
+ render: (protocols: string[]) => (
+ <Space size={4} wrap>
+ {protocols.map((protocol) => (
+ <Tag key={protocol} color={protocolConfig[protocol]?.color ??
'default'}>
+ {protocol}
+ </Tag>
+ ))}
+ </Space>
+ ),
+ },
+ {
+ title: `${t('clients.language')} / ${t('common.version')}`,
+ key: 'versions',
+ width: 220,
+ render: (_: unknown, record) => (
+ <Space size={4} wrap>
+ {record.languages.map((language) => (
+ <Tag key={language} color={languageConfig[language]?.color ??
'default'}>
+ {languageConfig[language]?.label ?? language}
+ </Tag>
+ ))}
+ {record.versions.map((version) => (
+ <Text key={version} code>
+ {version}
+ </Text>
+ ))}
+ </Space>
+ ),
+ },
+ {
+ title: t('clients.diagnosticIssues'),
+ dataIndex: 'issueCount',
+ key: 'issueCount',
+ width: 90,
+ render: (count: number) => <Tag color={count > 0 ? 'gold' :
'green'}>{count}</Tag>,
+ },
+ ];
+
+ const issueColumns: ColumnsType<ClientConnectionIssue> = [
+ {
+ title: t('clients.diagnosticSeverity'),
+ dataIndex: 'severity',
+ key: 'severity',
+ width: 90,
+ render: (severity: ClientConnectionIssue['severity']) => (
+ <Tag
color={issueSeverityColor[severity]}>{t(issueSeverityTextKey[severity])}</Tag>
+ ),
+ },
+ {
+ title: t('clients.diagnosticIssue'),
+ key: 'issue',
+ width: 260,
+ render: (_: unknown, record) => (
+ <Space direction="vertical" size={2}>
+ <Text strong>{record.title}</Text>
+ <Text type="secondary">{record.description}</Text>
+ </Space>
+ ),
+ },
+ {
+ title: t('clients.clientId'),
+ dataIndex: 'clientId',
+ key: 'clientId',
+ width: 180,
+ render: (clientId?: string) =>
+ clientId ? (
+ <Text style={{ fontFamily: 'monospace' }}>{clientId}</Text>
+ ) : (
+ <Text type="secondary">-</Text>
+ ),
+ },
+ {
+ title: t('clients.groupOrTopic'),
+ dataIndex: 'resource',
+ key: 'resource',
+ width: 160,
+ render: (resource?: string) => resource || '-',
+ },
+ {
+ title: t('clients.diagnosticEvidence'),
+ dataIndex: 'evidence',
+ key: 'evidence',
+ width: 220,
+ render: (evidence: string[]) => (
+ <Space size={4} wrap>
+ {evidence.length === 0 ? (
+ <Text type="secondary">-</Text>
+ ) : (
+ evidence.map((item) => (
+ <Text key={item} code>
+ {item}
+ </Text>
+ ))
+ )}
+ </Space>
+ ),
+ },
+ ];
+
/* ═══════════════════════════════════════════
Render
═══════════════════════════════════════════ */
@@ -582,6 +794,90 @@ const ClientsPage = () => {
</div>
</Flex>
+ <div
+ data-testid="client-connection-diagnostics"
+ style={{
+ marginBottom: 16,
+ padding: '16px',
+ background: token.colorBgContainer,
+ border: `1px solid ${token.colorBorderSecondary}`,
+ borderRadius: token.borderRadiusLG,
+ }}
+ >
+ <Flex gap={20} align="center" wrap="wrap" style={{ marginBottom: 16 }}>
+ <Progress
+ type="circle"
+ percent={clientDiagnostics.score}
+ size={92}
+ status={diagnosticProgressStatus}
+ strokeColor={diagnosticStrokeColor}
+ format={(percent) => `${percent}`}
+ />
+ <div style={{ minWidth: 220, flex: '1 1 260px' }}>
+ <Typography.Title level={5} style={{ margin: 0 }}>
+ {t('clients.diagnostics')}
+ </Typography.Title>
+ <Text type="secondary">{clientDiagnostics.statusText}</Text>
+ <div style={{ marginTop: 8 }}>
+ <Tag color={clientDiagnostics.statusColor}>
+ {t('clients.diagnosticIssues')}:
{clientDiagnostics.issues.length}
+ </Tag>
+ <Tag>
+ {t('clients.diagnosticUniqueClients')}:{' '}
+ {clientDiagnostics.summary.uniqueClientCount}
+ </Tag>
+ <Tag>
+ {t('clients.diagnosticUniqueAddresses')}:{' '}
+ {clientDiagnostics.summary.uniqueAddressCount}
+ </Tag>
+ </div>
+ </div>
+ <Flex gap={16} wrap="wrap" style={{ flex: '2 1 440px' }}>
+ {diagnosticSummaryItems.map((item) => (
+ <div key={item.key} style={{ minWidth: 126 }}>
+ <Statistic title={item.title} value={item.value} valueStyle={{
fontSize: 22 }} />
+ </div>
+ ))}
+ </Flex>
+ </Flex>
+
+ <Table<ClientResourceSummary>
+ columns={resourceColumns}
+ dataSource={clientDiagnostics.resources}
+ rowKey="id"
+ pagination={false}
+ size="small"
+ scroll={{ x: tableScrollX(resourceColumns) }}
+ locale={{ emptyText: t('common.noData') }}
+ style={{ marginBottom: 12 }}
+ />
+
+ {clientDiagnostics.issues.length > 0 ? (
+ <Table<ClientConnectionIssue>
+ columns={issueColumns}
+ dataSource={clientDiagnostics.issues}
+ rowKey="id"
+ pagination={false}
+ size="small"
+ scroll={{ x: tableScrollX(issueColumns) }}
+ style={{ marginBottom: 12 }}
+ />
+ ) : (
+ <Alert type="success" showIcon
message={t('clients.diagnosticHealthyMessage')} />
+ )}
+
+ <div style={{ marginTop: 12 }}>
+ <Text strong>{t('clients.diagnosticRecommendations')}</Text>
+ <ul style={{ margin: '8px 0 0', paddingLeft: 20 }}>
+ {clientDiagnostics.recommendations.map((item) => (
+ <li key={item}>
+ <Text>{item}</Text>
+ </li>
+ ))}
+ </ul>
+ </div>
+ </div>
+
{/* ─── Table ─── */}
<Card styles={{ body: { padding: 0 } }}>
<Table
diff --git a/web/src/utils/clientConnectionDiagnostics.test.ts
b/web/src/utils/clientConnectionDiagnostics.test.ts
new file mode 100644
index 000000000..30221b91d
--- /dev/null
+++ b/web/src/utils/clientConnectionDiagnostics.test.ts
@@ -0,0 +1,208 @@
+/*
+ * 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.
+ */
+
+import { describe, expect, it } from 'vitest';
+import type { ClientConnection } from '../api/connections';
+import { analyzeClientConnections } from './clientConnectionDiagnostics';
+
+const connection = (overrides: Partial<ClientConnection>): ClientConnection =>
({
+ clientId: '[email protected]:49152',
+ type: 'Producer',
+ groupOrTopic: 'order-events',
+ protocol: 'gRPC',
+ address: '10.0.1.10:49152',
+ language: 'Java',
+ version: '5.0.7',
+ connectedAt: '2026-07-01 08:30:00',
+ clusterName: 'ns-prod',
+ ...overrides,
+});
+
+describe('client connection diagnostics', () => {
+ it('marks a consistent multi-client inventory as healthy', () => {
+ const diagnostics = analyzeClientConnections([
+ connection({ clientId: 'producer-a', address: '10.0.1.10:49152' }),
+ connection({ clientId: 'producer-b', address: '10.0.1.11:49152' }),
+ connection({
+ clientId: 'consumer-a',
+ type: 'Consumer',
+ groupOrTopic: 'cg-order',
+ address: '10.0.2.10:49152',
+ }),
+ connection({
+ clientId: 'consumer-b',
+ type: 'Consumer',
+ groupOrTopic: 'cg-order',
+ address: '10.0.2.11:49152',
+ }),
+ ]);
+
+ expect(diagnostics.status).toBe('healthy');
+ expect(diagnostics.score).toBe(100);
+ expect(diagnostics.summary).toMatchObject({
+ totalConnections: 4,
+ uniqueClientCount: 4,
+ uniqueAddressCount: 4,
+ resourceCount: 2,
+ partialConnectionCount: 0,
+ mixedProtocolResourceCount: 0,
+ mixedVersionResourceCount: 0,
+ singleConsumerGroupCount: 0,
+ });
+ expect(diagnostics.issues).toEqual([]);
+ expect(diagnostics.recommendations).toEqual([
+ '保持客户端连接清单按集群定期巡检，重点关注协议和 SDK 版本收敛。',
+ ]);
+ });
+
+ it('reports an empty client inventory as critical', () => {
+ const diagnostics = analyzeClientConnections([]);
+
+ expect(diagnostics.status).toBe('critical');
+ expect(diagnostics.summary.totalConnections).toBe(0);
+ expect(diagnostics.resources).toEqual([]);
+ expect(diagnostics.issues).toEqual([
+ expect.objectContaining({
+ code: 'NO_CONNECTIONS',
+ severity: 'critical',
+ }),
+ ]);
+ });
+
+ it('detects partial scans, client ID collisions, and duplicated records', ()
=> {
+ const diagnostics = analyzeClientConnections([
+ connection({
+ clientId: 'shared-client',
+ groupOrTopic: 'order-events',
+ address: '10.0.1.10:49152',
+ partial: true,
+ }),
+ connection({
+ clientId: 'shared-client',
+ groupOrTopic: 'order-events',
+ address: '10.0.1.11:49152',
+ }),
+ connection({
+ clientId: 'dup-client',
+ groupOrTopic: 'payment-events',
+ address: '10.0.1.12:49152',
+ }),
+ connection({
+ clientId: 'dup-client',
+ groupOrTopic: 'payment-events',
+ address: '10.0.1.12:49152',
+ }),
+ ]);
+
+ expect(diagnostics.status).toBe('critical');
+ expect(diagnostics.summary.partialConnectionCount).toBe(1);
+ expect(diagnostics.issues).toEqual(
+ expect.arrayContaining([
+ expect.objectContaining({
+ code: 'PARTIAL_CONNECTION_SCAN',
+ severity: 'warning',
+ evidence: ['partial=1'],
+ }),
+ expect.objectContaining({
+ code: 'CLIENT_ID_COLLISION',
+ severity: 'critical',
+ clientId: 'shared-client',
+ }),
+ expect.objectContaining({
+ code: 'EXACT_DUPLICATE_CONNECTION',
+ severity: 'info',
+ clientId: 'dup-client',
+ }),
+ ]),
+ );
+ });
+
+ it('flags mixed protocols, mixed versions, and single consumer instances per
resource', () => {
+ const diagnostics = analyzeClientConnections([
+ connection({
+ type: 'Consumer',
+ clientId: 'consumer-a',
+ groupOrTopic: 'cg-order',
+ protocol: 'gRPC',
+ language: 'Java',
+ version: '5.0.7',
+ }),
+ connection({
+ type: 'Consumer',
+ clientId: 'consumer-b',
+ groupOrTopic: 'cg-order',
+ protocol: 'Remoting',
+ language: 'Java',
+ version: '4.9.8',
+ address: '10.0.1.11:49152',
+ }),
+ connection({
+ type: 'Consumer',
+ clientId: 'consumer-single',
+ groupOrTopic: 'cg-payment',
+ address: '10.0.2.10:49152',
+ }),
+ ]);
+
+ expect(diagnostics.status).toBe('warning');
+ expect(diagnostics.summary).toMatchObject({
+ mixedProtocolResourceCount: 1,
+ mixedVersionResourceCount: 1,
+ singleConsumerGroupCount: 1,
+ });
+ expect(diagnostics.resources[0]).toMatchObject({
+ type: 'Consumer',
+ resource: 'cg-order',
+ status: 'warning',
+ issueCount: 2,
+ });
+ expect(diagnostics.issues.map((item) => item.code)).toEqual(
+ expect.arrayContaining([
+ 'MIXED_PROTOCOL_RESOURCE',
+ 'MIXED_VERSION_RESOURCE',
+ 'SINGLE_CONSUMER_INSTANCE',
+ ]),
+ );
+ });
+
+ it('detects unknown fields, invalid timestamps, and address concentration',
() => {
+ const diagnostics = analyzeClientConnections([
+ connection({
+ clientId: 'client-a',
+ protocol: 'Custom',
+ language: 'Ruby',
+ version: '-',
+ connectedAt: 'not-a-date',
+ }),
+ connection({ clientId: 'client-b' }),
+ connection({ clientId: 'client-c' }),
+ connection({ clientId: 'client-d' }),
+ ]);
+
+ expect(diagnostics.status).toBe('critical');
+ expect(diagnostics.summary.concentratedAddressCount).toBe(1);
+ expect(diagnostics.issues.map((item) => item.code)).toEqual(
+ expect.arrayContaining([
+ 'UNKNOWN_PROTOCOL',
+ 'UNKNOWN_LANGUAGE',
+ 'UNKNOWN_VERSION',
+ 'INVALID_CONNECTION_TIME',
+ 'ADDRESS_CONCENTRATION',
+ ]),
+ );
+ });
+});
diff --git a/web/src/utils/clientConnectionDiagnostics.ts
b/web/src/utils/clientConnectionDiagnostics.ts
new file mode 100644
index 000000000..640fc16eb
--- /dev/null
+++ b/web/src/utils/clientConnectionDiagnostics.ts
@@ -0,0 +1,607 @@
+/*
+ * 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.
+ */
+
+import type { ClientConnection } from '../api/connections';
+
+export type ClientConnectionHealthStatus = 'healthy' | 'warning' | 'critical';
+export type ClientConnectionIssueSeverity =
+ Exclude<ClientConnectionHealthStatus, 'healthy'> | 'info';
+
+export type ClientConnectionIssueCode =
+ | 'NO_CONNECTIONS'
+ | 'PARTIAL_CONNECTION_SCAN'
+ | 'CLIENT_ID_COLLISION'
+ | 'EXACT_DUPLICATE_CONNECTION'
+ | 'MIXED_PROTOCOL_RESOURCE'
+ | 'MIXED_VERSION_RESOURCE'
+ | 'SINGLE_CONSUMER_INSTANCE'
+ | 'ADDRESS_CONCENTRATION'
+ | 'UNKNOWN_PROTOCOL'
+ | 'UNKNOWN_LANGUAGE'
+ | 'UNKNOWN_VERSION'
+ | 'INVALID_CONNECTION_TIME';
+
+export interface ClientConnectionIssue {
+ id: string;
+ code: ClientConnectionIssueCode;
+ severity: ClientConnectionIssueSeverity;
+ title: string;
+ description: string;
+ resource?: string;
+ clientId?: string;
+ evidence: string[];
+ recommendation: string;
+}
+
+export interface ClientResourceSummary {
+ id: string;
+ type: string;
+ resource: string;
+ connectionCount: number;
+ uniqueClientCount: number;
+ uniqueAddressCount: number;
+ protocols: string[];
+ languages: string[];
+ versions: string[];
+ partial: boolean;
+ status: ClientConnectionHealthStatus;
+ issueCount: number;
+}
+
+export interface ClientConnectionHealthSummary {
+ totalConnections: number;
+ uniqueClientCount: number;
+ uniqueAddressCount: number;
+ resourceCount: number;
+ partialConnectionCount: number;
+ mixedProtocolResourceCount: number;
+ mixedVersionResourceCount: number;
+ singleConsumerGroupCount: number;
+ concentratedAddressCount: number;
+}
+
+export interface ClientConnectionDiagnostics {
+ status: ClientConnectionHealthStatus;
+ statusText: string;
+ statusColor: 'success' | 'warning' | 'error';
+ score: number;
+ summary: ClientConnectionHealthSummary;
+ resources: ClientResourceSummary[];
+ issues: ClientConnectionIssue[];
+ recommendations: string[];
+}
+
+type ConnectionGroup = {
+ type: string;
+ resource: string;
+ connections: ClientConnection[];
+};
+
+const STATUS_TEXT: Record<ClientConnectionHealthStatus, string> = {
+ healthy: '客户端连接健康',
+ warning: '客户端连接需要关注',
+ critical: '客户端连接存在高风险',
+};
+
+const STATUS_COLOR: Record<ClientConnectionHealthStatus, 'success' | 'warning'
| 'error'> = {
+ healthy: 'success',
+ warning: 'warning',
+ critical: 'error',
+};
+
+const KNOWN_PROTOCOLS = new Set(['gRPC', 'Remoting']);
+const KNOWN_LANGUAGES = new Set(['Java', 'Go', 'Python', 'Rust', 'Cpp',
'CSharp', 'NodeJS', 'PHP']);
+
+const normalizeText = (value?: string | null, fallback = 'unknown'): string =>
{
+ const trimmed = (value ?? '').trim();
+ return trimmed || fallback;
+};
+
+const uniqueSorted = (values: string[]): string[] =>
+ [...new Set(values.map((value) =>
normalizeText(value)).filter(Boolean))].sort((a, b) =>
+ a.localeCompare(b),
+ );
+
+const countBy = (values: string[]): Map<string, number> => {
+ const counts = new Map<string, number>();
+ values.forEach((value) => {
+ const normalized = normalizeText(value);
+ counts.set(normalized, (counts.get(normalized) ?? 0) + 1);
+ });
+ return counts;
+};
+
+const issue = (
+ code: ClientConnectionIssueCode,
+ severity: ClientConnectionIssueSeverity,
+ title: string,
+ description: string,
+ recommendation: string,
+ options: {
+ resource?: string;
+ clientId?: string;
+ evidence?: string[];
+ id?: string;
+ } = {},
+): ClientConnectionIssue => ({
+ id:
+ options.id ??
+ [options.resource, options.clientId, code, ...(options.evidence ?? [])]
+ .filter(Boolean)
+ .join(':'),
+ code,
+ severity,
+ title,
+ description,
+ resource: options.resource,
+ clientId: options.clientId,
+ evidence: options.evidence ?? [],
+ recommendation,
+});
+
+const connectionIdentity = (connection: ClientConnection): string =>
+ [
+ normalizeText(connection.type),
+ normalizeText(connection.clientId),
+ normalizeText(connection.groupOrTopic),
+ normalizeText(connection.address),
+ ].join('|');
+
+const resourceKey = (connection: ClientConnection): string =>
+
`${normalizeText(connection.type)}:${normalizeText(connection.groupOrTopic)}`;
+
+const groupConnections = (connections: ClientConnection[]): ConnectionGroup[]
=> {
+ const groups = new Map<string, ConnectionGroup>();
+
+ connections.forEach((connection) => {
+ const key = resourceKey(connection);
+ const group = groups.get(key);
+ if (group) {
+ group.connections.push(connection);
+ return;
+ }
+ groups.set(key, {
+ type: normalizeText(connection.type),
+ resource: normalizeText(connection.groupOrTopic),
+ connections: [connection],
+ });
+ });
+
+ return [...groups.values()].sort(
+ (left, right) =>
+ left.type.localeCompare(right.type) ||
left.resource.localeCompare(right.resource),
+ );
+};
+
+const parseTime = (value?: string | null): number | null => {
+ if (!value) return null;
+ const normalized = value.includes('T') ? value : value.replace(' ', 'T');
+ const timestamp = Date.parse(normalized);
+ return Number.isFinite(timestamp) ? timestamp : null;
+};
+
+const resourceSeverity = (issues: ClientConnectionIssue[]):
ClientConnectionHealthStatus => {
+ if (issues.some((item) => item.severity === 'critical')) return 'critical';
+ if (issues.some((item) => item.severity === 'warning')) return 'warning';
+ return 'healthy';
+};
+
+const addInventoryIssues = (connections: ClientConnection[], issues:
ClientConnectionIssue[]) => {
+ if (connections.length === 0) {
+ issues.push(
+ issue(
+ 'NO_CONNECTIONS',
+ 'critical',
+ '未发现客户端连接',
+ '当前 NameServer 查询没有返回任何 Producer 或 Consumer 连接。',
+ '确认目标 NameServer、Proxy 和 Broker 侧客户端注册链路是否正常。',
+ ),
+ );
+ return;
+ }
+
+ const partialCount = connections.filter((connection) =>
connection.partial).length;
+ if (partialCount > 0) {
+ issues.push(
+ issue(
+ 'PARTIAL_CONNECTION_SCAN',
+ 'warning',
+ '客户端扫描结果不完整',
+ '部分 Producer 连接来自受限 Topic 扫描，当前列表可能不是完整客户端清单。',
+ '缩小 Topic 或集群范围后重新查询，并在排障时避免把当前列表视为全集。',
+ { evidence: [`partial=${partialCount}`] },
+ ),
+ );
+ }
+};
+
+const addClientIdIssues = (connections: ClientConnection[], issues:
ClientConnectionIssue[]) => {
+ const connectionsByClient = new Map<string, ClientConnection[]>();
+ const identityCounts = new Map<string, number>();
+
+ connections.forEach((connection) => {
+ const clientId = normalizeText(connection.clientId);
+ const existing = connectionsByClient.get(clientId) ?? [];
+ existing.push(connection);
+ connectionsByClient.set(clientId, existing);
+
+ const identity = connectionIdentity(connection);
+ identityCounts.set(identity, (identityCounts.get(identity) ?? 0) + 1);
+ });
+
+ connectionsByClient.forEach((clientConnections, clientId) => {
+ const addresses = uniqueSorted(clientConnections.map((connection) =>
connection.address));
+ if (addresses.length > 1) {
+ issues.push(
+ issue(
+ 'CLIENT_ID_COLLISION',
+ 'critical',
+ 'Client ID 连接到多个地址',
+ '同一个 Client ID 同时出现在多个远端地址，可能是实例 ID 配置冲突或旧连接未及时清理。',
+ '检查客户端 instanceName/clientId 配置，确保同一进程实例使用唯一标识。',
+ {
+ clientId,
+ evidence: addresses,
+ },
+ ),
+ );
+ }
+ });
+
+ identityCounts.forEach((count, identity) => {
+ if (count <= 1) return;
+ const [, clientId, resource, address] = identity.split('|');
+ issues.push(
+ issue(
+ 'EXACT_DUPLICATE_CONNECTION',
+ 'info',
+ '连接记录重复',
+ '相同客户端、资源和地址出现了重复记录，可能来自采集侧合并或上游返回重复项。',
+ '刷新连接清单；若重复持续存在，检查客户端连接采集路径是否重复汇总。',
+ {
+ clientId,
+ resource,
+ evidence: [`address=${address}`, `count=${count}`],
+ id: `${identity}:EXACT_DUPLICATE_CONNECTION`,
+ },
+ ),
+ );
+ });
+};
+
+const addUnknownFieldIssues = (
+ connections: ClientConnection[],
+ issues: ClientConnectionIssue[],
+) => {
+ connections.forEach((connection, index) => {
+ const clientId = normalizeText(connection.clientId);
+ const resource = normalizeText(connection.groupOrTopic);
+ const protocol = normalizeText(connection.protocol);
+ const language = normalizeText(connection.language);
+ const version = normalizeText(connection.version);
+
+ if (!KNOWN_PROTOCOLS.has(protocol)) {
+ issues.push(
+ issue(
+ 'UNKNOWN_PROTOCOL',
+ 'warning',
+ '协议类型未知',
+ '该客户端连接的协议不在 Studio 已知协议列表中，统计和排障可能不准确。',
+ '确认客户端协议版本和服务端采集字段，必要时补充协议映射。',
+ {
+ clientId,
+ resource,
+ evidence: [`protocol=${protocol}`],
+ id: `${index}:UNKNOWN_PROTOCOL`,
+ },
+ ),
+ );
+ }
+
+ if (!KNOWN_LANGUAGES.has(language)) {
+ issues.push(
+ issue(
+ 'UNKNOWN_LANGUAGE',
+ 'info',
+ '客户端语言未知',
+ '该客户端连接的语言不在 Studio 已知语言列表中。',
+ '确认客户端 SDK 语言和采集字段，必要时补充语言展示映射。',
+ {
+ clientId,
+ resource,
+ evidence: [`language=${language}`],
+ id: `${index}:UNKNOWN_LANGUAGE`,
+ },
+ ),
+ );
+ }
+
+ if (version === 'unknown' || version === '-') {
+ issues.push(
+ issue(
+ 'UNKNOWN_VERSION',
+ 'warning',
+ '客户端版本未知',
+ '该客户端没有上报明确版本，升级治理和兼容性判断缺少依据。',
+ '升级客户端 SDK 或检查版本字段采集，确保连接清单能展示真实客户端版本。',
+ {
+ clientId,
+ resource,
+ evidence: [`version=${version}`],
+ id: `${index}:UNKNOWN_VERSION`,
+ },
+ ),
+ );
+ }
+
+ if (connection.connectedAt && parseTime(connection.connectedAt) === null) {
+ issues.push(
+ issue(
+ 'INVALID_CONNECTION_TIME',
+ 'info',
+ '连接时间无法解析',
+ '该连接的时间字段无法被浏览器解析，排序和人工判断可能受影响。',
+ '统一连接时间格式，优先返回 ISO-8601 或 yyyy-MM-dd HH:mm:ss。',
+ {
+ clientId,
+ resource,
+ evidence: [connection.connectedAt],
+ id: `${index}:INVALID_CONNECTION_TIME`,
+ },
+ ),
+ );
+ }
+ });
+};
+
+const addResourceIssues = (groups: ConnectionGroup[], issues:
ClientConnectionIssue[]) => {
+ groups.forEach((group) => {
+ const protocols = uniqueSorted(group.connections.map((connection) =>
connection.protocol));
+ const versions = uniqueSorted(group.connections.map((connection) =>
connection.version));
+ const languages = uniqueSorted(group.connections.map((connection) =>
connection.language));
+ const clients = uniqueSorted(group.connections.map((connection) =>
connection.clientId));
+ const addresses = uniqueSorted(group.connections.map((connection) =>
connection.address));
+ const resource = `${group.type}:${group.resource}`;
+
+ if (protocols.length > 1) {
+ issues.push(
+ issue(
+ 'MIXED_PROTOCOL_RESOURCE',
+ 'warning',
+ '同一资源存在多协议连接',
+ '同一个 Group 或 Topic 同时存在 gRPC 与 Remoting 客户端，迁移期排障需要区分控制面来源。',
+ '确认该资源是否处于协议迁移期，并分别检查 Proxy 与 Broker 侧连接状态。',
+ {
+ resource,
+ evidence: protocols,
+ },
+ ),
+ );
+ }
+
+ if (versions.length > 1) {
+ issues.push(
+ issue(
+ 'MIXED_VERSION_RESOURCE',
+ 'warning',
+ '同一资源存在多版本客户端',
+ '同一个 Group 或 Topic 内客户端版本不一致，可能导致重试、负载均衡或协议能力差异。',
+ '按资源维度收敛客户端 SDK 版本，升级后再次确认连接清单。',
+ {
+ resource,
+ evidence: versions,
+ },
+ ),
+ );
+ }
+
+ if (group.type === 'Consumer' && clients.length === 1 && addresses.length
=== 1) {
+ issues.push(
+ issue(
+ 'SINGLE_CONSUMER_INSTANCE',
+ 'warning',
+ 'Consumer Group 只有单实例在线',
+ '该 Consumer Group 当前只有一个客户端实例，进程故障会直接影响消费连续性。',
+ '为关键 Consumer Group 保持至少两个实例在线，并确认负载均衡后队列分配正常。',
+ {
+ resource,
+ evidence: [`address=${addresses[0]}`],
+ },
+ ),
+ );
+ }
+
+ if (languages.length > 1 && versions.length > 1) {
+ issues.push(
+ issue(
+ 'MIXED_VERSION_RESOURCE',
+ 'info',
+ '同一资源存在多语言多版本客户端',
+ '该资源由多语言 SDK 共同访问，升级治理和兼容性排查需要同时关注语言与版本。',
+ '记录各语言 SDK 的目标版本矩阵，避免只按单一语言判断升级完成度。',
+ {
+ resource,
+ evidence: [...languages, ...versions],
+ id: `${resource}:MIXED_LANGUAGE_VERSION_RESOURCE`,
+ },
+ ),
+ );
+ }
+ });
+};
+
+const addAddressConcentrationIssues = (
+ connections: ClientConnection[],
+ issues: ClientConnectionIssue[],
+) => {
+ if (connections.length < 4) return;
+
+ const addressCounts = countBy(connections.map((connection) =>
connection.address));
+ const threshold = Math.max(4, Math.ceil(connections.length * 0.5));
+
+ addressCounts.forEach((count, address) => {
+ if (count < threshold) return;
+ issues.push(
+ issue(
+ 'ADDRESS_CONCENTRATION',
+ 'warning',
+ '连接集中在单一地址',
+ '较多客户端连接集中在同一个地址，主机或网关故障会影响多个生产或消费链路。',
+ '检查该地址上的客户端部署密度，必要时拆分实例或调整负载分布。',
+ {
+ evidence: [`${address}: ${count}/${connections.length}`],
+ id: `${address}:ADDRESS_CONCENTRATION`,
+ },
+ ),
+ );
+ });
+};
+
+const issuesForResource = (issues: ClientConnectionIssue[], resource: string)
=>
+ issues.filter((issue) => issue.resource === resource);
+
+const buildResourceSummaries = (
+ groups: ConnectionGroup[],
+ issues: ClientConnectionIssue[],
+): ClientResourceSummary[] =>
+ groups
+ .map((group) => {
+ const resource = `${group.type}:${group.resource}`;
+ const resourceIssues = issuesForResource(issues, resource);
+ return {
+ id: resource,
+ type: group.type,
+ resource: group.resource,
+ connectionCount: group.connections.length,
+ uniqueClientCount: uniqueSorted(group.connections.map((connection) =>
connection.clientId))
+ .length,
+ uniqueAddressCount: uniqueSorted(group.connections.map((connection) =>
connection.address))
+ .length,
+ protocols: uniqueSorted(group.connections.map((connection) =>
connection.protocol)),
+ languages: uniqueSorted(group.connections.map((connection) =>
connection.language)),
+ versions: uniqueSorted(group.connections.map((connection) =>
connection.version)),
+ partial: group.connections.some((connection) => connection.partial),
+ status: resourceSeverity(resourceIssues),
+ issueCount: resourceIssues.length,
+ };
+ })
+ .sort((left, right) => {
+ const statusOrder: Record<ClientConnectionHealthStatus, number> = {
+ critical: 0,
+ warning: 1,
+ healthy: 2,
+ };
+ return (
+ statusOrder[left.status] - statusOrder[right.status] ||
+ right.issueCount - left.issueCount ||
+ left.resource.localeCompare(right.resource)
+ );
+ });
+
+const buildSummary = (
+ connections: ClientConnection[],
+ issues: ClientConnectionIssue[],
+ resources: ClientResourceSummary[],
+): ClientConnectionHealthSummary => {
+ const addressConcentrationIssues = issues.filter((item) => item.code ===
'ADDRESS_CONCENTRATION');
+ return {
+ totalConnections: connections.length,
+ uniqueClientCount: uniqueSorted(connections.map((connection) =>
connection.clientId)).length,
+ uniqueAddressCount: uniqueSorted(connections.map((connection) =>
connection.address)).length,
+ resourceCount: resources.length,
+ partialConnectionCount: connections.filter((connection) =>
connection.partial).length,
+ mixedProtocolResourceCount: resources.filter((resource) =>
resource.protocols.length > 1)
+ .length,
+ mixedVersionResourceCount: resources.filter((resource) =>
resource.versions.length > 1).length,
+ singleConsumerGroupCount: issues.filter((item) => item.code ===
'SINGLE_CONSUMER_INSTANCE')
+ .length,
+ concentratedAddressCount: addressConcentrationIssues.length,
+ };
+};
+
+const scoreDiagnostics = (issues: ClientConnectionIssue[]): number => {
+ const penalty = issues.reduce((sum, item) => {
+ if (item.severity === 'critical') return sum + 24;
+ if (item.severity === 'warning') return sum + 9;
+ return sum + 3;
+ }, 0);
+ return Math.max(0, 100 - penalty);
+};
+
+const statusFromIssues = (
+ issues: ClientConnectionIssue[],
+ score: number,
+): ClientConnectionHealthStatus => {
+ if (issues.some((item) => item.severity === 'critical') || score < 60)
return 'critical';
+ if (issues.some((item) => item.severity === 'warning') || score < 90) return
'warning';
+ return 'healthy';
+};
+
+const buildRecommendations = (issues: ClientConnectionIssue[]): string[] => {
+ const recommendations: string[] = [];
+ const seen = new Set<string>();
+
+ issues.forEach((item) => {
+ if (seen.has(item.recommendation)) return;
+ seen.add(item.recommendation);
+ recommendations.push(item.recommendation);
+ });
+
+ if (recommendations.length === 0) {
+ recommendations.push('保持客户端连接清单按集群定期巡检，重点关注协议和 SDK 版本收敛。');
+ }
+
+ return recommendations.slice(0, 6);
+};
+
+export const analyzeClientConnections = (
+ connections: ClientConnection[],
+): ClientConnectionDiagnostics => {
+ const normalizedConnections = connections.map((connection) => ({
+ ...connection,
+ clientId: normalizeText(connection.clientId),
+ type: normalizeText(connection.type),
+ groupOrTopic: normalizeText(connection.groupOrTopic),
+ protocol: normalizeText(connection.protocol),
+ address: normalizeText(connection.address),
+ language: normalizeText(connection.language),
+ version: normalizeText(connection.version),
+ }));
+ const groups = groupConnections(normalizedConnections);
+ const issues: ClientConnectionIssue[] = [];
+
+ addInventoryIssues(normalizedConnections, issues);
+ addClientIdIssues(normalizedConnections, issues);
+ addUnknownFieldIssues(normalizedConnections, issues);
+ addResourceIssues(groups, issues);
+ addAddressConcentrationIssues(normalizedConnections, issues);
+
+ const resources = buildResourceSummaries(groups, issues);
+ const score = scoreDiagnostics(issues);
+ const status = statusFromIssues(issues, score);
+
+ return {
+ status,
+ statusText: STATUS_TEXT[status],
+ statusColor: STATUS_COLOR[status],
+ score,
+ summary: buildSummary(normalizedConnections, issues, resources),
+ resources,
+ issues,
+ recommendations: buildRecommendations(issues),
+ };
+};
