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.
2026-09-10 16:04:04 +00:00 · 2026-09-01 13:27:43 +01:00
parent 8b76f80b5c
commit 0bad9e07db
3 changed files with 205 additions and 1 deletions
@@ -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)
@@ -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 {
@@ -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" {