diff --git a/deploy/k8s/base/allocator-deployment.yaml b/deploy/k8s/base/allocator-deployment.yaml index 87e3e203..550a3072 100644 --- a/deploy/k8s/base/allocator-deployment.yaml +++ b/deploy/k8s/base/allocator-deployment.yaml @@ -57,13 +57,14 @@ spec: - --agones-url=https://kubernetes.default.svc - --agones-namespace=cosmic-clash - --provider-timeout=10s + - --readiness-max-stale=30s - --metrics-addr=:9091 ports: - name: metrics containerPort: 9091 readinessProbe: httpGet: - path: /metrics + path: /readyz port: metrics initialDelaySeconds: 2 periodSeconds: 5 @@ -71,7 +72,7 @@ spec: failureThreshold: 3 livenessProbe: httpGet: - path: /metrics + path: /healthz port: metrics initialDelaySeconds: 10 periodSeconds: 10 diff --git a/multiplayer-next.md b/multiplayer-next.md index 31dc5799..0fcec364 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1642,3 +1642,5 @@ Drain admission now fails at the handshake boundary: a new `_hello` is rejected Allocated team and slot assignments are now immutable after signed admission. MatchNet rejects client `_set_team` requests whenever join authorisation is required, preserving the signed global-slot/team pairing and its derived spawn index; direct/community lobbies retain team switching and its existing unready behavior. The Godot regression asserts both sides of that compatibility boundary. Per-IP API limiting now resolves the client behind the edge gateway instead of charging every player to the gateway's socket address. `X-Forwarded-For` is ignored unless the immediate peer belongs to an explicitly configured `--trusted-proxy-cidrs` range; trusted chains are walked from right to left past known proxies, while malformed/oversized chains fail closed to the immediate peer. The base deployment supplies private/CGNAT/ULA pod ranges under its edge-only ingress NetworkPolicy and calls out that production overlays should narrow them to the actual gateway CIDR. Tests cover spoofing from an untrusted peer, chained proxies, malformed input, invalid configuration, and independent clients behind one gateway. + +Allocator probes now distinguish process liveness from useful progress. `/healthz` remains live during dependency outages, while `/readyz` starts unavailable and requires a fully successful provider-list, Ready-registration, and worker cycle within `--readiness-max-stale` (30 seconds in the base deployment). The Kubernetes/Agones HTTP path is bounded by `--provider-timeout=10s`, so an unavailable provider cannot leave readiness green indefinitely; startup rejects a freshness window shorter than the poll interval plus provider timeout, and the probe listener has its own header-read deadline. Boundary and HTTP tests cover startup, exact staleness, clock reversal, recovery, method rejection, and metrics coexistence. diff --git a/server/allocator/health.go b/server/allocator/health.go new file mode 100644 index 00000000..f3e34fc5 --- /dev/null +++ b/server/allocator/health.go @@ -0,0 +1,65 @@ +package allocator + +import ( + "net/http" + "sync" + "time" +) + +// Health records only complete successful cycles. Readiness ages out when +// provider or durable-store work repeatedly fails or stalls, while liveness +// remains independent so Kubernetes does not restart a healthy process for a +// dependency outage. +type Health struct { + mu sync.RWMutex + lastSuccessfulCycle time.Time +} + +func (h *Health) ObserveSuccessfulCycle(at time.Time) { + if h == nil || at.IsZero() { + return + } + h.mu.Lock() + h.lastSuccessfulCycle = at + h.mu.Unlock() +} + +func (h *Health) Ready(now time.Time, maxStale time.Duration) bool { + if h == nil || now.IsZero() || maxStale <= 0 { + return false + } + h.mu.RLock() + lastSuccess := h.lastSuccessfulCycle + h.mu.RUnlock() + return !lastSuccess.IsZero() && !now.Before(lastSuccess) && now.Sub(lastSuccess) <= maxStale +} + +// RoleHandler exposes metrics plus distinct process-liveness and dependency- +// progress readiness endpoints on the allocator's private listener. +func RoleHandler(metrics *Metrics, health *Health, maxStale time.Duration, now func() time.Time) http.Handler { + mux := http.NewServeMux() + mux.Handle("/metrics", MetricsHandler(metrics)) + mux.HandleFunc("/healthz", methodGet(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + _, _ = w.Write([]byte("ok\n")) + })) + mux.HandleFunc("/readyz", methodGet(func(w http.ResponseWriter, _ *http.Request) { + if now == nil || !health.Ready(now(), maxStale) { + http.Error(w, "allocator has no recent successful cycle", http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + _, _ = w.Write([]byte("ready\n")) + })) + return mux +} + +func methodGet(next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + next(w, r) + } +} diff --git a/server/allocator/health_test.go b/server/allocator/health_test.go new file mode 100644 index 00000000..c39da314 --- /dev/null +++ b/server/allocator/health_test.go @@ -0,0 +1,53 @@ +package allocator + +import ( + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestHealthRequiresRecentSuccessfulCycle(t *testing.T) { + health := &Health{} + now := time.Unix(1000, 0) + if health.Ready(now, 30*time.Second) { + t.Fatal("allocator was ready before a successful cycle") + } + health.ObserveSuccessfulCycle(now) + if !health.Ready(now.Add(30*time.Second), 30*time.Second) { + t.Fatal("allocator was not ready at the staleness boundary") + } + if health.Ready(now.Add(30*time.Second+time.Nanosecond), 30*time.Second) { + t.Fatal("stale allocator remained ready") + } + if health.Ready(now.Add(-time.Second), 30*time.Second) { + t.Fatal("clock reversal was accepted as ready") + } +} + +func TestRoleHandlerSeparatesLivenessReadinessAndMetrics(t *testing.T) { + health := &Health{} + now := time.Unix(1000, 0) + handler := RoleHandler(NewMetrics(), health, 30*time.Second, func() time.Time { return now }) + status := func(method, path string) int { + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(method, path, nil)) + return recorder.Code + } + if got := status(http.MethodGet, "/healthz"); got != http.StatusOK { + t.Fatalf("liveness status = %d", got) + } + if got := status(http.MethodGet, "/readyz"); got != http.StatusServiceUnavailable { + t.Fatalf("startup readiness status = %d", got) + } + health.ObserveSuccessfulCycle(now) + if got := status(http.MethodGet, "/readyz"); got != http.StatusOK { + t.Fatalf("successful-cycle readiness status = %d", got) + } + if got := status(http.MethodGet, "/metrics"); got != http.StatusOK { + t.Fatalf("metrics status = %d", got) + } + if got := status(http.MethodPost, "/readyz"); got != http.StatusMethodNotAllowed { + t.Fatalf("readiness mutation status = %d", got) + } +} diff --git a/server/cmd/allocator/main.go b/server/cmd/allocator/main.go index f490d190..809ba460 100644 --- a/server/cmd/allocator/main.go +++ b/server/cmd/allocator/main.go @@ -26,6 +26,7 @@ func main() { kubernetesTokenPath := flag.String("kubernetes-token-path", envOrDefault("COSMIC_CLASH_KUBERNETES_TOKEN_PATH", "/var/run/secrets/kubernetes.io/serviceaccount/token"), "rotating Kubernetes service-account bearer token") kubernetesCAPath := flag.String("kubernetes-ca-path", envOrDefault("COSMIC_CLASH_KUBERNETES_CA_PATH", "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"), "Kubernetes API cluster CA bundle") providerTimeout := flag.Duration("provider-timeout", 10*time.Second, "timeout for each Kubernetes/Agones API request") + readinessMaxStale := flag.Duration("readiness-max-stale", 30*time.Second, "maximum age of the last fully successful allocator cycle") transport := flag.String("transport", envOrDefault("COSMIC_CLASH_TRANSPORT", "enet"), "game transport: enet or steam_sdr") interval := flag.Duration("interval", time.Second, "allocation poll interval") 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") @@ -36,8 +37,11 @@ func main() { if *dsn == "" || *agonesURL == "" { fatalf("--dsn/COSMIC_CLASH_POSTGRES_DSN and --agones-url/COSMIC_CLASH_AGONES_URL are required") } - if (*transport != "enet" && *transport != "steam_sdr") || *interval <= 0 || *providerTimeout <= 0 { - fatalf("--transport must be enet or steam_sdr and --interval/--provider-timeout must be positive") + if (*transport != "enet" && *transport != "steam_sdr") || *interval <= 0 || *providerTimeout <= 0 || *readinessMaxStale <= 0 { + fatalf("--transport must be enet or steam_sdr and --interval/--provider-timeout/--readiness-max-stale must be positive") + } + if *readinessMaxStale < *interval+*providerTimeout { + fatalf("--readiness-max-stale must be at least --interval plus --provider-timeout") } if *allocationQuota < 0 || *allocationQuotaWindow <= 0 { fatalf("--allocation-quota must be non-negative and --allocation-quota-window must be positive") @@ -68,6 +72,7 @@ func main() { log.Printf("allocator: enabled per-replica regional allocation quota=%d window=%s", *allocationQuota, *allocationQuotaWindow) } metrics := allocator.NewMetrics() + health := &allocator.Health{} providerHTTP, err := agones.NewKubernetesHTTPClient(*agonesURL, *kubernetesTokenPath, *kubernetesCAPath, *providerTimeout) if err != nil { fatalf("configure Kubernetes API client: %v", err) @@ -89,7 +94,7 @@ func main() { defer stop() var metricsServer *http.Server if *metricsAddr != "" { - metricsServer = &http.Server{Addr: *metricsAddr, Handler: allocator.MetricsHandler(metrics)} + metricsServer = &http.Server{Addr: *metricsAddr, Handler: allocator.RoleHandler(metrics, health, *readinessMaxStale, now), ReadHeaderTimeout: 5 * time.Second} go func() { if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Printf("allocator: metrics server: %v", err) @@ -107,19 +112,26 @@ func main() { ticker := time.NewTicker(*interval) defer ticker.Stop() for { + cycleHealthy := true servers, err := client.ListReadyServers(ctx) if err != nil && ctx.Err() == nil { + cycleHealthy = false log.Printf("allocator: list Ready GameServers: %v", err) } else { for _, server := range servers { if err := store.RegisterReadyServer(ctx, db, server, now()); err != nil && ctx.Err() == nil { + cycleHealthy = false log.Printf("allocator: register Ready GameServer %s: %v", server.ServerID, err) } } } if _, err := worker.RunOnce(ctx); err != nil && ctx.Err() == nil { + cycleHealthy = false log.Printf("allocator: run once: %v", err) } + if cycleHealthy && ctx.Err() == nil { + health.ObserveSuccessfulCycle(now()) + } select { case <-ctx.Done(): return diff --git a/server/security/test_kubernetes_policies.py b/server/security/test_kubernetes_policies.py index 465f00e2..8c94c170 100644 --- a/server/security/test_kubernetes_policies.py +++ b/server/security/test_kubernetes_policies.py @@ -39,6 +39,7 @@ class KubernetesPolicyTest(unittest.TestCase): "--metrics-addr=:9091", "containerPort: 9091", "key: dsn", "key: secret", "automountServiceAccountToken: true", "--agones-url=https://kubernetes.default.svc", "--provider-timeout=10s", + "--readiness-max-stale=30s", ): self.assertIn(required, deployment) self.assertNotRegex(deployment, r"(?im)^\s*(password|token|private.?key):\s*[^\n]+$") @@ -72,7 +73,7 @@ class KubernetesPolicyTest(unittest.TestCase): "type: RollingUpdate", "maxUnavailable: 0", "maxSurge: 1", "terminationGracePeriodSeconds: 10", "readinessProbe:", "livenessProbe:", - "path: /metrics", "port: metrics", + "path: /readyz", "path: /healthz", "port: metrics", ): self.assertIn(required, deployment)