diff --git a/deploy/observability/prometheus-rules.yaml b/deploy/observability/prometheus-rules.yaml index b6bc71a8..973a349d 100644 --- a/deploy/observability/prometheus-rules.yaml +++ b/deploy/observability/prometheus-rules.yaml @@ -52,3 +52,21 @@ spec: The 5-minute 5xx ratio for operation {{ $labels.operation }} has exceeded 1 percent for 5 minutes. runbook_url: https://example.invalid/cosmic-clash/runbooks/control-plane-api + - name: cosmic-clash.allocator + rules: + - alert: CosmicClashAllocatorQuotaDenials + expr: | + sum by (region) ( + increase(cosmic_clash_allocator_quota_denials_total[15m]) + ) > 0 + for: 5m + labels: + severity: warning + owner: allocator + annotations: + summary: Cosmic Clash allocator quota is denying allocation attempts + description: >- + The {{ $labels.region }} allocator has denied at least one + allocation attempt in the last 15 minutes; verify quota capacity, + provider health, and denial-of-wallet activity. + runbook_url: https://example.invalid/cosmic-clash/runbooks/allocator-quota diff --git a/multiplayer-next.md b/multiplayer-next.md index ed46825b..1ebec0e9 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1251,7 +1251,7 @@ the local/CI/community transport, not a silent production fallback. | 8.49 `[D:8.25,8.26,8.28,8.29,8.30,8.31,8.35,8.36]` | **IN PROGRESS.** `scripts/verify_kind_agones.sh` creates a disposable kind cluster, installs pinned Agones, loads the real `game-server` image, applies the Fleet in an explicitly separate Agones-only supervisor/UDP readiness mode, and verifies readiness plus allocation of a dynamic UDP endpoint; `.github/workflows/agones-integration.yml` runs it for infrastructure changes and on demand | The cloud-free runner is committed and fails clearly when Docker/kind/Helm are unavailable. CI/live evidence for production control-plane registration, roster/no-show, both readiness stages, races, multi-match node, result-pending reconciliation, drain, and rollback remains open | | 8.50 `[D:8.25,8.37,8.43,8.49]` | **IN PROGRESS.** `make verify-chaos-recovery` provides a disposable PostgreSQL + real testkit API + real maintenance flow: it restarts the API, injects a stale allocation, and verifies no-penalty requeue plus a durable participant-targeted lifecycle event | The API-restart/stalled-allocation slice is implemented and documented; 100 ms RTT/jitter/loss, matcher/client restart, game-pod death, node drain, Redis failover, control-plane loss, and live chaos evidence remain | | 8.51 `[D:8.17,8.18,8.30,8.31,8.45]` | **IN PROGRESS.** The opt-in `make verify-multiplayer-load` gate drives 10,000 real HTTP queue-create requests through the service with 256 in flight and records p95/p99, plus 100 concurrent proposal formations through the real matcher/domain path; the handler and in-process ownership boundary are exercised without weakening normal tests | Local API load passes at p95 <250 ms in normal and race runs, and the matcher forms 100 unique proposals; PostgreSQL saturation, durable matcher fencing under load, forecast launch concurrency x2, and replica scaling remain live infrastructure gates | -| 8.52 `[D:8.32,8.34,8.45,8.51]` | **IN PROGRESS.** Allocator supports both an opt-in per-replica fixed-window quota (`--allocation-quota` / `--allocation-quota-window`) and an optional PostgreSQL-backed EU/NA quota table consumed inside the serializable allocation transaction before any provider call; idempotent replays do not double-count | Normal/race/vet tests cover local quota exhaustion, window reset, region isolation, invalid input, and atomic concurrent consumption; migration/SQL coverage defines the shared quota boundary; measured regional cost model, budget alerts, and denial-of-wallet production rehearsal remain | +| 8.52 `[D:8.32,8.34,8.45,8.51]` | **IN PROGRESS.** Allocator supports both an opt-in per-replica fixed-window quota (`--allocation-quota` / `--allocation-quota-window`) and an optional PostgreSQL-backed EU/NA quota table consumed inside the serializable allocation transaction before any provider call; idempotent replays do not double-count. The allocator now exposes bounded EU/NA Prometheus counters at an explicit `/metrics` listener (`--metrics-addr`), including quota denials, and the checked-in rule warns on regional denial activity | Normal/race/vet tests cover local quota exhaustion, window reset, region isolation, invalid input, atomic concurrent consumption, bounded metrics labels, and read-only endpoint behavior; migration/SQL coverage defines the shared quota boundary; production scrape wiring, measured regional cost model, threshold tuning, and denial-of-wallet rehearsal remain | | 8.53 `[D:7.8,8.13,8.38,8.45,8.46,8.48,8.49,8.50,8.51,8.52]` | **IN PROGRESS.** `scripts/verify_release_gate.py` provides a fail-closed promotion check for the ordered development → internal → casual canary → casual → provisional ranked → ranked stages, requiring an evidence report for SLO, security, cost, rollback, EU+NA playtests, and both legacy gates | Validator and adversarial tests cover skipped stages, unknown stages, missing gates, non-boolean gate values, and blank release IDs; the actual reports, production rollback rehearsal, regional playtests, and live promotion remain open | Implementation invariants for every task above: diff --git a/scripts/verify_observability_manifests.py b/scripts/verify_observability_manifests.py index e4ff5873..9f9ee4e4 100644 --- a/scripts/verify_observability_manifests.py +++ b/scripts/verify_observability_manifests.py @@ -40,7 +40,7 @@ def verify(directory: Path, service_path: Path) -> None: if "kind: PrometheusRule" not in rules: raise ValueError("PrometheusRule resource is missing") - for alert in ("CosmicClashControlPlaneAPIP95High", "CosmicClashControlPlaneAPI5xxHigh"): + for alert in ("CosmicClashControlPlaneAPIP95High", "CosmicClashControlPlaneAPI5xxHigh", "CosmicClashAllocatorQuotaDenials"): if f"alert: {alert}" not in rules: raise ValueError(f"required alert is missing: {alert}") if "histogram_quantile" not in rules or "cosmic_clash_api_latency_seconds_bucket" not in rules: @@ -49,6 +49,8 @@ def verify(directory: Path, service_path: Path) -> None: raise ValueError("API error alert is not based on the exported counter") if "severity: page" not in rules or "owner: api" not in rules: raise ValueError("alerts must have bounded routing labels") + if "cosmic_clash_allocator_quota_denials_total" not in rules or "owner: allocator" not in rules: + raise ValueError("allocator quota alert is not based on bounded allocator metrics") routing = rules.split("labels:", 1)[-1].split("annotations:", 1)[0] if "{{ $labels." in routing: raise ValueError("dynamic labels were added to alert routing") diff --git a/server/allocator/metrics.go b/server/allocator/metrics.go new file mode 100644 index 00000000..808a109f --- /dev/null +++ b/server/allocator/metrics.go @@ -0,0 +1,99 @@ +package allocator + +import ( + "fmt" + "io" + "net/http" + "sync" +) + +// Metrics is a bounded allocator-role collector. Region is the only label so +// a bad request cannot create unbounded Prometheus cardinality. +type Metrics struct { + mu sync.Mutex + regions map[string]*allocationMetric +} + +type allocationMetric struct { + attempts uint64 + success uint64 + failure uint64 + denied uint64 +} + +func NewMetrics() *Metrics { + return &Metrics{regions: map[string]*allocationMetric{"EU": {}, "NA": {}}} +} + +func (m *Metrics) ObserveAttempt(region string) { + if metric := m.metric(region); metric != nil { + m.mu.Lock() + metric.attempts++ + m.mu.Unlock() + } +} + +func (m *Metrics) ObserveSuccess(region string) { + if metric := m.metric(region); metric != nil { + m.mu.Lock() + metric.success++ + m.mu.Unlock() + } +} + +func (m *Metrics) ObserveFailure(region string) { + if metric := m.metric(region); metric != nil { + m.mu.Lock() + metric.failure++ + m.mu.Unlock() + } +} + +func (m *Metrics) ObserveDenied(region string) { + if metric := m.metric(region); metric != nil { + m.mu.Lock() + metric.denied++ + m.mu.Unlock() + } +} + +func (m *Metrics) metric(region string) *allocationMetric { + if m == nil || (region != "EU" && region != "NA") { + return nil + } + return m.regions[region] +} + +func (m *Metrics) WritePrometheus(w io.Writer) error { + if m == nil { + return nil + } + m.mu.Lock() + defer m.mu.Unlock() + if _, err := io.WriteString(w, "# TYPE cosmic_clash_allocator_allocation_attempts_total counter\n# TYPE cosmic_clash_allocator_allocations_total counter\n# TYPE cosmic_clash_allocator_allocation_failures_total counter\n# TYPE cosmic_clash_allocator_quota_denials_total counter\n"); err != nil { + return err + } + for _, region := range []string{"EU", "NA"} { + metric := m.regions[region] + labels := fmt.Sprintf(`region="%s"`, region) + if _, err := fmt.Fprintf(w, "cosmic_clash_allocator_allocation_attempts_total{%s} %d\ncosmic_clash_allocator_allocations_total{%s} %d\ncosmic_clash_allocator_allocation_failures_total{%s} %d\ncosmic_clash_allocator_quota_denials_total{%s} %d\n", labels, metric.attempts, labels, metric.success, labels, metric.failure, labels, metric.denied); err != nil { + return err + } + } + return nil +} + +// MetricsHandler exposes only the read-only Prometheus endpoint. The caller +// owns the listener and can bind it to a private metrics network. +func MetricsHandler(metrics *Metrics) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + _ = metrics.WritePrometheus(w) + }) + return mux +} diff --git a/server/allocator/metrics_test.go b/server/allocator/metrics_test.go new file mode 100644 index 00000000..46451de5 --- /dev/null +++ b/server/allocator/metrics_test.go @@ -0,0 +1,60 @@ +package allocator + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestMetricsExportsFixedRegionalCounters(t *testing.T) { + metrics := NewMetrics() + metrics.ObserveAttempt("EU") + metrics.ObserveSuccess("EU") + metrics.ObserveFailure("EU") + metrics.ObserveDenied("EU") + metrics.ObserveAttempt("APAC") + var output strings.Builder + if err := metrics.WritePrometheus(&output); err != nil { + t.Fatal(err) + } + text := output.String() + for _, fragment := range []string{ + `cosmic_clash_allocator_allocation_attempts_total{region="EU"} 1`, + `cosmic_clash_allocator_allocations_total{region="EU"} 1`, + `cosmic_clash_allocator_allocation_failures_total{region="EU"} 1`, + `cosmic_clash_allocator_quota_denials_total{region="EU"} 1`, + `region="NA"`, + } { + if !strings.Contains(text, fragment) { + t.Fatalf("metrics missing %q: %s", fragment, text) + } + } + if strings.Contains(text, "APAC") { + t.Fatal("unbounded region label escaped into metrics") + } +} + +func TestNilMetricsAreSafe(t *testing.T) { + var metrics *Metrics + metrics.ObserveAttempt("EU") + if err := metrics.WritePrometheus(&strings.Builder{}); err != nil { + t.Fatal(err) + } +} + +func TestMetricsHandlerIsReadOnlyAndScoped(t *testing.T) { + metrics := NewMetrics() + metrics.ObserveAttempt("NA") + handler := MetricsHandler(metrics) + get := httptest.NewRecorder() + handler.ServeHTTP(get, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + if get.Code != http.StatusOK || !strings.Contains(get.Body.String(), `region="NA"`) { + t.Fatalf("GET /metrics status=%d body=%s", get.Code, get.Body.String()) + } + post := httptest.NewRecorder() + handler.ServeHTTP(post, httptest.NewRequest(http.MethodPost, "/metrics", nil)) + if post.Code != http.StatusMethodNotAllowed { + t.Fatalf("POST /metrics status=%d, want 405", post.Code) + } +} diff --git a/server/allocator/service.go b/server/allocator/service.go index 9bfccf87..ab771fbb 100644 --- a/server/allocator/service.go +++ b/server/allocator/service.go @@ -34,12 +34,20 @@ type SharedAllocationQuota interface { Consume(context.Context, string, time.Time) error } +type AllocationMetrics interface { + ObserveAttempt(string) + ObserveSuccess(string) + ObserveFailure(string) + ObserveDenied(string) +} + type Service struct { Provider Provider Durable Durable Roster RosterPublisher Budget AllocationBudget Quota SharedAllocationQuota + Metrics AllocationMetrics Now func() time.Time } @@ -88,23 +96,41 @@ func (s Service) Allocate(ctx context.Context, request domain.AllocationRequest, return agones.AllocatedServer{}, errNotConfigured } now := s.Now() + if s.Metrics != nil { + s.Metrics.ObserveAttempt(request.Region) + } if s.Budget != nil { if err := s.Budget.Allow(request.Region, now); err != nil { + if s.Metrics != nil { + s.Metrics.ObserveDenied(request.Region) + } return agones.AllocatedServer{}, err } } if s.Quota != nil { if err := s.Quota.Consume(ctx, request.Region, now); err != nil { + if s.Metrics != nil { + s.Metrics.ObserveDenied(request.Region) + } return agones.AllocatedServer{}, err } } result, err := s.Provider.Allocate(ctx, request, labels, now) if err != nil { + if s.Metrics != nil { + s.Metrics.ObserveFailure(request.Region) + } return agones.AllocatedServer{}, err } if _, err := s.Durable.RecordProviderAllocation(ctx, result.Allocation, now); err != nil { + if s.Metrics != nil { + s.Metrics.ObserveFailure(request.Region) + } return agones.AllocatedServer{}, err } + if s.Metrics != nil { + s.Metrics.ObserveSuccess(request.Region) + } return result, nil } @@ -114,10 +140,21 @@ func (s Service) RecordProviderAllocation(ctx context.Context, result agones.All } if s.Quota != nil { if err := s.Quota.Consume(ctx, result.Allocation.Region, now); err != nil { + if s.Metrics != nil { + s.Metrics.ObserveDenied(result.Allocation.Region) + } return domain.Allocation{}, err } } - return s.Durable.RecordProviderAllocation(ctx, result.Allocation, now) + allocation, err := s.Durable.RecordProviderAllocation(ctx, result.Allocation, now) + if s.Metrics != nil { + if err != nil { + s.Metrics.ObserveFailure(result.Allocation.Region) + } else { + s.Metrics.ObserveSuccess(result.Allocation.Region) + } + } + return allocation, err } var errNotConfigured = &configurationError{} diff --git a/server/allocator/service_test.go b/server/allocator/service_test.go index 116b4ce6..323613fd 100644 --- a/server/allocator/service_test.go +++ b/server/allocator/service_test.go @@ -3,6 +3,7 @@ package allocator import ( "context" "errors" + "strings" "testing" "time" @@ -56,11 +57,16 @@ func (d *durableSpy) RecordProviderAllocation(_ context.Context, allocation doma func TestServiceDurablyRecordsProviderAllocationBeforeReturning(t *testing.T) { provider := &providerSpy{result: agones.AllocatedServer{Allocation: domain.Allocation{AllocationID: "a", MatchID: "m", ServerID: "gs", State: domain.ServerAllocated}, Endpoint: "127.0.0.1:7777"}} durable := &durableSpy{} - service := Service{Provider: provider, Durable: durable, Now: func() time.Time { return time.Unix(1000, 0) }} + metrics := NewMetrics() + service := Service{Provider: provider, Durable: durable, Metrics: metrics, Now: func() time.Time { return time.Unix(1000, 0) }} result, err := service.Allocate(context.Background(), domain.AllocationRequest{AllocationID: "a", MatchID: "m", Region: "EU", Build: "b", Protocol: 1, Transport: "enet"}, map[string]string{"region": "EU"}) if err != nil || result.Endpoint == "" || durable.calls != 1 || durable.allocation.ServerID != "gs" { t.Fatalf("result=%+v err=%v durable=%+v", result, err, durable) } + var output strings.Builder + if err := metrics.WritePrometheus(&output); err != nil || !strings.Contains(output.String(), `allocations_total{region="EU"} 1`) { + t.Fatalf("success metric err=%v output=%s", err, output.String()) + } } func TestServiceDoesNotReturnProviderResultAfterDurableFailure(t *testing.T) { @@ -76,13 +82,19 @@ func TestServiceDoesNotReturnProviderResultAfterDurableFailure(t *testing.T) { func TestServiceConsumesSharedQuotaBeforeFreshProviderCall(t *testing.T) { provider := &providerSpy{result: agones.AllocatedServer{Allocation: domain.Allocation{AllocationID: "a", State: domain.ServerAllocated}, Endpoint: "127.0.0.1:7777"}} quota := "aSpy{err: errors.New("quota exhausted")} - service := Service{Provider: provider, Durable: &durableSpy{}, Quota: quota, Now: func() time.Time { return time.Unix(1000, 0) }} + metrics := NewMetrics() + service := Service{Provider: provider, Durable: &durableSpy{}, Quota: quota, Metrics: metrics, Now: func() time.Time { return time.Unix(1000, 0) }} if _, err := service.Allocate(context.Background(), domain.AllocationRequest{AllocationID: "a", MatchID: "m", Region: "EU", Build: "b", Protocol: 1, Transport: "enet"}, nil); err == nil { t.Fatal("quota rejection was ignored") } if quota.calls != 1 || provider.calls != 0 { t.Fatalf("quota/provider calls = %d/%d, want 1/0", quota.calls, provider.calls) } + var output strings.Builder + _ = metrics.WritePrometheus(&output) + if !strings.Contains(output.String(), `quota_denials_total{region="EU"} 1`) { + t.Fatalf("quota denial metric missing: %s", output.String()) + } } func TestServiceConsumesSharedQuotaOnceWhenReconcilingProviderResult(t *testing.T) { diff --git a/server/cmd/allocator/main.go b/server/cmd/allocator/main.go index 8d1047ff..8754e9b5 100644 --- a/server/cmd/allocator/main.go +++ b/server/cmd/allocator/main.go @@ -5,6 +5,7 @@ import ( "database/sql" "flag" "log" + "net/http" "os" "os/signal" "syscall" @@ -27,6 +28,7 @@ func main() { workloadSecret := flag.String("workload-secret", os.Getenv("COSMIC_CLASH_WORKLOAD_SECRET"), "HMAC secret for control-plane-issued workload tokens (see workload/signed_token.go); must match cmd/control-plane's own --workload-secret. Unset skips minting a cosmic-clash.io/workload-token annotation entirely") allocationQuota := flag.Int("allocation-quota", 0, "optional per-replica allocation attempts per region per quota window; zero disables this local guard") allocationQuotaWindow := flag.Duration("allocation-quota-window", time.Minute, "window for --allocation-quota") + metricsAddr := flag.String("metrics-addr", envOrDefault("COSMIC_CLASH_ALLOCATOR_METRICS_ADDR", ":9091"), "allocator Prometheus metrics address; empty disables metrics") flag.Parse() if *dsn == "" || *agonesURL == "" { fatalf("--dsn/COSMIC_CLASH_POSTGRES_DSN and --agones-url/COSMIC_CLASH_AGONES_URL are required") @@ -62,6 +64,7 @@ func main() { } log.Printf("allocator: enabled per-replica regional allocation quota=%d window=%s", *allocationQuota, *allocationQuotaWindow) } + metrics := allocator.NewMetrics() client := agones.Client{BaseURL: *agonesURL, Namespace: *namespace, WorkloadSecret: []byte(*workloadSecret)} worker := allocator.Worker{ Claims: store.AllocatingMatchClaims{DB: db, Transport: *transport}, @@ -70,12 +73,30 @@ func main() { Durable: store.AllocationRegistry{DB: db}, Quota: store.AllocationQuota{DB: db}, Budget: budget, + Metrics: metrics, Now: now, }, Now: now, } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + var metricsServer *http.Server + if *metricsAddr != "" { + metricsServer = &http.Server{Addr: *metricsAddr, Handler: allocator.MetricsHandler(metrics)} + go func() { + if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Printf("allocator: metrics server: %v", err) + } + }() + log.Printf("allocator: metrics listening on %s", *metricsAddr) + } + defer func() { + if metricsServer != nil { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = metricsServer.Shutdown(shutdownCtx) + } + }() ticker := time.NewTicker(*interval) defer ticker.Stop() for {