package metrics import ( "context" "fmt" "net/http" "runtime" "sync/atomic" "time" ) var startedAt = time.Now() var submitAccepted atomic.Uint64 var submitFailed atomic.Uint64 var submitDurationNanoseconds atomic.Uint64 var durationBuckets = [...]time.Duration{ 5 * time.Millisecond, 10 * time.Millisecond, 25 * time.Millisecond, 50 * time.Millisecond, 100 * time.Millisecond, 250 * time.Millisecond, 500 * time.Millisecond, time.Second, 3 * time.Second, 10 * time.Second, } type durationHistogram struct { count atomic.Uint64 sumNano atomic.Uint64 buckets [len(durationBuckets)]atomic.Uint64 } var inboundStageHistograms [4][2]durationHistogram var submitStageHistograms [5][2]durationHistogram type Snapshot struct { UpstreamDesired int UpstreamConnected int DownstreamConnected int SubmitWorkerUp bool SubmitWorkerConcurrency int SubmitWorkerInFlight int64 InboundSubmitConcurrency int InboundSubmitInFlight int64 QueueAvailable bool QueuePending int64 QueueLag int64 QueueOldestAgeSeconds float64 } type SnapshotFunc func(context.Context) Snapshot func ObserveSubmit(accepted bool, duration time.Duration) { if accepted { submitAccepted.Add(1) } else { submitFailed.Add(1) } submitDurationNanoseconds.Add(uint64(max(duration, 0))) } // ObserveInboundStage deliberately accepts only a fixed stage/result vocabulary. // Entity identifiers would create unbounded Prometheus series during high-volume traffic. func ObserveInboundStage(stage string, success bool, duration time.Duration) { stageIndex := inboundStageIndex(stage) if stageIndex < 0 { return } observeDuration(&inboundStageHistograms[stageIndex][boolIndex(success)], duration) } // ObserveSubmitStage separates queueing, limiting, connection-window, supplier and API callback time. func ObserveSubmitStage(stage string, success bool, duration time.Duration) { stageIndex := submitStageIndex(stage) if stageIndex < 0 { return } observeDuration(&submitStageHistograms[stageIndex][boolIndex(success)], duration) } func Handler(load SnapshotFunc) http.Handler { return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { if request.Method != http.MethodGet || request.URL.Path != "/metrics" { response.WriteHeader(http.StatusNotFound) return } ctx, cancel := context.WithTimeout(request.Context(), time.Second) defer cancel() snapshot := Snapshot{} if load != nil { snapshot = load(ctx) } var memory runtime.MemStats runtime.ReadMemStats(&memory) accepted := submitAccepted.Load() failed := submitFailed.Load() count := accepted + failed response.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") response.Header().Set("Cache-Control", "no-store") fmt.Fprintf(response, "# HELP cmpp_gateway_process_uptime_seconds Gateway process uptime.\n# TYPE cmpp_gateway_process_uptime_seconds gauge\ncmpp_gateway_process_uptime_seconds %f\n", time.Since(startedAt).Seconds()) fmt.Fprintf(response, "# HELP cmpp_gateway_go_goroutines Current goroutine count.\n# TYPE cmpp_gateway_go_goroutines gauge\ncmpp_gateway_go_goroutines %d\n", runtime.NumGoroutine()) fmt.Fprintf(response, "# HELP cmpp_gateway_go_heap_alloc_bytes Current Go heap allocation.\n# TYPE cmpp_gateway_go_heap_alloc_bytes gauge\ncmpp_gateway_go_heap_alloc_bytes %d\n", memory.HeapAlloc) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_total Upstream submit attempts by bounded result.\n# TYPE cmpp_gateway_submit_total counter\ncmpp_gateway_submit_total{result=\"accepted\"} %d\ncmpp_gateway_submit_total{result=\"failed\"} %d\n", accepted, failed) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_duration_seconds_sum Total upstream submit duration.\n# TYPE cmpp_gateway_submit_duration_seconds_sum counter\ncmpp_gateway_submit_duration_seconds_sum %f\n", float64(submitDurationNanoseconds.Load())/float64(time.Second)) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_duration_seconds_count Total measured upstream submits.\n# TYPE cmpp_gateway_submit_duration_seconds_count counter\ncmpp_gateway_submit_duration_seconds_count %d\n", count) fmt.Fprint(response, "# HELP cmpp_gateway_inbound_stage_duration_seconds CMPP inbound handler duration by bounded stage and result.\n# TYPE cmpp_gateway_inbound_stage_duration_seconds histogram\n") for index, stage := range []string{"decode", "api_roundtrip", "response_write", "handler_total"} { writeDurationHistogram(response, "cmpp_gateway_inbound_stage_duration_seconds", stage, &inboundStageHistograms[index]) } fmt.Fprint(response, "# HELP cmpp_gateway_submit_stage_duration_seconds Gateway submit duration by bounded stage and result.\n# TYPE cmpp_gateway_submit_stage_duration_seconds histogram\n") for index, stage := range []string{"stream_wait", "rate_limit_wait", "connection_wait", "supplier_rtt", "api_callback"} { writeDurationHistogram(response, "cmpp_gateway_submit_stage_duration_seconds", stage, &submitStageHistograms[index]) } fmt.Fprintf(response, "# HELP cmpp_gateway_upstream_connections Desired and live supplier connections.\n# TYPE cmpp_gateway_upstream_connections gauge\ncmpp_gateway_upstream_connections{state=\"desired\"} %d\ncmpp_gateway_upstream_connections{state=\"connected\"} %d\n", snapshot.UpstreamDesired, snapshot.UpstreamConnected) fmt.Fprintf(response, "# HELP cmpp_gateway_downstream_connections Authenticated client connections.\n# TYPE cmpp_gateway_downstream_connections gauge\ncmpp_gateway_downstream_connections %d\n", snapshot.DownstreamConnected) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_up Whether the submit worker was initialized.\n# TYPE cmpp_gateway_submit_worker_up gauge\ncmpp_gateway_submit_worker_up %d\n", boolNumber(snapshot.SubmitWorkerUp)) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_slots Configured and active bounded submit worker slots.\n# TYPE cmpp_gateway_submit_worker_slots gauge\ncmpp_gateway_submit_worker_slots{state=\"configured\"} %d\ncmpp_gateway_submit_worker_slots{state=\"in_flight\"} %d\n", snapshot.SubmitWorkerConcurrency, snapshot.SubmitWorkerInFlight) fmt.Fprintf(response, "# HELP cmpp_gateway_inbound_submit_slots Configured and active authenticated client Submit slots.\n# TYPE cmpp_gateway_inbound_submit_slots gauge\ncmpp_gateway_inbound_submit_slots{state=\"configured\"} %d\ncmpp_gateway_inbound_submit_slots{state=\"in_flight\"} %d\n", snapshot.InboundSubmitConcurrency, snapshot.InboundSubmitInFlight) if snapshot.QueueAvailable { fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_pending Pending entries owned by the consumer group.\n# TYPE cmpp_gateway_submit_queue_pending gauge\ncmpp_gateway_submit_queue_pending %d\n", snapshot.QueuePending) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_lag Undelivered entries for the consumer group.\n# TYPE cmpp_gateway_submit_queue_lag gauge\ncmpp_gateway_submit_queue_lag %d\n", snapshot.QueueLag) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_oldest_pending_age_seconds Age of the oldest pending entry.\n# TYPE cmpp_gateway_submit_queue_oldest_pending_age_seconds gauge\ncmpp_gateway_submit_queue_oldest_pending_age_seconds %f\n", snapshot.QueueOldestAgeSeconds) } }) } func observeDuration(histogram *durationHistogram, duration time.Duration) { duration = max(duration, 0) histogram.count.Add(1) histogram.sumNano.Add(uint64(duration)) for index, bucket := range durationBuckets { if duration <= bucket { histogram.buckets[index].Add(1) } } } func writeDurationHistogram(response http.ResponseWriter, metricName string, stage string, histograms *[2]durationHistogram) { for resultIndex, result := range []string{"failed", "success"} { histogram := &histograms[resultIndex] count := histogram.count.Load() if count == 0 { continue } for index, bucket := range durationBuckets { fmt.Fprintf(response, "%s_bucket{stage=%q,result=%q,le=%q} %d\n", metricName, stage, result, durationBucketLabel(bucket), histogram.buckets[index].Load()) } fmt.Fprintf(response, "%s_bucket{stage=%q,result=%q,le=\"+Inf\"} %d\n", metricName, stage, result, count) fmt.Fprintf(response, "%s_sum{stage=%q,result=%q} %f\n", metricName, stage, result, float64(histogram.sumNano.Load())/float64(time.Second)) fmt.Fprintf(response, "%s_count{stage=%q,result=%q} %d\n", metricName, stage, result, count) } } func durationBucketLabel(bucket time.Duration) string { return fmt.Sprintf("%g", bucket.Seconds()) } func boolIndex(value bool) int { if value { return 1 } return 0 } func inboundStageIndex(stage string) int { switch stage { case "decode": return 0 case "api_roundtrip": return 1 case "response_write": return 2 case "handler_total": return 3 default: return -1 } } func submitStageIndex(stage string) int { switch stage { case "stream_wait": return 0 case "rate_limit_wait": return 1 case "connection_wait": return 2 case "supplier_rtt": return 3 case "api_callback": return 4 default: return -1 } } func boolNumber(value bool) int { if value { return 1 } return 0 }