feat(multiplayer): export bounded API metrics

This commit is contained in:
Josh Creek
2026-09-01 17:06:38 +01:00
parent d85c8b8194
commit f1b8366531
7 changed files with 217 additions and 5 deletions
+2
View File
@@ -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.
+73 -5
View File
@@ -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"`
}
+29
View File
@@ -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()
+1
View File
@@ -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(),
}
}
+2
View File
@@ -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()
+86
View File
@@ -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)
}
+24
View File
@@ -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)
}
}