lingniu-vehicle-ingest/go/vehicle-gateway/internal/capacity/check.go

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"`
	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 {
	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.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 {
		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
}