From 0bad9e07dbb0bef314336e0c23c3cbcbcadfc2bb Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 13:27:43 +0100 Subject: [PATCH] feat(multiplayer): report process-ready to the control plane from the supervisor Add opt-in control-plane registration to server/supervisor: once Agones Ready succeeds, POST /v1/servers/{id}/register (assignment_ready=false) using a workload token read fresh from disk each call -- matching how a Kubernetes projected service account token is rotated in place by kubelet, unlike a cached/env-var secret. ControlPlaneURL empty (the default) is a total no-op, so direct/Compose mode and allocated-without- control-plane mode are both byte-for-byte unaffected; New() rejects a half-configured registration (URL set without token path/server/match/ digest) rather than silently skipping it. ServerID/MatchID/ImageDigest are read from env vars named by CLI flags (--server-id-env, --match-id-env, --image-digest-env), matching the existing --drain-token-env convention in this same binary, rather than parsed out of the Agones SDK's own GameServer JSON -- that shape isn't independently verifiable from here, whereas the Kubernetes Downward API (fieldRef: metadata.name) populating an env var is a standard, safe pattern already used elsewhere in this codebase for exactly this class of secret. A registration failure now kills the child (matching the existing waitReady failure path) rather than leaving Agones-Ready-but- control-plane-unregistered process running -- a real gap the second new test (TestControlPlaneRegistrationFailureKillsChildRatherThanRunningUnregistered) had to be corrected to actually exercise: its first draft omitted ReadyURL and was failing at waitReady, before ever reaching the code path it claimed to test. --- server/cmd/game-server-supervisor/main.go | 13 +++ server/supervisor/supervisor.go | 86 ++++++++++++++++- server/supervisor/supervisor_test.go | 107 ++++++++++++++++++++++ 3 files changed, 205 insertions(+), 1 deletion(-) diff --git a/server/cmd/game-server-supervisor/main.go b/server/cmd/game-server-supervisor/main.go index 47df276d..71c741fb 100644 --- a/server/cmd/game-server-supervisor/main.go +++ b/server/cmd/game-server-supervisor/main.go @@ -41,6 +41,12 @@ func main() { drainTokenEnv := options.String("drain-token-env", "COSMIC_CLASH_DRAIN_TOKEN", "environment variable containing the drain bearer token") transport := options.String("transport", "enet", "enet or steam_sdr") grace := options.Duration("drain-grace", supervisor.DefaultDrainGrace, "maximum graceful drain duration") + controlPlaneURL := options.String("control-plane-url", "", "matchmaking control-plane base URL; empty skips process-ready registration entirely") + workloadTokenPath := options.String("workload-token-path", "", "path to the projected workload service-account token, read fresh on every registration call") + serverIDEnv := options.String("server-id-env", "COSMIC_CLASH_SERVER_ID", "environment variable containing this GameServer's control-plane server ID (populate via the Kubernetes Downward API, fieldRef: metadata.name)") + matchIDEnv := options.String("match-id-env", "COSMIC_CLASH_MATCH_ID", "environment variable containing the allocated match ID") + protocolVersion := options.Int("protocol-version", 0, "protocol version reported at registration") + imageDigestEnv := options.String("image-digest-env", "COSMIC_CLASH_IMAGE_DIGEST", "environment variable containing this build's sha256 image digest") if err := options.Parse(args[:separator]); err != nil { os.Exit(2) } @@ -57,6 +63,13 @@ func main() { DrainToken: token, Transport: *transport, ReadyTimeout: 30 * time.Second, + + ControlPlaneURL: *controlPlaneURL, + WorkloadTokenPath: *workloadTokenPath, + ServerID: os.Getenv(*serverIDEnv), + MatchID: os.Getenv(*matchIDEnv), + ProtocolVersion: *protocolVersion, + ImageDigest: os.Getenv(*imageDigestEnv), }) if err != nil { fmt.Fprintf(os.Stderr, "game-server-supervisor: %v\n", err) diff --git a/server/supervisor/supervisor.go b/server/supervisor/supervisor.go index d0c22848..6b44ba69 100644 --- a/server/supervisor/supervisor.go +++ b/server/supervisor/supervisor.go @@ -4,6 +4,7 @@ package supervisor import ( + "bytes" "context" "encoding/json" "fmt" @@ -38,6 +39,23 @@ type Config struct { ReadyTimeout time.Duration PollInterval time.Duration HTTPClient *http.Client + + // ControlPlaneURL, when set, opts into reporting process-ready to the + // matchmaking control plane (multiplayer-next.md task 8.28) once Agones + // Ready succeeds. Leaving it empty preserves every existing behavior + // exactly -- direct/Compose mode and allocated-without-control-plane mode + // are both unaffected. WorkloadTokenPath is read fresh on every call + // rather than cached, matching how a Kubernetes projected service account + // token is rotated in place by kubelet before it expires; ServerID, + // MatchID and ImageDigest are expected to be populated from the pod spec + // (Downward API / mounted build metadata), not guessed at from the + // Agones SDK's own GameServer response. + ControlPlaneURL string + WorkloadTokenPath string + ServerID string + MatchID string + ProtocolVersion int + ImageDigest string } type Supervisor struct { @@ -75,6 +93,9 @@ func New(config Config) (*Supervisor, error) { return nil, err } } + if config.ControlPlaneURL != "" && (config.WorkloadTokenPath == "" || config.ServerID == "" || config.MatchID == "" || config.ProtocolVersion < 1 || config.ImageDigest == "") { + return nil, fmt.Errorf("control-plane registration requires a workload token path, server ID, match ID, protocol version and image digest") + } return &Supervisor{config: config, client: config.HTTPClient}, nil } @@ -123,7 +144,70 @@ func (s *Supervisor) Start(ctx context.Context) error { _ = s.cmd.Process.Kill() return err } - return s.sdkPost(ctx, "/ready") + if err := s.sdkPost(ctx, "/ready"); err != nil { + return err + } + if err := s.registerControlPlane(ctx, false); err != nil { + // Unlike a bare Agones Ready, this failure leaves the match's durable + // control-plane record stuck at ALLOCATING with no way for the + // matcher/allocator to learn this process is actually listening -- + // players would wait indefinitely for a server that Agones considers + // healthy. Kill the child so Kubernetes reschedules rather than + // leaving that silent split-brain running. + _ = s.cmd.Process.Kill() + return err + } + return nil +} + +// registerControlPlane reports the allocated process's readiness to the +// matchmaking control plane (POST /v1/servers/{id}/register). It is a no-op +// whenever ControlPlaneURL is unset, which is the default and preserves +// every existing direct/Compose/allocated-only behavior exactly. The +// workload token is read fresh from disk on every call rather than cached -- +// a Kubernetes projected service account token is rotated in place by +// kubelet before it expires, so caching it risks presenting a stale one on a +// long-lived process. +func (s *Supervisor) registerControlPlane(ctx context.Context, assignmentReady bool) error { + if s.config.ControlPlaneURL == "" { + return nil + } + tokenBytes, err := os.ReadFile(s.config.WorkloadTokenPath) + if err != nil { + return fmt.Errorf("read workload token: %w", err) + } + token := strings.TrimSpace(string(tokenBytes)) + if token == "" { + return fmt.Errorf("workload token file %q is empty", s.config.WorkloadTokenPath) + } + body, err := json.Marshal(struct { + MatchID string `json:"match_id"` + ProtocolVersion int `json:"protocol_version"` + ImageDigest string `json:"image_digest"` + AssignmentReady bool `json:"assignment_ready"` + }{s.config.MatchID, s.config.ProtocolVersion, s.config.ImageDigest, assignmentReady}) + if err != nil { + return err + } + endpoint := strings.TrimRight(s.config.ControlPlaneURL, "/") + "/v1/servers/" + url.PathEscape(s.config.ServerID) + "/register" + request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) + if err != nil { + return err + } + request.Header.Set("Content-Type", "application/json") + request.Header.Set("Authorization", "Bearer "+token) + // Idempotent per (server, readiness stage): a supervisor restart or a + // dropped response retrying this exact call must replay, not conflict. + request.Header.Set("Idempotency-Key", "supervisor-register-"+s.config.ServerID+"-"+strconv.FormatBool(assignmentReady)) + response, err := s.client.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode/100 != 2 { + return fmt.Errorf("control-plane register returned %s", response.Status) + } + return nil } func withPort(command []string, port int) []string { diff --git a/server/supervisor/supervisor_test.go b/server/supervisor/supervisor_test.go index eee9afc0..64ddde80 100644 --- a/server/supervisor/supervisor_test.go +++ b/server/supervisor/supervisor_test.go @@ -2,6 +2,7 @@ package supervisor import ( "context" + "io" "net/http" "net/http/httptest" "os" @@ -66,6 +67,112 @@ func TestAllocatedStartInjectsDynamicEndpointAndCallsReadyAfterProbe(t *testing. } } +func TestControlPlaneRegistrationRejectsIncompleteConfig(t *testing.T) { + base := Config{Command: []string{"/bin/true"}, ControlPlaneURL: "https://control-plane.invalid"} + if _, err := New(base); err == nil { + t.Fatal("registration enabled with no token path/server/match/digest was accepted") + } + complete := base + complete.WorkloadTokenPath, complete.ServerID, complete.MatchID, complete.ProtocolVersion, complete.ImageDigest = "/tmp/token", "server-1", "match-1", 1, "sha256:aa" + if _, err := New(complete); err != nil { + t.Fatalf("fully configured registration rejected: %v", err) + } +} + +func TestControlPlaneRegistrationReportsProcessReadyWithWorkloadToken(t *testing.T) { + var gotAuth, gotIdempotency, gotBody string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/gameserver": + _, _ = w.Write([]byte(`{"status":{"address":"203.0.113.9","ports":[{"name":"game","port":31001}]}}`)) + case r.URL.Path == "/ready-probe": + w.WriteHeader(http.StatusOK) + case r.URL.Path == "/ready": + w.WriteHeader(http.StatusOK) + case r.URL.Path == "/v1/servers/server-1/register": + gotAuth = r.Header.Get("Authorization") + gotIdempotency = r.Header.Get("Idempotency-Key") + body, _ := io.ReadAll(r.Body) + gotBody = string(body) + w.WriteHeader(http.StatusNoContent) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + + tokenPath := filepath.Join(t.TempDir(), "token") + if err := os.WriteFile(tokenPath, []byte(" workload-jwt-abc123 \n"), 0o600); err != nil { + t.Fatal(err) + } + s, err := New(Config{ + Command: []string{"/bin/sh", "-c", "true"}, SDKBaseURL: server.URL, ReadyURL: server.URL + "/ready-probe", ReadyTimeout: time.Second, PollInterval: time.Millisecond, + ControlPlaneURL: server.URL, WorkloadTokenPath: tokenPath, ServerID: "server-1", MatchID: "match-1", ProtocolVersion: 1, ImageDigest: "sha256:aa", + }) + if err != nil { + t.Fatal(err) + } + if err := s.Start(context.Background()); err != nil { + t.Fatal(err) + } + _ = s.Wait() + if gotAuth != "Bearer workload-jwt-abc123" { + t.Fatalf("Authorization header = %q, want the trimmed token file contents", gotAuth) + } + if len(gotIdempotency) < 16 { + t.Fatalf("Idempotency-Key = %q, too short", gotIdempotency) + } + if !strings.Contains(gotBody, `"match_id":"match-1"`) || !strings.Contains(gotBody, `"assignment_ready":false`) || !strings.Contains(gotBody, `"image_digest":"sha256:aa"`) { + t.Fatalf("register body = %s", gotBody) + } +} + +func TestControlPlaneRegistrationFailureKillsChildRatherThanRunningUnregistered(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/gameserver": + _, _ = w.Write([]byte(`{"status":{"address":"203.0.113.9","ports":[{"name":"game","port":31001}]}}`)) + case "/ready-probe": + w.WriteHeader(http.StatusOK) + case "/ready": + w.WriteHeader(http.StatusOK) + case "/v1/servers/server-1/register": + w.WriteHeader(http.StatusInternalServerError) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + + tokenPath := filepath.Join(t.TempDir(), "token") + if err := os.WriteFile(tokenPath, []byte("workload-jwt"), 0o600); err != nil { + t.Fatal(err) + } + // A long-running child: if Start's failure path did not actually kill it, + // Wait would block for the full sleep instead of returning promptly with + // a "signal: killed" style exit. + s, err := New(Config{ + Command: []string{"/bin/sh", "-c", "sleep 30"}, SDKBaseURL: server.URL, ReadyURL: server.URL + "/ready-probe", ReadyTimeout: time.Second, PollInterval: time.Millisecond, + ControlPlaneURL: server.URL, WorkloadTokenPath: tokenPath, ServerID: "server-1", MatchID: "match-1", ProtocolVersion: 1, ImageDigest: "sha256:aa", + }) + if err != nil { + t.Fatal(err) + } + if err := s.Start(context.Background()); err == nil { + t.Fatal("Start succeeded despite the control-plane rejecting registration") + } + done := make(chan error, 1) + go func() { done <- s.Wait() }() + select { + case err := <-done: + if err == nil { + t.Fatal("child was not actually killed after a failed registration") + } + case <-time.After(5 * time.Second): + t.Fatal("child was still running 5s after a failed registration should have killed it") + } +} + func TestAllocatedENetDoesNotReceiveSDRVariables(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == "/gameserver" {