From 3edd7462c796493c79ba5d445230298a4446965d Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 19:54:29 +0800 Subject: [PATCH] feat(go): add capacity health check tool --- docs/ops/go-service-observability.md | 8 ++ docs/ops/go-vehicle-ingest-memory.md | 3 + go/vehicle-gateway/cmd/capacity-check/main.go | 109 +++++++++++++++ .../cmd/capacity-check/main_test.go | 31 ++++ go/vehicle-gateway/internal/capacity/check.go | 132 ++++++++++++++++++ .../internal/capacity/check_test.go | 66 +++++++++ 6 files changed, 349 insertions(+) create mode 100644 go/vehicle-gateway/cmd/capacity-check/main.go create mode 100644 go/vehicle-gateway/cmd/capacity-check/main_test.go create mode 100644 go/vehicle-gateway/internal/capacity/check.go create mode 100644 go/vehicle-gateway/internal/capacity/check_test.go diff --git a/docs/ops/go-service-observability.md b/docs/ops/go-service-observability.md index 5904b77f..de64747b 100644 --- a/docs/ops/go-service-observability.md +++ b/docs/ops/go-service-observability.md @@ -29,6 +29,14 @@ curl -fsS http://127.0.0.1:20214/readyz curl -fsS http://127.0.0.1:20200/readyz ``` +本机容量健康摘要： + +```bash +/opt/lingniu-go-native/current/capacity-check +``` + +它会抓取 Gateway、History writer、Stat writer、NATS bridge、Realtime API、NATS fast writer 的本地 `/metrics`，输出 JSON。退出码 `0` 表示当前关键 backlog 和拒绝计数正常，退出码 `2` 表示存在 pending、Kafka lag、连接拒绝或 metrics 抓取失败，适合接入 cron/告警。 + ## Core Counters | Metric | Meaning | diff --git a/docs/ops/go-vehicle-ingest-memory.md b/docs/ops/go-vehicle-ingest-memory.md index af533d41..9c7d63df 100644 --- a/docs/ops/go-vehicle-ingest-memory.md +++ b/docs/ops/go-vehicle-ingest-memory.md @@ -24,6 +24,8 @@ Go 版本车辆数据接入链路已经作为生产主链路运行在 ECS `115.2 TDengine writer 的 raw/location 子表创建有进程内单飞保护：同一子表 key 并发首次写入时，只有一个 goroutine 执行 `CREATE TABLE IF NOT EXISTS` 和 tag 更新，其他 goroutine 等待结果后继续 INSERT，避免 10W 车辆启动或回放时对 TDengine 形成重复 DDL 风暴。 +容量健康摘要工具：`/opt/lingniu-go-native/current/capacity-check` 会抓本机各 Go 服务 `/metrics` 并输出 JSON。关键 backlog、Kafka lag、连接拒绝或 metrics 抓取失败时退出码为 `2`，可以接 cron/告警。 + ## 服务和端口 ECS：`115.29.187.205` @@ -316,6 +318,7 @@ GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tm GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tmp/lingniu-go-deploy/stat-writer ./cmd/stat-writer GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tmp/lingniu-go-deploy/nats-fast-writer ./cmd/nats-fast-writer GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tmp/lingniu-go-deploy/nats-kafka-bridge ./cmd/nats-kafka-bridge +GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tmp/lingniu-go-deploy/capacity-check ./cmd/capacity-check ``` 上传部署： diff --git a/go/vehicle-gateway/cmd/capacity-check/main.go b/go/vehicle-gateway/cmd/capacity-check/main.go new file mode 100644 index 00000000..84b37b77 --- /dev/null +++ b/go/vehicle-gateway/cmd/capacity-check/main.go @@ -0,0 +1,109 @@ +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strings" + "time" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/capacity" +) + +const defaultEndpoints = "gateway=http://127.0.0.1:20211,history=http://127.0.0.1:20212,stat=http://127.0.0.1:20213,bridge=http://127.0.0.1:20214,realtime=http://127.0.0.1:20200,fast-writer=http://127.0.0.1:20215" + +type endpoint struct { + Name string + URL string +} + +func main() { + var rawEndpoints string + var timeout time.Duration + flag.StringVar(&rawEndpoints, "endpoints", defaultEndpoints, "comma separated name=url endpoints; /metrics is appended when no path is provided") + flag.DurationVar(&timeout, "timeout", 2*time.Second, "per endpoint timeout") + flag.Parse() + + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + metricsByService, findings := collectMetrics(ctx, parseEndpoints(rawEndpoints), timeout) + report := capacity.Evaluate(metricsByService) + report.Findings = append(report.Findings, findings...) + if len(report.Findings) > 0 { + report.Status = capacity.StatusDegraded + } + encoder := json.NewEncoder(os.Stdout) + encoder.SetIndent("", " ") + _ = encoder.Encode(report) + if report.Status != capacity.StatusOK { + os.Exit(2) + } +} + +func parseEndpoints(value string) []endpoint { + var endpoints []endpoint + for _, item := range strings.Split(value, ",") { + item = strings.TrimSpace(item) + if item == "" { + continue + } + name, url, ok := strings.Cut(item, "=") + if !ok { + continue + } + name = strings.TrimSpace(name) + url = withMetricsPath(strings.TrimSpace(url)) + if name == "" || url == "" { + continue + } + endpoints = append(endpoints, endpoint{Name: name, URL: url}) + } + return endpoints +} + +func withMetricsPath(value string) string { + parsed, err := url.Parse(value) + if err != nil || parsed.Scheme == "" || parsed.Host == "" { + return value + } + if parsed.Path == "" || parsed.Path == "/" { + parsed.Path = "/metrics" + } + return parsed.String() +} + +func collectMetrics(ctx context.Context, endpoints []endpoint, timeout time.Duration) (map[string]string, []string) { + client := &http.Client{Timeout: timeout} + metricsByService := make(map[string]string, len(endpoints)) + var findings []string + for _, ep := range endpoints { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, ep.URL, nil) + if err != nil { + findings = append(findings, fmt.Sprintf("%s metrics request invalid: %v", ep.Name, err)) + continue + } + response, err := client.Do(request) + if err != nil { + findings = append(findings, fmt.Sprintf("%s metrics fetch failed: %v", ep.Name, err)) + continue + } + body, readErr := io.ReadAll(response.Body) + _ = response.Body.Close() + if readErr != nil { + findings = append(findings, fmt.Sprintf("%s metrics read failed: %v", ep.Name, readErr)) + continue + } + if response.StatusCode < 200 || response.StatusCode >= 300 { + findings = append(findings, fmt.Sprintf("%s metrics status %d", ep.Name, response.StatusCode)) + continue + } + metricsByService[ep.Name] = string(body) + } + return metricsByService, findings +} diff --git a/go/vehicle-gateway/cmd/capacity-check/main_test.go b/go/vehicle-gateway/cmd/capacity-check/main_test.go new file mode 100644 index 00000000..34d35363 --- /dev/null +++ b/go/vehicle-gateway/cmd/capacity-check/main_test.go @@ -0,0 +1,31 @@ +package main + +import ( + "strings" + "testing" +) + +func TestParseEndpoints(t *testing.T) { + endpoints := parseEndpoints("gateway=http://127.0.0.1:20211/metrics, fast=http://127.0.0.1:20215/metrics") + + if len(endpoints) != 2 { + t.Fatalf("len = %d, want 2: %#v", len(endpoints), endpoints) + } + if endpoints[0].Name != "gateway" || endpoints[0].URL != "http://127.0.0.1:20211/metrics" { + t.Fatalf("endpoint[0] = %#v", endpoints[0]) + } + if endpoints[1].Name != "fast" || endpoints[1].URL != "http://127.0.0.1:20215/metrics" { + t.Fatalf("endpoint[1] = %#v", endpoints[1]) + } +} + +func TestParseEndpointsUsesDefaultMetricsURLWhenOnlyBaseURLIsProvided(t *testing.T) { + endpoints := parseEndpoints("gateway=http://127.0.0.1:20211") + + if len(endpoints) != 1 { + t.Fatalf("len = %d, want 1", len(endpoints)) + } + if !strings.HasSuffix(endpoints[0].URL, "/metrics") { + t.Fatalf("url = %q, want /metrics suffix", endpoints[0].URL) + } +} diff --git a/go/vehicle-gateway/internal/capacity/check.go b/go/vehicle-gateway/internal/capacity/check.go new file mode 100644 index 00000000..e035a1fb --- /dev/null +++ b/go/vehicle-gateway/internal/capacity/check.go @@ -0,0 +1,132 @@ +package capacity + +import ( + "bufio" + "fmt" + "sort" + "strconv" + "strings" + "time" +) + +type Status string + +const ( + StatusOK Status = "ok" + StatusDegraded Status = "degraded" +) + +type Report struct { + Status Status `json:"status"` + CheckedAt time.Time `json:"checked_at"` + Totals Totals `json:"totals"` + Services []ServiceCheck `json:"services"` + Findings []string `json:"findings"` +} + +type Totals struct { + ActiveConnections int64 `json:"active_connections"` + KafkaLag float64 `json:"kafka_lag"` +} + +type ServiceCheck struct { + Name string `json:"name"` + Metrics map[string]float64 `json:"metrics"` +} + +func Evaluate(metricsByService map[string]string) Report { + report := Report{ + Status: StatusOK, + CheckedAt: time.Now().UTC(), + Services: make([]ServiceCheck, 0, len(metricsByService)), + } + var names []string + for name := range metricsByService { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + metrics := parseMetrics(metricsByService[name]) + report.Services = append(report.Services, ServiceCheck{Name: name, Metrics: metrics}) + report.Totals.ActiveConnections += int64(metrics["vehicle_gateway_active_connections"]) + 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.Findings = append(report.Findings, findingsForMetrics(metrics)...) + } + if len(report.Findings) > 0 { + report.Status = StatusDegraded + } + return report +} + +func findingsForMetrics(metrics map[string]float64) []string { + var findings []string + if value := metrics["vehicle_gateway_connection_rejections_total"]; value > 0 { + findings = append(findings, fmt.Sprintf("gateway connection rejections %.0f", value)) + } + if value := metrics["vehicle_async_sink_queue_depth"]; value > 10_000 { + findings = append(findings, fmt.Sprintf("async sink queue depth %.0f exceeds 10000", value)) + } + if value := metrics["vehicle_bridge_nats_consumer_ack_pending"]; value > 100 { + findings = append(findings, fmt.Sprintf("bridge ack pending %.0f exceeds 100", value)) + } + if value := metrics["vehicle_bridge_nats_consumer_pending"]; value > 10_000 { + findings = append(findings, fmt.Sprintf("bridge consumer pending %.0f exceeds 10000", value)) + } + if value := metrics["vehicle_bridge_batch_pending_messages"]; value > 1000 { + findings = append(findings, fmt.Sprintf("bridge batch pending %.0f exceeds 1000", value)) + } + if value := metrics["vehicle_fast_writer_nats_consumer_ack_pending"]; value > 10 { + findings = append(findings, fmt.Sprintf("fast writer ack pending %.0f exceeds 10", value)) + } + if value := metrics["vehicle_fast_writer_nats_consumer_pending"]; value > 10_000 { + findings = append(findings, fmt.Sprintf("fast writer consumer pending %.0f exceeds 10000", value)) + } + if value := metrics["vehicle_fast_writer_batch_pending_messages"]; value > 0 { + findings = append(findings, fmt.Sprintf("fast writer batch pending %.0f", value)) + } + if value := metrics["vehicle_history_batch_pending_messages"]; value > 0 { + findings = append(findings, fmt.Sprintf("history batch pending %.0f", value)) + } + if value := metrics["vehicle_history_batch_pending_rows"]; value > 0 { + findings = append(findings, fmt.Sprintf("history rows pending %.0f", value)) + } + if value := metrics["vehicle_history_kafka_lag"] + metrics["vehicle_stat_kafka_lag"] + metrics["vehicle_realtime_kafka_lag"]; value > 0 { + findings = append(findings, fmt.Sprintf("kafka lag %.0f", value)) + } + return findings +} + +func parseMetrics(text string) map[string]float64 { + out := map[string]float64{} + scanner := bufio.NewScanner(strings.NewReader(text)) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + name, value, ok := parseMetricLine(line) + if !ok { + continue + } + out[name] += value + } + return out +} + +func parseMetricLine(line string) (string, float64, bool) { + parts := strings.Fields(line) + if len(parts) < 2 { + return "", 0, false + } + name := parts[0] + if index := strings.IndexByte(name, '{'); index >= 0 { + name = name[:index] + } + value, err := strconv.ParseFloat(parts[1], 64) + if err != nil { + return "", 0, false + } + return name, value, true +} diff --git a/go/vehicle-gateway/internal/capacity/check_test.go b/go/vehicle-gateway/internal/capacity/check_test.go new file mode 100644 index 00000000..0f501b21 --- /dev/null +++ b/go/vehicle-gateway/internal/capacity/check_test.go @@ -0,0 +1,66 @@ +package capacity + +import ( + "strings" + "testing" +) + +func TestEvaluateReportsOKWhenCriticalBacklogsAreZero(t *testing.T) { + report := Evaluate(map[string]string{ + "gateway": `vehicle_gateway_active_connections{protocol="JT808"} 50000 +vehicle_gateway_connection_rejections_total{protocol="JT808",reason="max_connections"} 0 +vehicle_async_sink_queue_depth{sink="nats"} 0`, + "fast-writer": `vehicle_fast_writer_nats_consumer_ack_pending{consumer="vehicle-fast-writer",stream="VEHICLE_INGEST"} 0 +vehicle_fast_writer_nats_consumer_pending{consumer="vehicle-fast-writer",stream="VEHICLE_INGEST"} 0 +vehicle_fast_writer_batch_pending_messages 0`, + "history": `vehicle_history_batch_pending_messages 0 +vehicle_history_batch_pending_rows 0 +vehicle_history_kafka_lag{topic="vehicle.raw.go.jt808.v1",partition="0"} 0`, + }) + + if report.Status != StatusOK { + t.Fatalf("status = %s, want ok: %#v", report.Status, report) + } + if report.Totals.ActiveConnections != 50000 { + t.Fatalf("active connections = %d, want 50000", report.Totals.ActiveConnections) + } + if len(report.Findings) != 0 { + t.Fatalf("findings = %#v, want none", report.Findings) + } +} + +func TestEvaluateReportsDegradedForPendingAndRejects(t *testing.T) { + report := Evaluate(map[string]string{ + "gateway": `vehicle_gateway_active_connections{protocol="JT808"} 120000 +vehicle_gateway_connection_rejections_total{protocol="JT808",reason="max_connections"} 3 +vehicle_async_sink_queue_depth{sink="nats"} 25000`, + "bridge": `vehicle_bridge_nats_consumer_ack_pending{consumer="vehicle-kafka-bridge",stream="VEHICLE_INGEST"} 101 +vehicle_bridge_nats_consumer_pending{consumer="vehicle-kafka-bridge",stream="VEHICLE_INGEST"} 12001 +vehicle_bridge_batch_pending_messages 1001`, + "fast-writer": `vehicle_fast_writer_nats_consumer_ack_pending{consumer="vehicle-fast-writer",stream="VEHICLE_INGEST"} 11 +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_kafka_lag{topic="vehicle.raw.go.jt808.v1",partition="0"} 42`, + }) + + if report.Status != StatusDegraded { + t.Fatalf("status = %s, want degraded: %#v", report.Status, report) + } + joined := strings.Join(report.Findings, "\n") + for _, want := range []string{ + "gateway connection rejections", + "async sink queue depth", + "bridge ack pending", + "bridge consumer pending", + "bridge batch pending", + "fast writer ack pending", + "fast writer consumer pending", + "history batch pending", + "kafka lag", + } { + if !strings.Contains(joined, want) { + t.Fatalf("finding missing %q in:\n%s", want, joined) + } + } +}