From f1b8366531605884d2432c6b0091dc011a42fa08 Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 17:06:38 +0100 Subject: [PATCH] feat(multiplayer): export bounded API metrics --- multiplayer-next.md | 2 + server/api/service.go | 78 +++++++++++++++++++++++-- server/api/service_test.go | 29 ++++++++++ server/cmd/control-plane/main.go | 1 + server/cmd/testkit-api/main.go | 2 + server/observability/metrics.go | 86 ++++++++++++++++++++++++++++ server/observability/metrics_test.go | 24 ++++++++ 7 files changed, 217 insertions(+), 5 deletions(-) create mode 100644 server/observability/metrics.go create mode 100644 server/observability/metrics_test.go diff --git a/multiplayer-next.md b/multiplayer-next.md index 065adc5c..54604fa6 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1442,3 +1442,5 @@ The same allocation path now carries the matcher-selected playlist, preventing a The Godot control-plane client now retains the exact last idempotent mutation and exposes `retry_last_mutation()` for transport, timeout, rate-limit, and 5xx failures. Retries reuse the original idempotency key and expected revision, while 401 and 409 responses remain non-retryable; the harness covers the policy boundary. This closes the local duplicate-action recovery mechanism for heartbeat/cancel/proposal calls, with broader live UI retry verification still remaining. The matchmaking UI now exposes that retained replay through its existing action button as `Retry Request` while a heartbeat, cancellation, or proposal action has a retryable failure. Terminal, authentication, and revision-conflict paths remain ineligible, so the button cannot issue a stale blind command. + +The control plane now exports bounded Prometheus-compatible API request counters and latency summaries at `GET /metrics`, with fixed operation/status labels and no event-stream wrapping. Production and testkit services wire the collector; adversarial tests verify unknown paths cannot inject label cardinality or leak URL secrets, and full Go/race/vet checks pass. Durable SLO dashboards and alert routing remain operational work. diff --git a/server/api/service.go b/server/api/service.go index 3218ea55..54671dc7 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -132,6 +132,7 @@ type Service struct { // is a valid, silent no-op -- every call site must stay optional so // existing Service literals that don't set it keep working unchanged. Log func(observability.Event) + Metrics *observability.Metrics proposalMu sync.Mutex eventsMu sync.Mutex events *eventHub @@ -182,6 +183,7 @@ func (s *Service) logProposalOutcome(proposalID string, proposal domain.Proposal func (s *Service) Handler() http.Handler { mux := http.NewServeMux() mux.HandleFunc("/healthz", s.health) + mux.HandleFunc("/metrics", s.metrics) mux.HandleFunc("/v1/session/steam", s.steamSession) mux.HandleFunc("/v1/queue", s.queueCreate) mux.HandleFunc("/v1/queue/", s.queueMutation) @@ -201,18 +203,84 @@ func (s *Service) Handler() http.Handler { mux.HandleFunc("/api/v1/assignments/", s.contractAssignment) mux.HandleFunc("/api/v1/events", s.controlPlaneEvent) mux.HandleFunc("/api/v1/servers/", s.contractServerMutation) - if s.RateLimiter == nil { - return mux + var handler http.Handler = mux + if s.RateLimiter != nil { + handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !s.RateLimiter.Allow(requestRateKey(r), s.now()) { + writeError(w, http.StatusTooManyRequests, "rate_limited") + return + } + mux.ServeHTTP(w, r) + }) + } + if s.Metrics == nil { + return handler } return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if !s.RateLimiter.Allow(requestRateKey(r), s.now()) { - writeError(w, http.StatusTooManyRequests, "rate_limited") + if r.URL.Path == "/metrics" || r.URL.Path == "/healthz" || strings.HasSuffix(r.URL.Path, "/events") { + handler.ServeHTTP(w, r) return } - mux.ServeHTTP(w, r) + started := time.Now() + recorder := &statusRecorder{ResponseWriter: w} + handler.ServeHTTP(recorder, r) + code := recorder.code + if code == 0 { + code = http.StatusOK + } + s.Metrics.ObserveAPI(metricOperation(r.URL.Path), code, time.Since(started)) }) } +type statusRecorder struct { + http.ResponseWriter + code int +} + +func (w *statusRecorder) WriteHeader(code int) { + w.code = code + w.ResponseWriter.WriteHeader(code) +} + +func (w *statusRecorder) Write(body []byte) (int, error) { + if w.code == 0 { + w.WriteHeader(http.StatusOK) + } + return w.ResponseWriter.Write(body) +} + +func (s *Service) metrics(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || s.Metrics == nil { + writeError(w, http.StatusNotFound, "not_found") + return + } + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + _ = s.Metrics.WritePrometheus(w) +} + +func metricOperation(path string) string { + switch { + case strings.Contains(path, "/queue"): + return "queue" + case strings.Contains(path, "/proposals"): + return "proposal" + case strings.Contains(path, "/assignments"): + return "assignment" + case strings.Contains(path, "ranked"): + return "ranked_profile" + case strings.Contains(path, "/profile"): + return "profile" + case strings.Contains(path, "/servers"): + return "server" + case strings.Contains(path, "/session"): + return "session" + case strings.Contains(path, "/probes"): + return "probe" + default: + return "other" + } +} + type steamSessionRequest struct { WebAPITicket string `json:"web_api_ticket"` } diff --git a/server/api/service_test.go b/server/api/service_test.go index 5d10a0b4..9c9d36a6 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -1406,6 +1406,35 @@ func TestServerShutdownAPIRequiresBoundWorkloadAndDelegatesAcknowledgement(t *te response.Body.Close() } +func TestMetricsEndpointExportsBoundedAPILatencyAndSkipsItsOwnScrape(t *testing.T) { + metrics := observability.NewMetrics() + service := &Service{Metrics: metrics, Now: time.Now} + server := httptest.NewServer(service.Handler()) + defer server.Close() + response, err := http.Get(server.URL + "/healthz") + if err != nil { + t.Fatal(err) + } + response.Body.Close() + response, err = http.Get(server.URL + "/unknown/secret-token") + if err != nil { + t.Fatal(err) + } + response.Body.Close() + response, err = http.Get(server.URL + "/metrics") + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil { + t.Fatal(err) + } + if response.StatusCode != http.StatusOK || !strings.Contains(string(body), `operation="other",status="4xx"`) || strings.Contains(string(body), "secret-token") { + t.Fatalf("metrics status=%d body=%s", response.StatusCode, body) + } +} + func TestProbeAPIUsesServerEvidenceAndRejectsClientRTTField(t *testing.T) { now := time.Unix(1000, 0).UTC() sessions := domain.NewSessionStore() diff --git a/server/cmd/control-plane/main.go b/server/cmd/control-plane/main.go index 59e289e4..28b2a6c2 100644 --- a/server/cmd/control-plane/main.go +++ b/server/cmd/control-plane/main.go @@ -114,6 +114,7 @@ func newAPIService(db *sql.DB, workloadSecret string, indexes ...api.CandidateIn WorkloadVerify: api.WorkloadVerifierFromSignedToken([]byte(workloadSecret), db), Now: func() time.Time { return time.Now().UTC() }, Log: logEvent, + Metrics: observability.NewMetrics(), } } diff --git a/server/cmd/testkit-api/main.go b/server/cmd/testkit-api/main.go index 1e51a7f3..5557a0b1 100644 --- a/server/cmd/testkit-api/main.go +++ b/server/cmd/testkit-api/main.go @@ -28,6 +28,7 @@ import ( "github.com/cosmic-clash/cosmic-clash/server/api" "github.com/cosmic-clash/cosmic-clash/server/domain" "github.com/cosmic-clash/cosmic-clash/server/migrations" + "github.com/cosmic-clash/cosmic-clash/server/observability" "github.com/cosmic-clash/cosmic-clash/server/store" _ "github.com/jackc/pgx/v5/stdlib" ) @@ -72,6 +73,7 @@ func main() { }, ProbeRecorder: store.PostgresQueue{DB: db}, WorkloadVerify: api.WorkloadVerifierFromSignedToken([]byte(*workloadSecret), db), + Metrics: observability.NewMetrics(), Now: func() time.Time { return time.Now().UTC() }, } handler := service.Handler() diff --git a/server/observability/metrics.go b/server/observability/metrics.go new file mode 100644 index 00000000..80d3db2c --- /dev/null +++ b/server/observability/metrics.go @@ -0,0 +1,86 @@ +package observability + +import ( + "fmt" + "io" + "sort" + "sync" + "time" +) + +// Metrics is a bounded in-process collector for API request health. Operation +// names are normalized to a fixed vocabulary before storage. +type Metrics struct { + mu sync.Mutex + counts map[metricKey]uint64 + sums map[metricKey]time.Duration +} + +type metricKey struct{ operation, status string } + +func NewMetrics() *Metrics { + return &Metrics{counts: make(map[metricKey]uint64), sums: make(map[metricKey]time.Duration)} +} + +func (m *Metrics) ObserveAPI(operation string, statusCode int, duration time.Duration) { + if m == nil { + return + } + if duration < 0 { + duration = 0 + } + key := metricKey{normalizeOperation(operation), statusClass(statusCode)} + m.mu.Lock() + m.counts[key]++ + m.sums[key] += duration + m.mu.Unlock() +} + +func (m *Metrics) WritePrometheus(w io.Writer) error { + if m == nil { + return nil + } + m.mu.Lock() + keys := make([]metricKey, 0, len(m.counts)) + for key := range m.counts { + keys = append(keys, key) + } + sort.Slice(keys, func(i, j int) bool { + if keys[i].operation != keys[j].operation { + return keys[i].operation < keys[j].operation + } + return keys[i].status < keys[j].status + }) + counts := make(map[metricKey]uint64, len(keys)) + sums := make(map[metricKey]time.Duration, len(keys)) + for _, key := range keys { + counts[key], sums[key] = m.counts[key], m.sums[key] + } + m.mu.Unlock() + if _, err := io.WriteString(w, "# TYPE cosmic_clash_api_requests_total counter\n# TYPE cosmic_clash_api_latency_seconds summary\n"); err != nil { + return err + } + for _, key := range keys { + labels := fmt.Sprintf(`operation="%s",status="%s"`, key.operation, key.status) + if _, err := fmt.Fprintf(w, "cosmic_clash_api_requests_total{%s} %d\ncosmic_clash_api_latency_seconds_count{%s} %d\ncosmic_clash_api_latency_seconds_sum{%s} %.9f\n", labels, counts[key], labels, counts[key], labels, sums[key].Seconds()); err != nil { + return err + } + } + return nil +} + +func normalizeOperation(operation string) string { + for _, allowed := range []string{"queue", "proposal", "assignment", "profile", "ranked_profile", "server", "events", "session", "probe"} { + if operation == allowed { + return allowed + } + } + return "other" +} + +func statusClass(code int) string { + if code < 100 || code > 599 { + return "unknown" + } + return fmt.Sprintf("%dxx", code/100) +} diff --git a/server/observability/metrics_test.go b/server/observability/metrics_test.go new file mode 100644 index 00000000..4ea6a9a3 --- /dev/null +++ b/server/observability/metrics_test.go @@ -0,0 +1,24 @@ +package observability + +import ( + "strings" + "testing" + "time" +) + +func TestMetricsNormalizesOperationsAndExportsBoundedLabels(t *testing.T) { + m := NewMetrics() + m.ObserveAPI("queue", 201, 10*time.Millisecond) + m.ObserveAPI("/crafted/path/with-secret", 500, time.Second) + var output strings.Builder + if err := m.WritePrometheus(&output); err != nil { + t.Fatal(err) + } + text := output.String() + if !strings.Contains(text, `operation="queue",status="2xx"`) || !strings.Contains(text, `operation="other",status="5xx"`) { + t.Fatalf("metrics output = %s", text) + } + if strings.Contains(text, "crafted") || strings.Contains(text, "secret") { + t.Fatalf("unbounded operation label leaked: %s", text) + } +}