shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(platform): expose structured capacity metrics

Browse Source
This commit is contained in:
lingniu
2026-07-04 19:10:04 +08:00
parent 68b260c704
commit cbe7882d51
10 changed files with 297 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
go/vehicle-gateway/internal/capacity/check.go
View File

@@ -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 {

19
go/vehicle-gateway/internal/capacity/check_test.go
View File

@@ -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)
}
}

37
vehicle-data-platform/apps/api/internal/platform/capacity_check.go
View File

@@ -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
}

25
vehicle-data-platform/apps/api/internal/platform/mock_store.go
View File

@@ -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
}

30
vehicle-data-platform/apps/api/internal/platform/model.go
View File

@@ -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 {

16
vehicle-data-platform/apps/api/internal/platform/production_store.go
View File

@@ -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) {

23
vehicle-data-platform/apps/api/internal/platform/production_store_test.go
View File

@@ -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,
},
}},
}

14
vehicle-data-platform/apps/web/src/api/types.ts
View File

@@ -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;

134
vehicle-data-platform/apps/web/src/pages/OpsQuality.tsx
View File

@@ -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<string, 'green' | 'orange' | 'red' | 'grey'> = {
@@ -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() {
</Space>
</Card>
<Card bordered title="消息链路积压" loading={loading} style={{ marginTop: 16 }}>
<Table<CapacityMetricRow>
pagination={false}
dataSource={capacityRows}
rowKey={(row?: CapacityMetricRow) => row?.key ?? ''}
columns={[
{ title: '指标', dataIndex: 'name' },
{
title: '当前值',
width: 140,
render: (_: unknown, row: CapacityMetricRow) => (
<Typography.Text strong>{formatNumber(row.value)}</Typography.Text>
)
},
{
title: '状态',
width: 120,
render: (_: unknown, row: CapacityMetricRow) => <Tag color={statusColor[row.status]}>{row.status}</Tag>
},
{ title: '阈值', width: 150, dataIndex: 'threshold' },
{ title: '说明', dataIndex: 'detail' }
]}
/>
</Card>
{capacityRisks.length ? (
<Card bordered title="容量与风险发现" style={{ marginTop: 16 }}>
<div className="vp-risk-grid">

21
vehicle-data-platform/apps/web/src/test/App.test.tsx
View File

@@ -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);
});