diff --git a/go/vehicle-gateway/internal/capacity/check.go b/go/vehicle-gateway/internal/capacity/check.go index e035a1fb..991f884a 100644 --- a/go/vehicle-gateway/internal/capacity/check.go +++ b/go/vehicle-gateway/internal/capacity/check.go @@ -25,8 +25,16 @@ type Report struct { } type Totals struct { - ActiveConnections int64 `json:"active_connections"` - KafkaLag float64 `json:"kafka_lag"` + ActiveConnections int64 `json:"active_connections"` + KafkaLag float64 `json:"kafka_lag"` + BridgeConsumerPending float64 `json:"bridge_consumer_pending"` + BridgeAckPending float64 `json:"bridge_ack_pending"` + BridgeBatchPendingMessages float64 `json:"bridge_batch_pending_messages"` + FastWriterConsumerPending float64 `json:"fast_writer_consumer_pending"` + FastWriterAckPending float64 `json:"fast_writer_ack_pending"` + FastWriterBatchPending float64 `json:"fast_writer_batch_pending_messages"` + HistoryBatchPending float64 `json:"history_batch_pending_messages"` + HistoryRowsPending float64 `json:"history_rows_pending"` } type ServiceCheck struct { @@ -52,6 +60,14 @@ func Evaluate(metricsByService map[string]string) Report { report.Totals.KafkaLag += metrics["vehicle_history_kafka_lag"] report.Totals.KafkaLag += metrics["vehicle_stat_kafka_lag"] report.Totals.KafkaLag += metrics["vehicle_realtime_kafka_lag"] + report.Totals.BridgeConsumerPending += metrics["vehicle_bridge_nats_consumer_pending"] + report.Totals.BridgeAckPending += metrics["vehicle_bridge_nats_consumer_ack_pending"] + report.Totals.BridgeBatchPendingMessages += metrics["vehicle_bridge_batch_pending_messages"] + report.Totals.FastWriterConsumerPending += metrics["vehicle_fast_writer_nats_consumer_pending"] + report.Totals.FastWriterAckPending += metrics["vehicle_fast_writer_nats_consumer_ack_pending"] + report.Totals.FastWriterBatchPending += metrics["vehicle_fast_writer_batch_pending_messages"] + report.Totals.HistoryBatchPending += metrics["vehicle_history_batch_pending_messages"] + report.Totals.HistoryRowsPending += metrics["vehicle_history_batch_pending_rows"] report.Findings = append(report.Findings, findingsForMetrics(metrics)...) } if len(report.Findings) > 0 { diff --git a/go/vehicle-gateway/internal/capacity/check_test.go b/go/vehicle-gateway/internal/capacity/check_test.go index 0f501b21..32553fde 100644 --- a/go/vehicle-gateway/internal/capacity/check_test.go +++ b/go/vehicle-gateway/internal/capacity/check_test.go @@ -41,6 +41,7 @@ vehicle_bridge_batch_pending_messages 1001`, vehicle_fast_writer_nats_consumer_pending{consumer="vehicle-fast-writer",stream="VEHICLE_INGEST"} 12001 vehicle_fast_writer_batch_pending_messages 9`, "history": `vehicle_history_batch_pending_messages 6 +vehicle_history_batch_pending_rows 7 vehicle_history_kafka_lag{topic="vehicle.raw.go.jt808.v1",partition="0"} 42`, }) @@ -63,4 +64,22 @@ vehicle_history_kafka_lag{topic="vehicle.raw.go.jt808.v1",partition="0"} 42`, t.Fatalf("finding missing %q in:\n%s", want, joined) } } + if report.Totals.BridgeAckPending != 101 { + t.Fatalf("bridge ack pending = %v, want 101", report.Totals.BridgeAckPending) + } + if report.Totals.BridgeConsumerPending != 12001 { + t.Fatalf("bridge consumer pending = %v, want 12001", report.Totals.BridgeConsumerPending) + } + if report.Totals.FastWriterAckPending != 11 { + t.Fatalf("fast writer ack pending = %v, want 11", report.Totals.FastWriterAckPending) + } + if report.Totals.FastWriterBatchPending != 9 { + t.Fatalf("fast writer batch pending = %v, want 9", report.Totals.FastWriterBatchPending) + } + if report.Totals.HistoryBatchPending != 6 { + t.Fatalf("history batch pending = %v, want 6", report.Totals.HistoryBatchPending) + } + if report.Totals.HistoryRowsPending != 7 { + t.Fatalf("history rows pending = %v, want 7", report.Totals.HistoryRowsPending) + } } diff --git a/vehicle-data-platform/apps/api/internal/platform/capacity_check.go b/vehicle-data-platform/apps/api/internal/platform/capacity_check.go index 6decbe22..a2f1e355 100644 --- a/vehicle-data-platform/apps/api/internal/platform/capacity_check.go +++ b/vehicle-data-platform/apps/api/internal/platform/capacity_check.go @@ -9,17 +9,24 @@ import ( ) type CapacityCheckResult struct { - Status string `json:"status"` - ActiveConnections int64 `json:"active_connections"` - KafkaLag float64 `json:"kafka_lag"` - Findings []string `json:"findings"` + Status string `json:"status"` + CapacityMetrics CapacityMetrics `json:"capacity_metrics"` + Findings []string `json:"findings"` } type capacityCheckReport struct { Status string `json:"status"` Totals struct { - ActiveConnections int64 `json:"active_connections"` - KafkaLag float64 `json:"kafka_lag"` + ActiveConnections int64 `json:"active_connections"` + KafkaLag float64 `json:"kafka_lag"` + BridgeConsumerPending float64 `json:"bridge_consumer_pending"` + BridgeAckPending float64 `json:"bridge_ack_pending"` + BridgeBatchPendingMessages float64 `json:"bridge_batch_pending_messages"` + FastWriterConsumerPending float64 `json:"fast_writer_consumer_pending"` + FastWriterAckPending float64 `json:"fast_writer_ack_pending"` + FastWriterBatchPending float64 `json:"fast_writer_batch_pending_messages"` + HistoryBatchPending float64 `json:"history_batch_pending_messages"` + HistoryRowsPending float64 `json:"history_rows_pending"` } `json:"totals"` Findings []string `json:"findings"` } @@ -45,9 +52,19 @@ func (c *CapacityCheckCommand) CheckCapacity(ctx context.Context) (CapacityCheck return CapacityCheckResult{}, jsonErr } return CapacityCheckResult{ - Status: report.Status, - ActiveConnections: report.Totals.ActiveConnections, - KafkaLag: report.Totals.KafkaLag, - Findings: report.Findings, + Status: report.Status, + CapacityMetrics: CapacityMetrics{ + ActiveConnections: int(report.Totals.ActiveConnections), + KafkaLag: int(report.Totals.KafkaLag), + BridgeConsumerPending: int(report.Totals.BridgeConsumerPending), + BridgeAckPending: int(report.Totals.BridgeAckPending), + BridgeBatchPendingMessages: int(report.Totals.BridgeBatchPendingMessages), + FastWriterConsumerPending: int(report.Totals.FastWriterConsumerPending), + FastWriterAckPending: int(report.Totals.FastWriterAckPending), + FastWriterBatchPending: int(report.Totals.FastWriterBatchPending), + HistoryBatchPending: int(report.Totals.HistoryBatchPending), + HistoryRowsPending: int(report.Totals.HistoryRowsPending), + }, + Findings: report.Findings, }, nil } diff --git a/vehicle-data-platform/apps/api/internal/platform/mock_store.go b/vehicle-data-platform/apps/api/internal/platform/mock_store.go index 7f1f2c42..384d4d88 100644 --- a/vehicle-data-platform/apps/api/internal/platform/mock_store.go +++ b/vehicle-data-platform/apps/api/internal/platform/mock_store.go @@ -672,12 +672,27 @@ func bucketStats(values map[string]int) []QualityBucketStat { func (m *MockStore) OpsHealth(context.Context) (OpsHealth, error) { summary, _ := m.DashboardSummary(context.Background()) redisOnlineKeys := 92 + activeConnections := 120000 + capacityMetrics := CapacityMetrics{ + ActiveConnections: activeConnections, + KafkaLag: 0, + BridgeConsumerPending: 0, + BridgeAckPending: 0, + BridgeBatchPendingMessages: 0, + FastWriterConsumerPending: 0, + FastWriterAckPending: 0, + FastWriterBatchPending: 0, + HistoryBatchPending: 0, + HistoryRowsPending: 0, + } return OpsHealth{ - LinkHealth: summary.LinkHealth, - KafkaLag: summary.KafkaLag, - RedisOnlineKeys: &redisOnlineKeys, - TDengineWritable: true, - MySQLWritable: true, + LinkHealth: summary.LinkHealth, + KafkaLag: summary.KafkaLag, + ActiveConnections: &activeConnections, + CapacityMetrics: capacityMetrics, + RedisOnlineKeys: &redisOnlineKeys, + TDengineWritable: true, + MySQLWritable: true, }, nil } diff --git a/vehicle-data-platform/apps/api/internal/platform/model.go b/vehicle-data-platform/apps/api/internal/platform/model.go index fcf1704b..d0f8e406 100644 --- a/vehicle-data-platform/apps/api/internal/platform/model.go +++ b/vehicle-data-platform/apps/api/internal/platform/model.go @@ -359,14 +359,28 @@ type QualityPriorityIssue struct { } type OpsHealth struct { - LinkHealth []LinkHealth `json:"linkHealth"` - KafkaLag *int `json:"kafkaLag"` - ActiveConnections *int `json:"activeConnections"` - CapacityFindings []string `json:"capacityFindings"` - RedisOnlineKeys *int `json:"redisOnlineKeys"` - TDengineWritable bool `json:"tdengineWritable"` - MySQLWritable bool `json:"mysqlWritable"` - Runtime RuntimeInfo `json:"runtime"` + LinkHealth []LinkHealth `json:"linkHealth"` + KafkaLag *int `json:"kafkaLag"` + ActiveConnections *int `json:"activeConnections"` + CapacityMetrics CapacityMetrics `json:"capacityMetrics"` + CapacityFindings []string `json:"capacityFindings"` + RedisOnlineKeys *int `json:"redisOnlineKeys"` + TDengineWritable bool `json:"tdengineWritable"` + MySQLWritable bool `json:"mysqlWritable"` + Runtime RuntimeInfo `json:"runtime"` +} + +type CapacityMetrics struct { + ActiveConnections int `json:"activeConnections"` + KafkaLag int `json:"kafkaLag"` + BridgeConsumerPending int `json:"bridgeConsumerPending"` + BridgeAckPending int `json:"bridgeAckPending"` + BridgeBatchPendingMessages int `json:"bridgeBatchPendingMessages"` + FastWriterConsumerPending int `json:"fastWriterConsumerPending"` + FastWriterAckPending int `json:"fastWriterAckPending"` + FastWriterBatchPending int `json:"fastWriterBatchPending"` + HistoryBatchPending int `json:"historyBatchPending"` + HistoryRowsPending int `json:"historyRowsPending"` } type RuntimeInfo struct { diff --git a/vehicle-data-platform/apps/api/internal/platform/production_store.go b/vehicle-data-platform/apps/api/internal/platform/production_store.go index aa298572..6600e706 100644 --- a/vehicle-data-platform/apps/api/internal/platform/production_store.go +++ b/vehicle-data-platform/apps/api/internal/platform/production_store.go @@ -929,7 +929,7 @@ func (s *ProductionStore) OpsHealth(ctx context.Context) (OpsHealth, error) { locationHealth := s.mysqlTableReadHealth(ctx, "vehicle_realtime_location", "读取实时位置表正常") tdengineHealth := s.tdengineRawFrameHealth(ctx) redisHealth, redisOnlineKeys := s.redisOnlineKeyHealth(ctx) - capacityHealth, kafkaLag, activeConnections, capacityFindings := s.capacityCheckHealth(ctx) + capacityHealth, kafkaLag, activeConnections, capacityMetrics, capacityFindings := s.capacityCheckHealth(ctx) mysqlWritable := mysqlStatus == "ok" && snapshotHealth.Status == "ok" && locationHealth.Status == "ok" return OpsHealth{ LinkHealth: []LinkHealth{ @@ -942,6 +942,7 @@ func (s *ProductionStore) OpsHealth(ctx context.Context) (OpsHealth, error) { }, KafkaLag: kafkaLag, ActiveConnections: activeConnections, + CapacityMetrics: capacityMetrics, CapacityFindings: capacityFindings, RedisOnlineKeys: redisOnlineKeys, TDengineWritable: tdengineHealth.Status == "ok", @@ -949,19 +950,20 @@ func (s *ProductionStore) OpsHealth(ctx context.Context) (OpsHealth, error) { }, nil } -func (s *ProductionStore) capacityCheckHealth(ctx context.Context) (LinkHealth, *int, *int, []string) { +func (s *ProductionStore) capacityCheckHealth(ctx context.Context) (LinkHealth, *int, *int, CapacityMetrics, []string) { health := LinkHealth{Name: "Capacity check", Status: "warning", Detail: "平台暂未接入 capacity-check"} if s.capacityCheck == nil { - return health, nil, nil, []string{} + return health, nil, nil, CapacityMetrics{}, []string{} } result, err := s.capacityCheck.CheckCapacity(ctx) if err != nil { health.Status = "error" health.Detail = err.Error() - return health, nil, nil, []string{} + return health, nil, nil, CapacityMetrics{}, []string{} } - kafkaLag := int(result.KafkaLag) - activeConnections := int(result.ActiveConnections) + capacityMetrics := result.CapacityMetrics + kafkaLag := capacityMetrics.KafkaLag + activeConnections := capacityMetrics.ActiveConnections detail := "active_connections=" + strconv.Itoa(activeConnections) + ", kafka_lag=" + strconv.Itoa(kafkaLag) if len(result.Findings) > 0 { detail += ", findings=" + strings.Join(result.Findings, "; ") @@ -972,7 +974,7 @@ func (s *ProductionStore) capacityCheckHealth(ctx context.Context) (LinkHealth, health.Status = "warning" } health.Detail = detail - return health, &kafkaLag, &activeConnections, append([]string{}, result.Findings...) + return health, &kafkaLag, &activeConnections, capacityMetrics, append([]string{}, result.Findings...) } func (s *ProductionStore) redisOnlineKeyHealth(ctx context.Context) (LinkHealth, *int) { diff --git a/vehicle-data-platform/apps/api/internal/platform/production_store_test.go b/vehicle-data-platform/apps/api/internal/platform/production_store_test.go index 732c1c53..62764e04 100644 --- a/vehicle-data-platform/apps/api/internal/platform/production_store_test.go +++ b/vehicle-data-platform/apps/api/internal/platform/production_store_test.go @@ -213,10 +213,14 @@ func TestOpsHealthUsesCapacityCheckProbeForKafkaLag(t *testing.T) { store := &ProductionStore{ db: db, capacityCheck: fakeCapacityCheckProbe{result: CapacityCheckResult{ - Status: "degraded", - ActiveConnections: 120000, - KafkaLag: 42, - Findings: []string{"kafka lag 42"}, + Status: "degraded", + CapacityMetrics: CapacityMetrics{ + ActiveConnections: 120000, + KafkaLag: 42, + BridgeConsumerPending: 1024, + BridgeAckPending: 2, + }, + Findings: []string{"kafka lag 42"}, }}, } @@ -230,6 +234,9 @@ func TestOpsHealthUsesCapacityCheckProbeForKafkaLag(t *testing.T) { if health.ActiveConnections == nil || *health.ActiveConnections != 120000 { t.Fatalf("OpsHealth should expose capacity-check active connections, got %+v", health.ActiveConnections) } + if health.CapacityMetrics.BridgeConsumerPending != 1024 || health.CapacityMetrics.BridgeAckPending != 2 { + t.Fatalf("OpsHealth should expose structured capacity metrics, got %+v", health.CapacityMetrics) + } if len(health.CapacityFindings) != 1 || health.CapacityFindings[0] != "kafka lag 42" { t.Fatalf("OpsHealth should expose structured capacity findings, got %+v", health.CapacityFindings) } @@ -254,9 +261,11 @@ func TestOpsHealthReturnsEmptyCapacityFindingsArray(t *testing.T) { store := &ProductionStore{ db: db, capacityCheck: fakeCapacityCheckProbe{result: CapacityCheckResult{ - Status: "ok", - ActiveConnections: 254, - KafkaLag: 0, + Status: "ok", + CapacityMetrics: CapacityMetrics{ + ActiveConnections: 254, + KafkaLag: 0, + }, }}, } diff --git a/vehicle-data-platform/apps/web/src/api/types.ts b/vehicle-data-platform/apps/web/src/api/types.ts index 4835674f..feb4b640 100644 --- a/vehicle-data-platform/apps/web/src/api/types.ts +++ b/vehicle-data-platform/apps/web/src/api/types.ts @@ -351,6 +351,7 @@ export interface OpsHealth { linkHealth: LinkHealth[]; kafkaLag: number | null; activeConnections: number | null; + capacityMetrics?: CapacityMetrics; capacityFindings: string[]; redisOnlineKeys: number | null; tdengineWritable: boolean; @@ -358,6 +359,19 @@ export interface OpsHealth { runtime: RuntimeInfo; } +export interface CapacityMetrics { + activeConnections: number; + kafkaLag: number; + bridgeConsumerPending: number; + bridgeAckPending: number; + bridgeBatchPendingMessages: number; + fastWriterConsumerPending: number; + fastWriterAckPending: number; + fastWriterBatchPending: number; + historyBatchPending: number; + historyRowsPending: number; +} + export interface RuntimeInfo { requestTimeoutMs: number; amapWebJsConfigured?: boolean; diff --git a/vehicle-data-platform/apps/web/src/pages/OpsQuality.tsx b/vehicle-data-platform/apps/web/src/pages/OpsQuality.tsx index 6828a991..64fb4774 100644 --- a/vehicle-data-platform/apps/web/src/pages/OpsQuality.tsx +++ b/vehicle-data-platform/apps/web/src/pages/OpsQuality.tsx @@ -1,7 +1,7 @@ import { Button, Card, Space, Table, Tag, Toast, Typography } from '@douyinfe/semi-ui'; import { useEffect, useState } from 'react'; import { api } from '../api/client'; -import type { LinkHealth, OpsHealth } from '../api/types'; +import type { CapacityMetrics, LinkHealth, OpsHealth } from '../api/types'; import { PageHeader } from '../components/PageHeader'; const statusColor: Record = { @@ -47,6 +47,112 @@ type CapacityRisk = { action: string; }; +type CapacityMetricRow = { + key: keyof CapacityMetrics; + name: string; + value: number; + threshold: string; + status: 'ok' | 'warning' | 'error'; + detail: string; +}; + +const emptyCapacityMetrics: CapacityMetrics = { + activeConnections: 0, + kafkaLag: 0, + bridgeConsumerPending: 0, + bridgeAckPending: 0, + bridgeBatchPendingMessages: 0, + fastWriterConsumerPending: 0, + fastWriterAckPending: 0, + fastWriterBatchPending: 0, + historyBatchPending: 0, + historyRowsPending: 0 +}; + +function getCapacityMetrics(health: OpsHealth | null): CapacityMetrics { + return { + ...emptyCapacityMetrics, + ...(health?.capacityMetrics ?? {}), + activeConnections: health?.capacityMetrics?.activeConnections ?? health?.activeConnections ?? 0, + kafkaLag: health?.capacityMetrics?.kafkaLag ?? health?.kafkaLag ?? 0 + }; +} + +function metricStatus(value: number, warningAt: number, errorAt?: number): 'ok' | 'warning' | 'error' { + if (errorAt != null && value > errorAt) return 'error'; + return value > warningAt ? 'warning' : 'ok'; +} + +function capacityMetricRows(health: OpsHealth | null): CapacityMetricRow[] { + const metrics = getCapacityMetrics(health); + return [ + { + key: 'bridgeConsumerPending', + name: 'NATS 桥接待消费', + value: metrics.bridgeConsumerPending, + threshold: '> 10,000 预警', + status: metricStatus(metrics.bridgeConsumerPending, 10000), + detail: 'RAW/FIELDS 从 NATS 进入 Kafka 前的待消费量。' + }, + { + key: 'bridgeAckPending', + name: 'NATS 桥接 ACK', + value: metrics.bridgeAckPending, + threshold: '> 100 严重', + status: metricStatus(metrics.bridgeAckPending, 0, 100), + detail: '桥接消费者已拉取但未确认的消息，持续不为 0 需要优先排查。' + }, + { + key: 'fastWriterConsumerPending', + name: '快速写入待消费', + value: metrics.fastWriterConsumerPending, + threshold: '> 10,000 预警', + status: metricStatus(metrics.fastWriterConsumerPending, 10000), + detail: '实时数据写 TDengine、Redis 前的 NATS 待消费量。' + }, + { + key: 'fastWriterAckPending', + name: '快速写入 ACK', + value: metrics.fastWriterAckPending, + threshold: '> 10 严重', + status: metricStatus(metrics.fastWriterAckPending, 0, 10), + detail: '快速写入消费者已拉取但未确认的消息。' + }, + { + key: 'historyBatchPending', + name: '历史批次待写', + value: metrics.historyBatchPending, + threshold: '> 0 预警', + status: metricStatus(metrics.historyBatchPending, 0), + detail: '历史落库批处理内存队列，正常应快速归零。' + }, + { + key: 'historyRowsPending', + name: '历史行待写', + value: metrics.historyRowsPending, + threshold: '> 0 预警', + status: metricStatus(metrics.historyRowsPending, 0), + detail: '历史落库批处理待写行数，用于判断 TDengine/MySQL 写入是否滞后。' + }, + { + key: 'kafkaLag', + name: 'Kafka 消费 Lag', + value: metrics.kafkaLag, + threshold: '> 0 预警', + status: metricStatus(metrics.kafkaLag, 0), + detail: '后续统计和历史消费链路的 Kafka 积压。' + }, + { + key: 'activeConnections', + name: '网关活跃连接', + value: metrics.activeConnections, + threshold: '> 100,000 预警', + status: metricStatus(metrics.activeConnections, 100000), + detail: '32960、808、MQTT 当前活跃连接总量。' + } + ]; +} + function firstNumber(value: string) { const match = value.match(/[\d,.]+/); return match ? match[0] : '-'; @@ -138,6 +244,7 @@ export function OpsQuality() { const runtime = health?.runtime; const capacityRisks = (health?.capacityFindings ?? []).map(capacityRisk); + const capacityRows = capacityMetricRows(health); const bridgeRisks = capacityRisks.filter((item) => item.category === 'NATS -> Kafka'); return ( @@ -196,6 +303,31 @@ export function OpsQuality() { + + + pagination={false} + dataSource={capacityRows} + rowKey={(row?: CapacityMetricRow) => row?.key ?? ''} + columns={[ + { title: '指标', dataIndex: 'name' }, + { + title: '当前值', + width: 140, + render: (_: unknown, row: CapacityMetricRow) => ( + {formatNumber(row.value)} + ) + }, + { + title: '状态', + width: 120, + render: (_: unknown, row: CapacityMetricRow) => {row.status} + }, + { title: '阈值', width: 150, dataIndex: 'threshold' }, + { title: '说明', dataIndex: 'detail' } + ]} + /> + + {capacityRisks.length ? (
diff --git a/vehicle-data-platform/apps/web/src/test/App.test.tsx b/vehicle-data-platform/apps/web/src/test/App.test.tsx index 027e1670..c138b5ea 100644 --- a/vehicle-data-platform/apps/web/src/test/App.test.tsx +++ b/vehicle-data-platform/apps/web/src/test/App.test.tsx @@ -3374,6 +3374,18 @@ test('renders ops quality as a standalone runtime health page', async () => { ], kafkaLag: 42, activeConnections: 1234, + capacityMetrics: { + activeConnections: 1234, + kafkaLag: 42, + bridgeConsumerPending: 1482047, + bridgeAckPending: 4000, + bridgeBatchPendingMessages: 12, + fastWriterConsumerPending: 8, + fastWriterAckPending: 0, + fastWriterBatchPending: 0, + historyBatchPending: 2, + historyRowsPending: 320 + }, capacityFindings: [ 'Kafka lag 42', 'bridge ack pending 4000 exceeds 100', @@ -3413,7 +3425,7 @@ test('renders ops quality as a standalone runtime health page', async () => { expect(await screen.findByRole('heading', { name: '运维质量' })).toBeInTheDocument(); expect(screen.getByText('platform-ops-test')).toBeInTheDocument(); expect(screen.getAllByText('42').length).toBeGreaterThanOrEqual(1); - expect(screen.getByText('1,234')).toBeInTheDocument(); + expect(screen.getAllByText('1,234').length).toBeGreaterThanOrEqual(1); expect(screen.getByText('368')).toBeInTheDocument(); expect(screen.getByText('TDengine 写入')).toBeInTheDocument(); expect(screen.getByText('MySQL 写入')).toBeInTheDocument(); @@ -3421,9 +3433,14 @@ test('renders ops quality as a standalone runtime health page', async () => { expect(screen.getByText('/_AMapService')).toBeInTheDocument(); expect(screen.getByText('安全码未暴露')).toBeInTheDocument(); expect(screen.getByText('gb32960-gateway')).toBeInTheDocument(); + expect(screen.getByText('消息链路积压')).toBeInTheDocument(); + expect(screen.getByText('快速写入待消费')).toBeInTheDocument(); + expect(screen.getByText('历史行待写')).toBeInTheDocument(); + expect(screen.getByText('1,482,047')).toBeInTheDocument(); + expect(screen.getByText('4,000')).toBeInTheDocument(); expect(screen.getByText('Kafka lag 42')).toBeInTheDocument(); expect(screen.getByText('NATS 桥接 ACK 未确认')).toBeInTheDocument(); - expect(screen.getByText('NATS 桥接待消费')).toBeInTheDocument(); + expect(screen.getAllByText('NATS 桥接待消费').length).toBeGreaterThanOrEqual(1); expect(screen.getByText('桥接处置顺序')).toBeInTheDocument(); expect(fetchMock).toHaveBeenCalledWith(expect.stringContaining('/api/ops/health'), undefined); });