feat(go): add capacity health check tool

2026-07-03 19:54:29 +08:00
parent 68b729fa7b
commit 3edd7462c7
6 changed files with 349 additions and 0 deletions
--- a/go/vehicle-gateway/internal/capacity/check.go
+++ 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
+}
--- a/go/vehicle-gateway/internal/capacity/check_test.go
+++ 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)
+		}
+	}
+}