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 bfe7d6f28 feat(dashboard): add traffic insights (#3042)
bfe7d6f28 is described below
commit bfe7d6f28148effa41ca5327e91fd00e875584f9
Author: Xiao Yang <[email protected]>
AuthorDate: Wed Sep 9 19:33:05 2026 +0800
feat(dashboard): add traffic insights (#3042)
The home dashboard listed per-cluster tpsIn/tpsOut with a trend bar but gave
no way to tell whether traffic was healthy, balanced or heading somewhere
bad:
the numbers were only readable one cluster at a time.
A traffic insights card now sits above the cluster table, deriving active
cluster count, top-cluster share, a 0-100 HHI balance score and unhealthy
cluster traffic share from the dashboard payload already loaded, and raising
graded findings for concentration, unhealthy traffic, broker load skew,
recent
drops and spikes, idle clusters and unavailable topology counts. The cluster
table gains total TPS, share with a progress bar, per-broker TPS and a
send/receive ratio, and the trend column now labels each series as rising,
falling or steady with its percentage. All of it is computed client side
from
the existing response, so no new endpoint is involved.
The aggregation is memoised in the page and handed to the insights card as a
prop, so it runs once per dashboard payload rather than once in the page and
again in the card. The two formatters it needs live in
utils/dashboardTrafficInsights.ts and are shared by both, instead of being
defined twice under different names.
---
web/src/i18n/translations.ts | 57 +++
web/src/pages/home/DashboardTrafficInsights.tsx | 164 +++++++++
.../pages/home/__tests__/DashboardPage.test.tsx | 74 +++-
.../__tests__/DashboardTrafficInsights.test.tsx | 155 ++++++++
web/src/pages/home/dashboard.tsx | 131 ++++++-
web/src/utils/dashboardTrafficInsights.test.ts | 269 ++++++++++++++
web/src/utils/dashboardTrafficInsights.ts | 400 +++++++++++++++++++++
7 files changed, 1224 insertions(+), 26 deletions(-)
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index a66f67fba..f71894289 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -112,6 +112,63 @@ const translations: Record<string, Record<Lang, string>> =
{
'dashboard.consumerGroups': { zh: '{n} 消费组', en: '{n} Groups' },
'dashboard.healthy': { zh: '健康', en: 'Healthy' },
'dashboard.last12h': { zh: '近 12 小时', en: 'Last 12 hours' },
+ 'dashboardTraffic.title': { zh: '流量洞察', en: 'Traffic Insights' },
+ 'dashboardTraffic.activeClusters': { zh: '活跃集群', en: 'Active Clusters' },
+ 'dashboardTraffic.topClusterShare': { zh: '最高流量占比', en: 'Top Traffic Share'
},
+ 'dashboardTraffic.balanceScore': { zh: '均衡评分', en: 'Balance Score' },
+ 'dashboardTraffic.unhealthyTraffic': { zh: '异常流量', en: 'Unhealthy Traffic' },
+ 'dashboardTraffic.findings': { zh: '需要关注的流量信号', en: 'Traffic signals to
review' },
+ 'dashboardTraffic.noCluster': { zh: '暂无集群流量数据', en: 'No cluster traffic
data' },
+ 'dashboardTraffic.allClusters': { zh: '全部集群', en: 'all clusters' },
+ 'dashboardTraffic.totalTps': { zh: '总 TPS', en: 'Total TPS' },
+ 'dashboardTraffic.share': { zh: '占比', en: 'Share' },
+ 'dashboardTraffic.perBroker': { zh: '单 Broker TPS', en: 'TPS per Broker' },
+ 'dashboardTraffic.inOutRatio': { zh: '出入比', en: 'Out/In Ratio' },
+ 'dashboardTraffic.trendRising': { zh: '上升', en: 'Rising' },
+ 'dashboardTraffic.trendFalling': { zh: '下降', en: 'Falling' },
+ 'dashboardTraffic.trendStable': { zh: '平稳', en: 'Stable' },
+ 'dashboardTraffic.trendUnknown': { zh: '未知', en: 'Unknown' },
+ 'dashboardTraffic.level.healthy': { zh: '健康', en: 'Healthy' },
+ 'dashboardTraffic.level.notice': { zh: '提示', en: 'Notice' },
+ 'dashboardTraffic.level.warning': { zh: '告警', en: 'Warning' },
+ 'dashboardTraffic.level.critical': { zh: '严重', en: 'Critical' },
+ 'dashboardTraffic.issueNoActiveTraffic': {
+ zh: '未检测到活跃流量',
+ en: 'No active traffic detected',
+ },
+ 'dashboardTraffic.issueTrafficConcentration': {
+ zh: '{cluster} 承载 {value} 流量',
+ en: '{cluster} carries {value} of traffic',
+ },
+ 'dashboardTraffic.issueUnhealthyTraffic': {
+ zh: '非健康集群承载 {value} 流量',
+ en: 'Unhealthy clusters carry {value} of traffic',
+ },
+ 'dashboardTraffic.issueUnhealthyClusterTraffic': {
+ zh: '{cluster} 在非健康状态下承载流量',
+ en: '{cluster} carries traffic while unhealthy',
+ },
+ 'dashboardTraffic.issueBrokerLoadSkew': {
+ zh: '{cluster} 单 Broker 流量偏高',
+ en: '{cluster} has high per-broker traffic',
+ },
+ 'dashboardTraffic.issueRecentDrop': {
+ zh: '近期总流量下降 {value}',
+ en: 'Recent total traffic dropped by {value}',
+ },
+ 'dashboardTraffic.issueRecentSpike': {
+ zh: '近期总流量上升 {value}',
+ en: 'Recent total traffic increased by {value}',
+ },
+ 'dashboardTraffic.issueTopologyUnavailable': {
+ zh: '{cluster} 拓扑数量不可用',
+ en: '{cluster} has unavailable topology counts',
+ },
+ 'dashboardTraffic.issueIdleCluster': {
+ zh: '{cluster} 当前无流量',
+ en: '{cluster} is currently idle',
+ },
+ 'dashboardTraffic.issueUnknown': { zh: '未知流量信号', en: 'Unknown traffic
signal' },
// ─── Cluster Page ───
'cluster.title': { zh: 'RocketMQ 集群', en: 'RocketMQ Cluster' },
diff --git a/web/src/pages/home/DashboardTrafficInsights.tsx
b/web/src/pages/home/DashboardTrafficInsights.tsx
new file mode 100644
index 000000000..6bfe372cc
--- /dev/null
+++ b/web/src/pages/home/DashboardTrafficInsights.tsx
@@ -0,0 +1,164 @@
+/*
+ * 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 { Alert, Card, Col, Empty, Flex, Row, Statistic, Tag, Typography } from
'antd';
+import { useLang } from '../../i18n/LangContext';
+import {
+ formatTrafficPercent,
+ formatTrafficTps,
+ type DashboardTrafficInsights,
+ type DashboardTrafficIssue,
+ type TrafficHealthLevel,
+} from '../../utils/dashboardTrafficInsights';
+
+const { Text } = Typography;
+
+interface Props {
+ insights: DashboardTrafficInsights;
+}
+
+const levelColor: Record<TrafficHealthLevel, string> = {
+ healthy: 'success',
+ notice: 'processing',
+ warning: 'warning',
+ critical: 'error',
+};
+
+const issueTextKey = (issue: DashboardTrafficIssue) => {
+ switch (issue.code) {
+ case 'NO_ACTIVE_TRAFFIC':
+ return 'dashboardTraffic.issueNoActiveTraffic';
+ case 'TRAFFIC_CONCENTRATION':
+ return 'dashboardTraffic.issueTrafficConcentration';
+ case 'UNHEALTHY_TRAFFIC':
+ return issue.clusterName
+ ? 'dashboardTraffic.issueUnhealthyClusterTraffic'
+ : 'dashboardTraffic.issueUnhealthyTraffic';
+ case 'BROKER_LOAD_SKEW':
+ return 'dashboardTraffic.issueBrokerLoadSkew';
+ case 'RECENT_TRAFFIC_DROP':
+ return 'dashboardTraffic.issueRecentDrop';
+ case 'RECENT_TRAFFIC_SPIKE':
+ return 'dashboardTraffic.issueRecentSpike';
+ case 'TOPOLOGY_COUNT_UNAVAILABLE':
+ return 'dashboardTraffic.issueTopologyUnavailable';
+ case 'IDLE_CLUSTER':
+ return 'dashboardTraffic.issueIdleCluster';
+ default:
+ return 'dashboardTraffic.issueUnknown';
+ }
+};
+
+const DashboardTrafficInsights = ({ insights }: Props) => {
+ const { t } = useLang();
+ const visibleIssues = insights.issues.slice(0, 4);
+
+ const renderIssue = (issue: DashboardTrafficIssue) => (
+ <Tag key={`${issue.code}-${issue.clusterId ?? 'global'}`}
color={levelColor[issue.level]}>
+ {t(issueTextKey(issue), {
+ cluster: issue.clusterName ?? t('dashboardTraffic.allClusters'),
+ value:
+ issue.value == null
+ ? '-'
+ : formatTrafficPercent(
+ issue.code === 'RECENT_TRAFFIC_DROP' ? Math.abs(issue.value) :
issue.value,
+ ),
+ threshold: issue.threshold == null ? '-' :
formatTrafficPercent(issue.threshold),
+ })}
+ </Tag>
+ );
+
+ return (
+ <Card
+ title={t('dashboardTraffic.title')}
+ extra={
+ <Tag color={levelColor[insights.level]}>
+ {t(`dashboardTraffic.level.${insights.level}`)}
+ </Tag>
+ }
+ style={{ marginBottom: 24 }}
+ styles={{ body: { padding: 20 } }}
+ >
+ <Row gutter={[12, 12]} style={{ marginBottom: 16 }}>
+ <Col xs={12} lg={6}>
+ <Card size="small">
+ <Statistic
+ title={t('dashboardTraffic.activeClusters')}
+
value={`${insights.activeClusterCount}/${insights.totalClusterCount}`}
+ />
+ </Card>
+ </Col>
+ <Col xs={12} lg={6}>
+ <Card size="small">
+ <Statistic
+ title={t('dashboardTraffic.topClusterShare')}
+ value={insights.topClusterSharePercent}
+ suffix="%"
+ precision={1}
+ />
+ <Text type="secondary">{insights.topCluster?.name ?? '-'}</Text>
+ </Card>
+ </Col>
+ <Col xs={12} lg={6}>
+ <Card size="small">
+ <Statistic
+ title={t('dashboardTraffic.balanceScore')}
+ value={insights.balanceScore}
+ suffix="/100"
+ />
+ </Card>
+ </Col>
+ <Col xs={12} lg={6}>
+ <Card size="small">
+ <Statistic
+ title={t('dashboardTraffic.unhealthyTraffic')}
+ value={formatTrafficTps(insights.unhealthyTrafficTps)}
+ suffix="/s"
+ valueStyle={{ color: insights.unhealthyTrafficTps > 0 ?
'#cf1322' : undefined }}
+ />
+ </Card>
+ </Col>
+ </Row>
+
+ {visibleIssues.length > 0 && (
+ <Alert
+ showIcon
+ type={
+ insights.level === 'critical'
+ ? 'error'
+ : insights.level === 'warning'
+ ? 'warning'
+ : 'info'
+ }
+ message={t('dashboardTraffic.findings')}
+ description={
+ <Flex wrap="wrap" gap={8}>
+ {visibleIssues.map(renderIssue)}
+ </Flex>
+ }
+ style={{ marginBottom: 16 }}
+ />
+ )}
+
+ {insights.rows.length === 0 ? (
+ <Empty image={Empty.PRESENTED_IMAGE_SIMPLE}
description={t('dashboardTraffic.noCluster')} />
+ ) : null}
+ </Card>
+ );
+};
+
+export default DashboardTrafficInsights;
diff --git a/web/src/pages/home/__tests__/DashboardPage.test.tsx
b/web/src/pages/home/__tests__/DashboardPage.test.tsx
index 898eb9f20..baafc771c 100644
--- a/web/src/pages/home/__tests__/DashboardPage.test.tsx
+++ b/web/src/pages/home/__tests__/DashboardPage.test.tsx
@@ -65,6 +65,35 @@ const unavailableTopologyDashboard = (): DashboardData => ({
})),
});
+const trafficDashboard = (): DashboardData => ({
+ ...dashboard('traffic-a'),
+ stats: {
+ ...dashboard('traffic-a').stats,
+ totalClusters: 2,
+ totalBrokers: 3,
+ totalProxies: 1,
+ tpsIn: 125,
+ tpsOut: 75,
+ messagesPerSecond: 200,
+ },
+ clusters: [
+ {
+ ...dashboard('traffic-a').clusters[0],
+ brokers: 2,
+ tpsIn: 100,
+ tpsOut: 50,
+ throughput: [10, 20, 30, 80],
+ },
+ {
+ ...dashboard('traffic-b').clusters[0],
+ brokers: 1,
+ tpsIn: 25,
+ tpsOut: 25,
+ throughput: [25, 25, 25, 25],
+ },
+ ],
+});
+
const deferred = <T,>() => {
let resolve!: (value: T) => void;
const promise = new Promise<T>((promiseResolve) => {
@@ -136,14 +165,41 @@ describe('DashboardPage', () => {
vi.mocked(dashboardService.getDashboard).mockResolvedValue(unavailableTopologyDashboard());
renderWithProviders(<DashboardPage />);
- await screen.findByText('proxy-cluster');
+ await screen.findAllByText('proxy-cluster');
expect(await screen.findByText(/1 Brokers · N\/A
Proxy/u)).toBeInTheDocument();
expect(screen.queryByText('0 Proxy')).not.toBeInTheDocument();
- const row = screen.getByText('proxy-cluster').closest('tr');
+ const clusterHealthCard = screen.getByText('集群健康概览').closest('.ant-card');
+ expect(clusterHealthCard).not.toBeNull();
+ const row = within(clusterHealthCard as HTMLElement)
+ .getByText('proxy-cluster')
+ .closest('tr');
expect(row).not.toBeNull();
expect(within(row as
HTMLElement).getAllByText('N/A').length).toBeGreaterThanOrEqual(1);
});
+ it('merges traffic insight metrics into the existing cluster health table',
async () => {
+
vi.mocked(dashboardService.getDashboard).mockResolvedValue(trafficDashboard());
+ renderWithProviders(<DashboardPage />);
+
+ await screen.findAllByText('traffic-a');
+ const clusterHealthCard = screen.getByText('集群健康概览').closest('.ant-card');
+ expect(clusterHealthCard).not.toBeNull();
+ const clusterHealth = within(clusterHealthCard as HTMLElement);
+ expect(clusterHealth.getAllByText('总 TPS')).not.toHaveLength(0);
+ expect(clusterHealth.getAllByText('占比')).not.toHaveLength(0);
+ expect(clusterHealth.getAllByText('单 Broker TPS')).not.toHaveLength(0);
+ expect(clusterHealth.getAllByText('出入比')).not.toHaveLength(0);
+
+ const row = clusterHealth.getByText('traffic-a').closest('tr');
+ expect(row).not.toBeNull();
+ const clusterRow = within(row as HTMLElement);
+ expect(clusterRow.getByText('150/s')).toBeInTheDocument();
+ expect(clusterRow.getByText('75%')).toBeInTheDocument();
+ expect(clusterRow.getByText('75/s')).toBeInTheDocument();
+ expect(clusterRow.getByText('0.5:1')).toBeInTheDocument();
+ expect(clusterRow.getByText(/上升/u)).toBeInTheDocument();
+ });
+
it('does not show dashboard data from the previous instance while loading a
new selection', async () => {
const instanceA = deferred<DashboardData>();
vi.mocked(dashboardService.getDashboard)
@@ -152,7 +208,7 @@ describe('DashboardPage', () => {
const user = userEvent.setup();
renderWithProviders(<DashboardPage />);
- await screen.findByText('initial-cluster');
+ await screen.findAllByText('initial-cluster');
const selector = screen.getByRole('combobox', { name: 'Dashboard instance'
});
await user.click(selector);
await user.click(
@@ -162,7 +218,7 @@ describe('DashboardPage', () => {
expect(screen.queryByText('initial-cluster')).not.toBeInTheDocument();
instanceA.resolve(dashboard('instance-a-cluster'));
- await screen.findByText('instance-a-cluster');
+ await screen.findAllByText('instance-a-cluster');
});
it('does not let a stale instance response overwrite the latest selection',
async () => {
@@ -175,7 +231,7 @@ describe('DashboardPage', () => {
const user = userEvent.setup();
renderWithProviders(<DashboardPage />);
- await screen.findByText('initial-cluster');
+ await screen.findAllByText('initial-cluster');
const selector = screen.getByRole('combobox', { name: 'Dashboard instance'
});
await user.click(selector);
await user.click(
@@ -187,13 +243,13 @@ describe('DashboardPage', () => {
);
instanceB.resolve(dashboard('instance-b-cluster'));
- expect(await screen.findByText('instance-b-cluster')).toBeInTheDocument();
+ expect(await
screen.findAllByText('instance-b-cluster')).not.toHaveLength(0);
instanceA.resolve(dashboard('instance-a-cluster'));
await waitFor(() => {
expect(screen.queryByText('instance-a-cluster')).not.toBeInTheDocument();
});
- expect(screen.getByText('instance-b-cluster')).toBeInTheDocument();
+ expect(screen.getAllByText('instance-b-cluster')).not.toHaveLength(0);
});
it('preserves the selected instance when navigating to the cluster page',
async () => {
@@ -206,7 +262,7 @@ describe('DashboardPage', () => {
</>,
);
- await screen.findByText('instance-a-cluster');
+ await screen.findAllByText('instance-a-cluster');
const selector = screen.getByRole('combobox', { name: 'Dashboard instance'
});
await user.click(selector);
await user.click(
@@ -248,7 +304,7 @@ describe('DashboardPage', () => {
const user = userEvent.setup();
renderWithProviders(<DashboardPage />);
- await screen.findByText('apache-cluster');
+ await screen.findAllByText('apache-cluster');
await user.click(screen.getByRole('combobox', { name: 'Dashboard instance'
}));
expect(
diff --git a/web/src/pages/home/__tests__/DashboardTrafficInsights.test.tsx
b/web/src/pages/home/__tests__/DashboardTrafficInsights.test.tsx
new file mode 100644
index 000000000..ea08d0176
--- /dev/null
+++ b/web/src/pages/home/__tests__/DashboardTrafficInsights.test.tsx
@@ -0,0 +1,155 @@
+/*
+ * 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 { App } from 'antd';
+import { render, screen } from '@testing-library/react';
+import { beforeAll, describe, expect, it, vi } from 'vitest';
+import type { DashboardData } from '../../../api/metrics';
+import { LangProvider } from '../../../i18n/LangContext';
+import { buildDashboardTrafficInsights } from
'../../../utils/dashboardTrafficInsights';
+import DashboardTrafficInsights from '../DashboardTrafficInsights';
+
+const dashboard: DashboardData = {
+ stats: {
+ totalClusters: 3,
+ healthyClusters: 2,
+ totalBrokers: 8,
+ totalProxies: 3,
+ totalNameServers: 3,
+ totalTopics: 100,
+ totalConsumerGroups: 30,
+ totalMessagesToday: 1_000_000,
+ messagesPerSecond: 3200,
+ tpsIn: 1600,
+ tpsOut: 1600,
+ },
+ clusters: [
+ {
+ id: 'prod',
+ name: 'prod',
+ type: 'V5_PROXY_CLUSTER',
+ status: 'healthy',
+ brokers: 4,
+ proxies: 2,
+ topics: 70,
+ groups: 20,
+ tpsIn: 1400,
+ tpsOut: 1400,
+ version: '5.2.0',
+ throughput: [800, 900, 1000, 1300, 1400, 1500],
+ },
+ {
+ id: 'canary',
+ name: 'canary',
+ type: 'V5_PROXY_CLUSTER',
+ status: 'warning',
+ brokers: 2,
+ proxies: 1,
+ topics: 20,
+ groups: 8,
+ tpsIn: 180,
+ tpsOut: 180,
+ version: '5.2.0',
+ throughput: [220, 230, 240, 180, 170, 160],
+ },
+ {
+ id: 'idle',
+ name: 'idle',
+ type: 'V4_NAMESRV',
+ status: 'healthy',
+ brokers: 2,
+ proxies: null,
+ topics: 10,
+ groups: 2,
+ tpsIn: 0,
+ tpsOut: 0,
+ version: '4.9.8',
+ throughput: [0, 0, 0, 0],
+ },
+ ],
+};
+
+const renderPanel = (data = dashboard) =>
+ render(
+ <App>
+ <LangProvider>
+ <DashboardTrafficInsights
insights={buildDashboardTrafficInsights(data)} />
+ </LangProvider>
+ </App>,
+ );
+
+beforeAll(() => {
+ Object.defineProperty(window, 'matchMedia', {
+ writable: true,
+ value: vi.fn().mockImplementation((query: string) => ({
+ matches: false,
+ media: query,
+ onchange: null,
+ addListener: vi.fn(),
+ removeListener: vi.fn(),
+ addEventListener: vi.fn(),
+ removeEventListener: vi.fn(),
+ dispatchEvent: vi.fn(),
+ })),
+ });
+});
+
+describe('DashboardTrafficInsights', () => {
+ it('renders traffic findings and summary cards without duplicating cluster
rows', () => {
+ renderPanel();
+
+ expect(screen.getByText('流量洞察')).toBeInTheDocument();
+ expect(screen.getByText('严重')).toBeInTheDocument();
+ expect(screen.getByText('活跃集群')).toBeInTheDocument();
+ expect(screen.getByText('2/3')).toBeInTheDocument();
+ expect(screen.getByText('最高流量占比')).toBeInTheDocument();
+ expect(screen.getAllByText((_, element) => element?.textContent ===
'88.6%')).not.toHaveLength(
+ 0,
+ );
+ expect(screen.getByText('异常流量')).toBeInTheDocument();
+ expect(screen.getAllByText((_, element) => element?.textContent ===
'360/s')).not.toHaveLength(
+ 0,
+ );
+
+ expect(screen.queryByRole('table')).not.toBeInTheDocument();
+ expect(screen.getByText(/prod 承载 88.6% 流量/u)).toBeInTheDocument();
+ expect(screen.getByText(/canary 在非健康状态下承载流量/u)).toBeInTheDocument();
+ });
+
+ it('shows an empty state when the dashboard has no cluster rows', () => {
+ renderPanel({
+ stats: {
+ totalClusters: 0,
+ healthyClusters: 0,
+ totalBrokers: 0,
+ totalProxies: 0,
+ totalNameServers: 0,
+ totalTopics: 0,
+ totalConsumerGroups: 0,
+ totalMessagesToday: 0,
+ messagesPerSecond: 0,
+ tpsIn: 0,
+ tpsOut: 0,
+ },
+ clusters: [],
+ });
+
+ expect(screen.getByText('流量洞察')).toBeInTheDocument();
+ expect(screen.getByText('暂无集群流量数据')).toBeInTheDocument();
+ expect(screen.getByText('未检测到活跃流量')).toBeInTheDocument();
+ });
+});
diff --git a/web/src/pages/home/dashboard.tsx b/web/src/pages/home/dashboard.tsx
index 24c36c25b..98cefbf57 100644
--- a/web/src/pages/home/dashboard.tsx
+++ b/web/src/pages/home/dashboard.tsx
@@ -1,10 +1,12 @@
-import { useCallback, useEffect, useRef, useState } from 'react';
+import { useCallback, useEffect, useMemo, useRef, useState } from 'react';
import { useNavigate } from 'react-router-dom';
import {
Alert,
Button,
Card,
Col,
+ Flex,
+ Progress,
Row,
Select,
Skeleton,
@@ -14,6 +16,7 @@ import {
Tag,
Typography,
} from 'antd';
+import type { ColumnsType } from 'antd/es/table';
import { ClusterOutlined, ThunderboltOutlined } from '@ant-design/icons';
import { ListDashes, ArrowDown } from '@phosphor-icons/react';
import PageHeader from '../../components/PageHeader';
@@ -26,12 +29,42 @@ import type { DashboardData } from '../../api/metrics';
import { supportsApacheRuntime, type Instance } from '../../api/instance';
import { listInstances } from '../../services/instanceService';
import { useLang } from '../../i18n/LangContext';
+import DashboardTrafficInsights from './DashboardTrafficInsights';
+import {
+ buildDashboardTrafficInsights,
+ formatTrafficPercent,
+ formatTrafficTps,
+ type TrafficTrendDirection,
+} from '../../utils/dashboardTrafficInsights';
+import { tableScrollX } from '../../utils/table';
const { Text } = Typography;
+type ClusterRow = DashboardData['clusters'][number];
+
+const trafficTrendColor: Record<TrafficTrendDirection, string> = {
+ rising: 'green',
+ falling: 'volcano',
+ stable: 'blue',
+ unknown: 'default',
+};
+
+const trafficTrendLabelKey: Record<TrafficTrendDirection, string> = {
+ rising: 'dashboardTraffic.trendRising',
+ falling: 'dashboardTraffic.trendFalling',
+ stable: 'dashboardTraffic.trendStable',
+ unknown: 'dashboardTraffic.trendUnknown',
+};
+
const renderTopologyCount = (value: number | null) =>
value === null ? 'N/A' : value.toLocaleString();
+const formatTrafficTrendDelta = (value: number | null) => {
+ if (value == null) return '';
+ const sign = value > 0 ? '+' : '';
+ return ` ${sign}${formatTrafficPercent(value)}`;
+};
+
const DashboardPage = () => {
const navigate = useNavigate();
const { t } = useLang();
@@ -87,6 +120,11 @@ const DashboardPage = () => {
const visibleDashboard = dashboardInstanceId === selectedInstanceId ?
dashboard : null;
+ const trafficInsights = useMemo(
+ () => buildDashboardTrafficInsights(visibleDashboard),
+ [visibleDashboard],
+ );
+
const dashboardHeader = (
<PageHeader
title={t('dashboard.title')}
@@ -139,6 +177,7 @@ const DashboardPage = () => {
}
const { stats, clusters } = visibleDashboard;
+ const trafficInsightByClusterId = new Map(trafficInsights.rows.map((row) =>
[row.id, row]));
const statCards = [
{
@@ -177,7 +216,7 @@ const DashboardPage = () => {
},
];
- const clusterColumns = [
+ const clusterColumns: ColumnsType<ClusterRow> = [
{
title: t('dashboard.clusterName'),
dataIndex: 'name',
@@ -252,25 +291,80 @@ const DashboardPage = () => {
align: 'right' as const,
render: (v: number) => v.toLocaleString(),
},
+ {
+ title: t('dashboardTraffic.totalTps'),
+ key: 'trafficTotalTps',
+ width: 120,
+ align: 'right' as const,
+ render: (_, record) => {
+ const insight = trafficInsightByClusterId.get(record.id);
+ return insight ? `${formatTrafficTps(insight.totalTps)}/s` : '-';
+ },
+ },
+ {
+ title: t('dashboardTraffic.share'),
+ key: 'trafficShare',
+ width: 150,
+ render: (_, record) => {
+ const insight = trafficInsightByClusterId.get(record.id);
+ if (!insight) return '-';
+ return (
+ <Flex vertical gap={4}>
+ <Text>{formatTrafficPercent(insight.sharePercent)}</Text>
+ <Progress percent={Math.min(100, insight.sharePercent)}
showInfo={false} size="small" />
+ </Flex>
+ );
+ },
+ },
+ {
+ title: t('dashboardTraffic.perBroker'),
+ key: 'trafficPerBroker',
+ width: 130,
+ align: 'right' as const,
+ render: (_, record) => {
+ const insight = trafficInsightByClusterId.get(record.id);
+ return insight ? `${formatTrafficTps(insight.perBrokerTps)}/s` : '-';
+ },
+ },
+ {
+ title: t('dashboardTraffic.inOutRatio'),
+ key: 'trafficInOutRatio',
+ width: 110,
+ align: 'right' as const,
+ render: (_, record) => {
+ const insight = trafficInsightByClusterId.get(record.id);
+ return insight?.inOutRatio == null ? 'N/A' : `${insight.inOutRatio}:1`;
+ },
+ },
{
title: t('dashboard.trend'),
dataIndex: 'throughput',
key: 'throughput',
- width: 110,
- render: (data: number[], record: DashboardData['clusters'][0]) => (
- <MiniBar
- data={data}
- color={
- record.status === 'healthy'
- ? '#52c41a'
- : record.status === 'warning'
- ? '#faad14'
- : '#d9d9d9'
- }
- height={26}
- width={100}
- />
- ),
+ width: 150,
+ render: (data: number[], record) => {
+ const insight = trafficInsightByClusterId.get(record.id);
+ const trendDirection = insight?.trendDirection ?? 'unknown';
+ return (
+ <Space direction="vertical" size={2}>
+ <MiniBar
+ data={data}
+ color={
+ record.status === 'healthy'
+ ? '#52c41a'
+ : record.status === 'warning'
+ ? '#faad14'
+ : '#d9d9d9'
+ }
+ height={26}
+ width={100}
+ />
+ <Tag color={trafficTrendColor[trendDirection]}>
+ {t(trafficTrendLabelKey[trendDirection])}
+ {formatTrafficTrendDelta(insight?.trendDeltaPercent ?? null)}
+ </Tag>
+ </Space>
+ );
+ },
},
];
@@ -297,6 +391,8 @@ const DashboardPage = () => {
))}
</Row>
+ <DashboardTrafficInsights insights={trafficInsights} />
+
<Card
title={t('dashboard.clusterHealth')}
extra={<a onClick={() =>
navigate(clusterPagePath)}>{t('common.viewAll')}</a>}
@@ -308,6 +404,7 @@ const DashboardPage = () => {
rowKey="id"
size="small"
pagination={false}
+ scroll={{ x: tableScrollX(clusterColumns) }}
onRow={() => ({
style: { cursor: 'pointer' },
onClick: () => navigate(clusterPagePath),
diff --git a/web/src/utils/dashboardTrafficInsights.test.ts
b/web/src/utils/dashboardTrafficInsights.test.ts
new file mode 100644
index 000000000..008d3a51c
--- /dev/null
+++ b/web/src/utils/dashboardTrafficInsights.test.ts
@@ -0,0 +1,269 @@
+/*
+ * 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 { ClusterOverview, DashboardData } from '../api/metrics';
+import { buildDashboardTrafficInsights } from './dashboardTrafficInsights';
+
+const baseCluster = (overrides: Partial<ClusterOverview> = {}):
ClusterOverview => ({
+ id: 'cluster-a',
+ name: 'cluster-a',
+ type: 'V5_PROXY_CLUSTER',
+ status: 'healthy',
+ brokers: 4,
+ proxies: 2,
+ topics: 32,
+ groups: 16,
+ tpsIn: 500,
+ tpsOut: 500,
+ version: '5.2.0',
+ throughput: [900, 940, 980, 1000, 1020, 1040],
+ ...overrides,
+});
+
+const dashboard = (clusters: ClusterOverview[]): DashboardData => ({
+ stats: {
+ totalClusters: clusters.length,
+ healthyClusters: clusters.filter((cluster) => cluster.status ===
'healthy').length,
+ totalBrokers: clusters.reduce((sum, cluster) => sum + cluster.brokers, 0),
+ totalProxies: clusters.reduce((sum, cluster) => sum + (cluster.proxies ??
0), 0),
+ totalNameServers: 2,
+ totalTopics: clusters.reduce((sum, cluster) => sum + cluster.topics, 0),
+ totalConsumerGroups: clusters.reduce((sum, cluster) => sum +
cluster.groups, 0),
+ totalMessagesToday: 0,
+ messagesPerSecond: clusters.reduce((sum, cluster) => sum + cluster.tpsIn +
cluster.tpsOut, 0),
+ tpsIn: clusters.reduce((sum, cluster) => sum + cluster.tpsIn, 0),
+ tpsOut: clusters.reduce((sum, cluster) => sum + cluster.tpsOut, 0),
+ },
+ clusters,
+});
+
+describe('buildDashboardTrafficInsights', () => {
+ it('summarizes balanced healthy cluster traffic without warnings', () => {
+ const result = buildDashboardTrafficInsights(
+ dashboard([
+ baseCluster({ id: 'cluster-a', name: 'cluster-a', tpsIn: 500, tpsOut:
500 }),
+ baseCluster({ id: 'cluster-b', name: 'cluster-b', tpsIn: 450, tpsOut:
550 }),
+ ]),
+ );
+
+ expect(result.level).toBe('healthy');
+ expect(result.totalTps).toBe(2000);
+ expect(result.activeClusterCount).toBe(2);
+ expect(result.topCluster?.name).toBe('cluster-a');
+ expect(result.topClusterSharePercent).toBe(50);
+ expect(result.balanceScore).toBe(100);
+ expect(result.averagePerBrokerTps).toBe(250);
+ expect(result.issues).toEqual([]);
+ expect(result.rows.map((row) => row.name)).toEqual(['cluster-a',
'cluster-b']);
+ });
+
+ it('flags traffic concentration when one cluster owns most active
throughput', () => {
+ const result = buildDashboardTrafficInsights(
+ dashboard([
+ baseCluster({ id: 'prod', name: 'prod', brokers: 8, tpsIn: 1400,
tpsOut: 1600 }),
+ baseCluster({ id: 'pre', name: 'pre', brokers: 4, tpsIn: 100, tpsOut:
100 }),
+ ]),
+ );
+
+ expect(result.level).toBe('critical');
+ expect(result.topCluster?.name).toBe('prod');
+ expect(result.topClusterSharePercent).toBe(93.8);
+ expect(result.balanceScore).toBeLessThan(30);
+ expect(result.issues).toEqual(
+ expect.arrayContaining([
+ expect.objectContaining({
+ code: 'TRAFFIC_CONCENTRATION',
+ level: 'critical',
+ clusterName: 'prod',
+ }),
+ ]),
+ );
+ });
+
+ it('surfaces unhealthy clusters carrying live traffic', () => {
+ const result = buildDashboardTrafficInsights(
+ dashboard([
+ baseCluster({ id: 'healthy', name: 'healthy', tpsIn: 700, tpsOut: 700
}),
+ baseCluster({
+ id: 'degraded',
+ name: 'degraded',
+ status: 'warning',
+ tpsIn: 300,
+ tpsOut: 300,
+ }),
+ ]),
+ );
+
+ expect(result.unhealthyClusterCount).toBe(1);
+ expect(result.unhealthyTrafficTps).toBe(600);
+ expect(result.unhealthyTrafficPercent).toBe(30);
+ expect(result.level).toBe('critical');
+ expect(result.rows.find((row) => row.name === 'degraded')?.issues).toEqual(
+ expect.arrayContaining([
+ expect.objectContaining({
+ code: 'UNHEALTHY_TRAFFIC',
+ level: 'critical',
+ clusterName: 'degraded',
+ }),
+ ]),
+ );
+ });
+
+ it('detects broker load skew using normalized per-broker traffic', () => {
+ const result = buildDashboardTrafficInsights(
+ dashboard([
+ baseCluster({ id: 'hot', name: 'hot', brokers: 1, tpsIn: 900, tpsOut:
900 }),
+ baseCluster({ id: 'wide-a', name: 'wide-a', brokers: 12, tpsIn: 300,
tpsOut: 300 }),
+ baseCluster({ id: 'wide-b', name: 'wide-b', brokers: 12, tpsIn: 300,
tpsOut: 300 }),
+ ]),
+ );
+
+ expect(result.averagePerBrokerTps).toBe(120);
+ expect(result.rows[0].perBrokerTps).toBe(1800);
+ expect(result.issues).toEqual(
+ expect.arrayContaining([
+ expect.objectContaining({
+ code: 'BROKER_LOAD_SKEW',
+ level: 'warning',
+ clusterName: 'hot',
+ threshold: 300,
+ }),
+ ]),
+ );
+ });
+
+ it('classifies falling aggregate traffic from the dashboard history', () => {
+ const result = buildDashboardTrafficInsights(
+ dashboard([
+ baseCluster({
+ id: 'cluster-a',
+ throughput: [1000, 1000, 1000, 400, 400, 400],
+ }),
+ baseCluster({
+ id: 'cluster-b',
+ throughput: [500, 500, 500, 200, 200, 200],
+ }),
+ ]),
+ );
+
+ expect(result.trendDirection).toBe('falling');
+ expect(result.trendDeltaPercent).toBe(-60);
+ expect(result.aggregateThroughput).toEqual([1500, 1500, 1500, 600, 600,
600]);
+ expect(result.issues).toEqual(
+ expect.arrayContaining([
+ expect.objectContaining({
+ code: 'RECENT_TRAFFIC_DROP',
+ level: 'warning',
+ value: -60,
+ }),
+ ]),
+ );
+ });
+
+ it('classifies recent spikes as informational notices instead of errors', ()
=> {
+ const result = buildDashboardTrafficInsights(
+ dashboard([
+ baseCluster({
+ id: 'cluster-a',
+ throughput: [200, 200, 200, 500, 500, 500],
+ }),
+ ]),
+ );
+
+ expect(result.trendDirection).toBe('rising');
+ expect(result.trendDeltaPercent).toBe(150);
+ expect(result.issues).toEqual(
+ expect.arrayContaining([
+ expect.objectContaining({
+ code: 'RECENT_TRAFFIC_SPIKE',
+ level: 'notice',
+ value: 150,
+ }),
+ ]),
+ );
+ });
+
+ it('marks missing topology counts without treating them as zero-capacity
proxies', () => {
+ const result = buildDashboardTrafficInsights(
+ dashboard([
+ baseCluster({
+ id: 'proxy-cluster',
+ name: 'proxy-cluster',
+ proxies: null,
+ tpsIn: 100,
+ tpsOut: 100,
+ }),
+ ]),
+ );
+
+ expect(result.rows[0].proxies).toBeNull();
+ expect(result.rows[0].issues).toEqual(
+ expect.arrayContaining([
+ expect.objectContaining({
+ code: 'TOPOLOGY_COUNT_UNAVAILABLE',
+ level: 'notice',
+ clusterName: 'proxy-cluster',
+ }),
+ ]),
+ );
+ });
+
+ it('treats empty dashboards as no active traffic with an empty row set', ()
=> {
+ const result = buildDashboardTrafficInsights(null);
+
+ expect(result.totalTps).toBe(0);
+ expect(result.activeClusterCount).toBe(0);
+ expect(result.totalClusterCount).toBe(0);
+ expect(result.topCluster).toBeNull();
+ expect(result.rows).toEqual([]);
+ expect(result.issues).toEqual([
+ expect.objectContaining({
+ code: 'NO_ACTIVE_TRAFFIC',
+ level: 'notice',
+ }),
+ ]);
+ });
+
+ it('normalizes invalid and negative numeric samples before computing
shares', () => {
+ const malformed = baseCluster({
+ id: 'malformed',
+ name: 'malformed',
+ brokers: -1,
+ topics: -20,
+ groups: Number.NaN,
+ tpsIn: Number.POSITIVE_INFINITY,
+ tpsOut: -100,
+ throughput: [Number.NaN, -10, 20],
+ }) as ClusterOverview;
+
+ const result = buildDashboardTrafficInsights(dashboard([malformed]));
+
+ expect(result.totalTps).toBe(0);
+ expect(result.rows[0]).toEqual(
+ expect.objectContaining({
+ brokers: 0,
+ topics: 0,
+ groups: 0,
+ totalTps: 0,
+ sharePercent: 0,
+ perBrokerTps: 0,
+ throughput: [0, 0, 20],
+ }),
+ );
+ });
+});
diff --git a/web/src/utils/dashboardTrafficInsights.ts
b/web/src/utils/dashboardTrafficInsights.ts
new file mode 100644
index 000000000..a73ef436d
--- /dev/null
+++ b/web/src/utils/dashboardTrafficInsights.ts
@@ -0,0 +1,400 @@
+/*
+ * 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 { ClusterOverview, DashboardData } from '../api/metrics';
+
+export type TrafficHealthLevel = 'healthy' | 'notice' | 'warning' | 'critical';
+
+export type TrafficTrendDirection = 'rising' | 'falling' | 'stable' |
'unknown';
+
+export type DashboardTrafficIssueCode =
+ | 'NO_ACTIVE_TRAFFIC'
+ | 'TRAFFIC_CONCENTRATION'
+ | 'UNHEALTHY_TRAFFIC'
+ | 'BROKER_LOAD_SKEW'
+ | 'RECENT_TRAFFIC_DROP'
+ | 'RECENT_TRAFFIC_SPIKE'
+ | 'TOPOLOGY_COUNT_UNAVAILABLE'
+ | 'IDLE_CLUSTER';
+
+export interface DashboardTrafficIssue {
+ code: DashboardTrafficIssueCode;
+ level: TrafficHealthLevel;
+ clusterId?: string;
+ clusterName?: string;
+ value?: number;
+ threshold?: number;
+}
+
+export interface DashboardTrafficClusterInsight {
+ id: string;
+ name: string;
+ status: string;
+ tpsIn: number;
+ tpsOut: number;
+ totalTps: number;
+ sharePercent: number;
+ brokers: number;
+ topics: number;
+ groups: number;
+ proxies: number | null;
+ perBrokerTps: number;
+ inOutRatio: number | null;
+ trendDirection: TrafficTrendDirection;
+ trendDeltaPercent: number | null;
+ throughput: number[];
+ issues: DashboardTrafficIssue[];
+}
+
+export interface DashboardTrafficInsights {
+ level: TrafficHealthLevel;
+ totalTps: number;
+ activeClusterCount: number;
+ totalClusterCount: number;
+ unhealthyClusterCount: number;
+ unhealthyTrafficTps: number;
+ unhealthyTrafficPercent: number;
+ topCluster: DashboardTrafficClusterInsight | null;
+ topClusterSharePercent: number;
+ balanceScore: number;
+ averagePerBrokerTps: number;
+ trendDirection: TrafficTrendDirection;
+ trendDeltaPercent: number | null;
+ aggregateThroughput: number[];
+ rows: DashboardTrafficClusterInsight[];
+ issues: DashboardTrafficIssue[];
+}
+
+const CONCENTRATION_WARNING_PERCENT = 60;
+const CONCENTRATION_CRITICAL_PERCENT = 75;
+const UNHEALTHY_TRAFFIC_WARNING_PERCENT = 10;
+const UNHEALTHY_TRAFFIC_CRITICAL_PERCENT = 25;
+const BROKER_SKEW_MULTIPLIER = 2.5;
+const MIN_TRAFFIC_FOR_SKEW = 100;
+const TRAFFIC_DROP_WARNING_PERCENT = -30;
+const TRAFFIC_SPIKE_NOTICE_PERCENT = 50;
+const TREND_THRESHOLD_PERCENT = 10;
+
+const issueLevelWeight: Record<TrafficHealthLevel, number> = {
+ healthy: 0,
+ notice: 1,
+ warning: 2,
+ critical: 3,
+};
+
+const isHealthyStatus = (status?: string | null) => {
+ const normalized = (status ?? '').trim().toLowerCase();
+ return normalized === 'healthy' || normalized === 'running';
+};
+
+const finiteNumber = (value: number | null | undefined): number =>
+ typeof value === 'number' && Number.isFinite(value) ? value : 0;
+
+const nonNegativeNumber = (value: number | null | undefined): number =>
+ Math.max(0, finiteNumber(value));
+
+const sanitizeSeries = (values: number[] | null | undefined): number[] =>
+ (values ?? []).map(nonNegativeNumber);
+
+const round = (value: number, precision = 1): number => {
+ const factor = 10 ** precision;
+ return Math.round(value * factor) / factor;
+};
+
+const percent = (value: number, total: number): number => {
+ if (total <= 0) return 0;
+ return round((value / total) * 100);
+};
+
+const average = (values: number[]): number => {
+ if (values.length === 0) return 0;
+ return values.reduce((sum, value) => sum + value, 0) / values.length;
+};
+
+const maxIssueLevel = (issues: DashboardTrafficIssue[]): TrafficHealthLevel =>
+ issues.reduce<TrafficHealthLevel>(
+ (current, issue) =>
+ issueLevelWeight[issue.level] > issueLevelWeight[current] ? issue.level
: current,
+ 'healthy',
+ );
+
+const compareTrend = (
+ series: number[],
+): {
+ direction: TrafficTrendDirection;
+ deltaPercent: number | null;
+} => {
+ if (series.length < 4) {
+ return { direction: 'unknown', deltaPercent: null };
+ }
+
+ const midpoint = Math.floor(series.length / 2);
+ const earlyAverage = average(series.slice(0, midpoint));
+ const lateAverage = average(series.slice(midpoint));
+ if (earlyAverage <= 0 && lateAverage <= 0) {
+ return { direction: 'stable', deltaPercent: 0 };
+ }
+ if (earlyAverage <= 0) {
+ return { direction: 'rising', deltaPercent: 100 };
+ }
+
+ const deltaPercent = round(((lateAverage - earlyAverage) / earlyAverage) *
100);
+ if (deltaPercent >= TREND_THRESHOLD_PERCENT) {
+ return { direction: 'rising', deltaPercent };
+ }
+ if (deltaPercent <= -TREND_THRESHOLD_PERCENT) {
+ return { direction: 'falling', deltaPercent };
+ }
+ return { direction: 'stable', deltaPercent };
+};
+
+const aggregateThroughput = (clusters: ClusterOverview[]): number[] => {
+ const maxLength = Math.max(0, ...clusters.map((cluster) =>
cluster.throughput?.length ?? 0));
+ return Array.from({ length: maxLength }, (_, index) =>
+ clusters.reduce((sum, cluster) => sum +
nonNegativeNumber(cluster.throughput?.[index]), 0),
+ );
+};
+
+const calculateBalanceScore = (rows: DashboardTrafficClusterInsight[]): number
=> {
+ const activeRows = rows.filter((row) => row.totalTps > 0);
+ if (activeRows.length <= 1) return 100;
+
+ const hhi = activeRows.reduce((sum, row) => {
+ const share = row.sharePercent / 100;
+ return sum + share * share;
+ }, 0);
+ const idealHhi = 1 / activeRows.length;
+ const normalized = (hhi - idealHhi) / (1 - idealHhi);
+ return Math.max(0, Math.min(100, round((1 - normalized) * 100, 0)));
+};
+
+const buildClusterIssues = (
+ cluster: ClusterOverview,
+ totalTps: number,
+ clusterTps: number,
+): DashboardTrafficIssue[] => {
+ const issues: DashboardTrafficIssue[] = [];
+ if (clusterTps === 0) {
+ issues.push({
+ code: 'IDLE_CLUSTER',
+ level: 'notice',
+ clusterId: cluster.id,
+ clusterName: cluster.name,
+ });
+ }
+ if (!isHealthyStatus(cluster.status) && clusterTps > 0) {
+ issues.push({
+ code: 'UNHEALTHY_TRAFFIC',
+ level:
+ percent(clusterTps, totalTps) >= UNHEALTHY_TRAFFIC_CRITICAL_PERCENT
+ ? 'critical'
+ : 'warning',
+ clusterId: cluster.id,
+ clusterName: cluster.name,
+ value: percent(clusterTps, totalTps),
+ threshold: UNHEALTHY_TRAFFIC_WARNING_PERCENT,
+ });
+ }
+ if (cluster.proxies == null) {
+ issues.push({
+ code: 'TOPOLOGY_COUNT_UNAVAILABLE',
+ level: 'notice',
+ clusterId: cluster.id,
+ clusterName: cluster.name,
+ });
+ }
+ return issues;
+};
+
+const buildRows = (
+ clusters: ClusterOverview[],
+ totalTps: number,
+): DashboardTrafficClusterInsight[] =>
+ clusters
+ .map((cluster) => {
+ const tpsIn = nonNegativeNumber(cluster.tpsIn);
+ const tpsOut = nonNegativeNumber(cluster.tpsOut);
+ const clusterTps = tpsIn + tpsOut;
+ const brokers = Math.max(0,
Math.floor(nonNegativeNumber(cluster.brokers)));
+ const throughput = sanitizeSeries(cluster.throughput);
+ const trend = compareTrend(throughput);
+ return {
+ id: cluster.id,
+ name: cluster.name,
+ status: cluster.status,
+ tpsIn,
+ tpsOut,
+ totalTps: clusterTps,
+ sharePercent: percent(clusterTps, totalTps),
+ brokers,
+ topics: Math.max(0, Math.floor(nonNegativeNumber(cluster.topics))),
+ groups: Math.max(0, Math.floor(nonNegativeNumber(cluster.groups))),
+ proxies:
+ cluster.proxies == null
+ ? null
+ : Math.max(0, Math.floor(nonNegativeNumber(cluster.proxies))),
+ perBrokerTps: brokers > 0 ? round(clusterTps / brokers) : 0,
+ inOutRatio: tpsIn > 0 ? round(tpsOut / tpsIn, 2) : null,
+ trendDirection: trend.direction,
+ trendDeltaPercent: trend.deltaPercent,
+ throughput,
+ issues: buildClusterIssues(cluster, totalTps, clusterTps),
+ };
+ })
+ .sort((left, right) => right.totalTps - left.totalTps ||
left.name.localeCompare(right.name));
+
+const findBrokerSkewIssue = (
+ rows: DashboardTrafficClusterInsight[],
+ averagePerBrokerTps: number,
+): DashboardTrafficIssue | null => {
+ if (averagePerBrokerTps <= 0) return null;
+ const skewed = rows.find(
+ (row) =>
+ row.totalTps >= MIN_TRAFFIC_FOR_SKEW &&
+ row.perBrokerTps >= averagePerBrokerTps * BROKER_SKEW_MULTIPLIER,
+ );
+ if (!skewed) return null;
+ return {
+ code: 'BROKER_LOAD_SKEW',
+ level: 'warning',
+ clusterId: skewed.id,
+ clusterName: skewed.name,
+ value: skewed.perBrokerTps,
+ threshold: round(averagePerBrokerTps * BROKER_SKEW_MULTIPLIER),
+ };
+};
+
+const buildGlobalIssues = (
+ rows: DashboardTrafficClusterInsight[],
+ topCluster: DashboardTrafficClusterInsight | null,
+ unhealthyTrafficPercent: number,
+ trend: ReturnType<typeof compareTrend>,
+ averagePerBrokerTps: number,
+): DashboardTrafficIssue[] => {
+ const issues: DashboardTrafficIssue[] = [];
+ const activeRows = rows.filter((row) => row.totalTps > 0);
+
+ if (activeRows.length === 0) {
+ issues.push({ code: 'NO_ACTIVE_TRAFFIC', level: 'notice' });
+ }
+
+ if (
+ topCluster &&
+ activeRows.length > 1 &&
+ topCluster.sharePercent >= CONCENTRATION_WARNING_PERCENT
+ ) {
+ issues.push({
+ code: 'TRAFFIC_CONCENTRATION',
+ level: topCluster.sharePercent >= CONCENTRATION_CRITICAL_PERCENT ?
'critical' : 'warning',
+ clusterId: topCluster.id,
+ clusterName: topCluster.name,
+ value: topCluster.sharePercent,
+ threshold: CONCENTRATION_WARNING_PERCENT,
+ });
+ }
+
+ if (unhealthyTrafficPercent >= UNHEALTHY_TRAFFIC_WARNING_PERCENT) {
+ issues.push({
+ code: 'UNHEALTHY_TRAFFIC',
+ level: unhealthyTrafficPercent >= UNHEALTHY_TRAFFIC_CRITICAL_PERCENT ?
'critical' : 'warning',
+ value: unhealthyTrafficPercent,
+ threshold: UNHEALTHY_TRAFFIC_WARNING_PERCENT,
+ });
+ }
+
+ const skewIssue = findBrokerSkewIssue(rows, averagePerBrokerTps);
+ if (skewIssue) issues.push(skewIssue);
+
+ if (
+ trend.direction === 'falling' &&
+ trend.deltaPercent != null &&
+ trend.deltaPercent <= TRAFFIC_DROP_WARNING_PERCENT
+ ) {
+ issues.push({
+ code: 'RECENT_TRAFFIC_DROP',
+ level: 'warning',
+ value: trend.deltaPercent,
+ threshold: TRAFFIC_DROP_WARNING_PERCENT,
+ });
+ }
+
+ if (
+ trend.direction === 'rising' &&
+ trend.deltaPercent != null &&
+ trend.deltaPercent >= TRAFFIC_SPIKE_NOTICE_PERCENT
+ ) {
+ issues.push({
+ code: 'RECENT_TRAFFIC_SPIKE',
+ level: 'notice',
+ value: trend.deltaPercent,
+ threshold: TRAFFIC_SPIKE_NOTICE_PERCENT,
+ });
+ }
+
+ return issues;
+};
+
+export function buildDashboardTrafficInsights(
+ dashboard: DashboardData | null | undefined,
+): DashboardTrafficInsights {
+ const clusters = dashboard?.clusters ?? [];
+ const totalTps = clusters.reduce(
+ (sum, cluster) => sum + nonNegativeNumber(cluster.tpsIn) +
nonNegativeNumber(cluster.tpsOut),
+ 0,
+ );
+ const rows = buildRows(clusters, totalTps);
+ const activeClusterCount = rows.filter((row) => row.totalTps > 0).length;
+ const unhealthyClusterCount = rows.filter((row) =>
!isHealthyStatus(row.status)).length;
+ const unhealthyTrafficTps = rows
+ .filter((row) => !isHealthyStatus(row.status))
+ .reduce((sum, row) => sum + row.totalTps, 0);
+ const totalBrokers = rows.reduce((sum, row) => sum + row.brokers, 0);
+ const averagePerBrokerTps = totalBrokers > 0 ? round(totalTps /
totalBrokers) : 0;
+ const topCluster = rows[0] ?? null;
+ const aggregateSeries = aggregateThroughput(clusters);
+ const trend = compareTrend(aggregateSeries);
+ const unhealthyTrafficPercent = percent(unhealthyTrafficTps, totalTps);
+ const issues = [
+ ...buildGlobalIssues(rows, topCluster, unhealthyTrafficPercent, trend,
averagePerBrokerTps),
+ ...rows.flatMap((row) => row.issues),
+ ];
+
+ return {
+ level: maxIssueLevel(issues),
+ totalTps,
+ activeClusterCount,
+ totalClusterCount: rows.length,
+ unhealthyClusterCount,
+ unhealthyTrafficTps,
+ unhealthyTrafficPercent,
+ topCluster,
+ topClusterSharePercent: topCluster?.sharePercent ?? 0,
+ balanceScore: calculateBalanceScore(rows),
+ averagePerBrokerTps,
+ trendDirection: trend.direction,
+ trendDeltaPercent: trend.deltaPercent,
+ aggregateThroughput: aggregateSeries,
+ rows,
+ issues,
+ };
+}
+
+export const formatTrafficTps = (value: number) =>
+ value.toLocaleString(undefined, { maximumFractionDigits: value >= 100 ? 0 :
1 });
+
+export const formatTrafficPercent = (value: number) =>
+ `${value.toLocaleString(undefined, { maximumFractionDigits: 1 })}%`;
