// Package supervisor contains the small PID-1 lifecycle boundary around an // allocated Godot process. The Agones client is HTTP-only so local/Compose // execution remains independent of the cloud SDK. package supervisor import ( "bytes" "context" "encoding/json" "fmt" "io" "net" "net/http" "net/url" "os" "os/exec" "path/filepath" "strconv" "strings" "time" ) type GameServer struct { // ObjectMeta.Annotations carries per-allocation data the agones package // requests on the GameServerAllocation (server/agones/allocation.go) -- // currently cosmic-clash.io/match-id and cosmic-clash.io/allocation-id. // This is the only channel for match-specific config to reach an // already-Ready pod: env vars are fixed at pod creation, before Agones // assigns a match to it. NOTE: the exact JSON key for this field // (object_meta vs objectMeta) is not independently verified against a // live Agones SDK sidecar from this sandbox; if it turns out wrong, // annotationMatchID simply returns "" and callers fall back to whatever // was explicitly configured, so this degrades safely either way. ObjectMeta struct { Annotations map[string]string `json:"annotations"` } `json:"object_meta"` Status struct { Address string `json:"address"` Ports []struct { Name string `json:"name"` Port int `json:"port"` } `json:"ports"` } `json:"status"` } type Config struct { Command []string Environment []string SDKBaseURL string ReadyURL string Transport string DrainURL string DrainToken string 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, if set, 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 -- this // is for a future Kubernetes-JWT-based WorkloadVerify (server/workload/ // jwt.go), not yet wired server-side. Today the control plane instead // verifies a self-issued signed token (server/workload/signed_token.go), // which reaches this process via the cosmic-clash.io/workload-token // annotation Agones applies to the allocated GameServer (see // server/agones.Client.Allocate) -- see workloadToken() for the // precedence between the two sources. ServerID and ImageDigest are // expected to be populated from the pod spec (Downward API / mounted // build metadata). MatchID may be left empty here and is then read from // the allocated GameServer's own annotations (see GameServer.ObjectMeta // above) -- an explicit value here always wins. ControlPlaneURL string WorkloadTokenPath string ServerID string MatchID string ProtocolVersion int ImageDigest string // AssignmentReadyAttempts/AssignmentReadyBackoff bound the retry loop for // reporting assignment-ready once process-ready has already succeeded. // The control-plane's own durable gate (every participant already // holding a live, unexpired assignment -- see // AdvanceServerRegistrationSQL) may not be satisfied on the very first // attempt if the signed roster is still propagating, and that is // expected, not fatal: unlike a process-ready registration failure, this // does not kill the child, since the process is already legitimately // listening and usable either way. Default 5 attempts, 2s apart. AssignmentReadyAttempts int AssignmentReadyBackoff time.Duration // RosterPath is an operator-mounted writable path where the supervisor // materializes the workload-authenticated signed roster before starting // Godot. It is deliberately separate from WorkloadTokenPath: the former // contains match join envelopes, the latter contains a bearer credential. RosterPath string } type Supervisor struct { config Config client *http.Client cmd *exec.Cmd lastGameServer GameServer } const DefaultDrainGrace = 285 * time.Second func New(config Config) (*Supervisor, error) { if len(config.Command) == 0 || config.Command[0] == "" { return nil, fmt.Errorf("supervisor command is required") } if config.ReadyTimeout <= 0 { config.ReadyTimeout = 30 * time.Second } if config.PollInterval <= 0 { config.PollInterval = 100 * time.Millisecond } if config.AssignmentReadyAttempts <= 0 { config.AssignmentReadyAttempts = 5 } if config.AssignmentReadyBackoff <= 0 { config.AssignmentReadyBackoff = 2 * time.Second } if config.Transport == "" { config.Transport = "enet" } if config.Transport != "enet" && config.Transport != "steam_sdr" { return nil, fmt.Errorf("unsupported transport %q", config.Transport) } if config.HTTPClient == nil { config.HTTPClient = http.DefaultClient } if (config.DrainURL == "") != (config.DrainToken == "") { return nil, fmt.Errorf("drain URL and token must be configured together") } if config.DrainURL != "" { if err := validateLocalDrainURL(config.DrainURL); err != nil { return nil, err } } if config.ControlPlaneURL != "" && (config.ServerID == "" || config.ProtocolVersion < 1 || config.ImageDigest == "") { return nil, fmt.Errorf("control-plane registration requires a server ID, protocol version and image digest") } if config.RosterPath != "" && config.ControlPlaneURL == "" { return nil, fmt.Errorf("roster path requires control-plane URL") } // Neither MatchID nor WorkloadTokenPath is required here: both can // instead be resolved at Start time from the allocated GameServer's own // annotations (see registerControlPlane/workloadToken/matchID). They are // validated to actually be resolvable there, not silently skipped. return &Supervisor{config: config, client: config.HTTPClient}, nil } func validateLocalDrainURL(raw string) error { parsed, err := url.Parse(raw) if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" || parsed.Path == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" { return fmt.Errorf("drain URL must be a loopback HTTP endpoint") } host := parsed.Hostname() if host != "localhost" { ip := net.ParseIP(host) if ip == nil || !ip.IsLoopback() { return fmt.Errorf("drain URL must be a loopback HTTP endpoint") } } return nil } // Start launches the process and marks Agones Ready only after the explicit // readiness probe succeeds. No stdout/log scraping is used. With no SDK URL, // this is direct/Compose mode and the command is simply started. func (s *Supervisor) Start(ctx context.Context) error { env := append([]string(nil), os.Environ()...) env = append(env, s.config.Environment...) if s.config.SDKBaseURL != "" { port, address, err := s.assignedEndpoint(ctx) if err != nil { return err } if s.config.Transport == "steam_sdr" { env = append(env, "SDR_LISTEN_PORT="+strconv.Itoa(port), "SDR_IP="+address+":"+strconv.Itoa(port)) } rosterExpiry, err := s.fetchRoster(ctx) if err != nil { return err } command := withAllocatedConfig(s.config.Command, s.matchID(), s.config.ServerID, s.config.ImageDigest, rosterExpiry) command = withPort(command, port) s.cmd = exec.CommandContext(ctx, command[0], command[1:]...) } else { s.cmd = exec.CommandContext(ctx, s.config.Command[0], s.config.Command[1:]...) } s.cmd.Env = env if err := s.cmd.Start(); err != nil { return err } if s.config.SDKBaseURL == "" { return nil } if err := s.waitReady(ctx); err != nil { _ = s.cmd.Process.Kill() return err } 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 } s.reportAssignmentReady(ctx) return nil } func (s *Supervisor) fetchRoster(ctx context.Context) (time.Time, error) { if s.config.RosterPath == "" { return time.Time{}, nil } matchID := s.matchID() if matchID == "" { return time.Time{}, fmt.Errorf("roster fetch has no match ID") } token, err := s.workloadToken() if err != nil { return time.Time{}, err } rosterURL := strings.TrimRight(s.config.ControlPlaneURL, "/") + "/v1/servers/" + url.PathEscape(s.config.ServerID) + "/roster" request, err := http.NewRequestWithContext(ctx, http.MethodGet, rosterURL, nil) if err != nil { return time.Time{}, err } request.Header.Set("Authorization", "Bearer "+token) response, err := s.client.Do(request) if err != nil { return time.Time{}, err } defer response.Body.Close() if response.StatusCode/100 != 2 { return time.Time{}, fmt.Errorf("control-plane roster returned %s", response.Status) } var roster []json.RawMessage if err := json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&roster); err != nil || len(roster) == 0 { if err == nil { err = fmt.Errorf("empty roster") } return time.Time{}, fmt.Errorf("decode control-plane roster: %w", err) } var expiry time.Time for _, envelope := range roster { if len(envelope) == 0 || string(envelope) == "null" { return time.Time{}, fmt.Errorf("control-plane roster contains an invalid envelope") } var decoded struct { Authorisation struct { ExpiresAt time.Time `json:"expires_at"` } `json:"authorisation"` } if err := json.Unmarshal(envelope, &decoded); err != nil || decoded.Authorisation.ExpiresAt.IsZero() { return time.Time{}, fmt.Errorf("control-plane roster contains an envelope without expiry") } if expiry.IsZero() || decoded.Authorisation.ExpiresAt.Before(expiry) { expiry = decoded.Authorisation.ExpiresAt } } contents, err := json.Marshal(roster) if err != nil { return time.Time{}, fmt.Errorf("encode roster: %w", err) } directory := filepath.Dir(s.config.RosterPath) temporary, err := os.CreateTemp(directory, ".cosmic-clash-roster-*") if err != nil { return time.Time{}, fmt.Errorf("create roster file: %w", err) } temporaryName := temporary.Name() defer os.Remove(temporaryName) if err := temporary.Chmod(0600); err != nil { _ = temporary.Close() return time.Time{}, fmt.Errorf("secure roster file: %w", err) } _, err = temporary.Write(contents) if closeErr := temporary.Close(); err == nil { err = closeErr } if err != nil { return time.Time{}, fmt.Errorf("write roster file: %w", err) } if err := os.Rename(temporaryName, s.config.RosterPath); err != nil { return time.Time{}, fmt.Errorf("install roster file: %w", err) } return expiry, nil } func withAllocatedConfig(command []string, matchID, serverID, imageDigest string, rosterExpiry time.Time) []string { result := append([]string(nil), command...) values := map[string]string{ "match-id": matchID, "server-id": serverID, "server-image-digest": imageDigest, } if !rosterExpiry.IsZero() { values["assignment-expiry-unix"] = strconv.FormatInt(rosterExpiry.Unix(), 10) } for key, value := range values { if value == "" { continue } prefix := "--" + key + "=" replaced := false for i, arg := range result { if strings.HasPrefix(arg, prefix) { result[i] = prefix + value replaced = true break } } if !replaced { result = append(result, prefix+value) } } return result } // reportAssignmentReady is best-effort: process-ready has already succeeded, // so the process is legitimately usable either way. A persistent failure is // written to stderr rather than returned, since treating it as fatal would // kill a perfectly healthy process over what is usually just the signed // roster's durable rows not having propagated yet. func (s *Supervisor) reportAssignmentReady(ctx context.Context) { if s.config.ControlPlaneURL == "" { return } var lastErr error for attempt := 0; attempt < s.config.AssignmentReadyAttempts; attempt++ { if attempt > 0 { select { case <-ctx.Done(): return case <-time.After(s.config.AssignmentReadyBackoff): } } if lastErr = s.registerControlPlane(ctx, true); lastErr == nil { return } } fmt.Fprintf(os.Stderr, "game-server-supervisor: assignment-ready registration did not succeed after %d attempts: %v\n", s.config.AssignmentReadyAttempts, lastErr) } // 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. // matchID resolves the match ID for control-plane registration: an // explicitly configured value always wins, otherwise it falls back to the // cosmic-clash.io/match-id annotation Agones applied to this GameServer at // allocation time (see server/agones.Client.Allocate). Empty if neither is // available. func (s *Supervisor) matchID() string { if s.config.MatchID != "" { return s.config.MatchID } return s.lastGameServer.ObjectMeta.Annotations["cosmic-clash.io/match-id"] } // workloadToken resolves the bearer credential for control-plane // registration. WorkloadTokenPath, when configured, always wins -- it is // for a future Kubernetes-projected-JWT WorkloadVerify path (see the Config // field's doc comment) and an operator who explicitly set it presumably // wants it used. Otherwise it falls back to the cosmic-clash.io/workload- // token annotation Agones applied to this GameServer at allocation time // (server/agones.Client.Allocate, verified by // api.WorkloadVerifierFromSignedToken today) -- the same annotation-fallback // pattern matchID already uses for cosmic-clash.io/match-id. func (s *Supervisor) workloadToken() (string, error) { if s.config.WorkloadTokenPath != "" { 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) } return token, nil } token := s.lastGameServer.ObjectMeta.Annotations["cosmic-clash.io/workload-token"] if token == "" { return "", fmt.Errorf("control-plane registration has no workload token: no --workload-token-path configured, and no cosmic-clash.io/workload-token annotation was present on the allocated GameServer") } return token, nil } func (s *Supervisor) registerControlPlane(ctx context.Context, assignmentReady bool) error { if s.config.ControlPlaneURL == "" { return nil } matchID := s.matchID() if matchID == "" { return fmt.Errorf("control-plane registration has no match ID: not configured, and no cosmic-clash.io/match-id annotation was present on the allocated GameServer") } token, err := s.workloadToken() if err != nil { return err } 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"` }{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, match, readiness stage): a supervisor restart // or a dropped response retrying this exact call must replay, not // conflict. The API enforces a 16-128 byte key; ServerID and MatchID are // both already required non-empty by this point. key := "supervisor-register-" + s.config.ServerID + "-" + matchID + "-" + strconv.FormatBool(assignmentReady) if len(key) > 128 { key = key[:128] } request.Header.Set("Idempotency-Key", key) 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 { result := append([]string(nil), command...) for i, arg := range result { if strings.HasPrefix(arg, "--port=") { result[i] = "--port=" + strconv.Itoa(port) return result } } return append(result, "--port="+strconv.Itoa(port)) } func (s *Supervisor) Wait() error { if s.cmd == nil { return fmt.Errorf("supervisor has not started") } return s.cmd.Wait() } // Run owns the PID-1 termination sequence. The child gets its own context so // cancellation of the supervisor does not kill it before the authenticated // drain request has had a chance to stop new admissions. A non-responsive // child is force-killed after drainGrace; a drain failure is recorded only by // the returned error if the child exits cleanly, while the deadline still // prevents a stuck process from hanging termination forever. func (s *Supervisor) Run(ctx context.Context, drainGrace time.Duration) error { if s == nil || ctx == nil { return fmt.Errorf("supervisor context is required") } if drainGrace <= 0 { drainGrace = DefaultDrainGrace } processCtx, cancel := context.WithCancel(context.Background()) defer cancel() if err := s.Start(processCtx); err != nil { return err } wait := make(chan error, 1) go func() { wait <- s.Wait() }() select { case err := <-wait: return err case <-ctx.Done(): } var drainErr error if s.config.DrainURL != "" { drainCtx, drainCancel := context.WithTimeout(context.Background(), 5*time.Second) drainErr = s.Drain(drainCtx) drainCancel() } var shutdownErr error if s.config.ControlPlaneURL != "" && drainErr == nil { shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) shutdownErr = s.acknowledgeShutdown(shutdownCtx, "server_draining") shutdownCancel() } timer := time.NewTimer(drainGrace) defer timer.Stop() select { case err := <-wait: if drainErr != nil { return fmt.Errorf("child exited after drain failure: %w", drainErr) } if shutdownErr != nil { return fmt.Errorf("child exited after shutdown acknowledgement failure: %w", shutdownErr) } return err case <-timer.C: if s.cmd != nil && s.cmd.Process != nil { _ = s.cmd.Process.Kill() } <-wait if drainErr != nil { return fmt.Errorf("drain failed and child was force-killed: %w", drainErr) } if shutdownErr != nil { return fmt.Errorf("shutdown acknowledgement failed and child was force-killed: %w", shutdownErr) } return fmt.Errorf("child force-killed after drain deadline") } } // Drain asks the allocated Godot process to stop accepting new work. The // token is sent only over the configured localhost control endpoint and is // never placed in command arguments or logs. func (s *Supervisor) Drain(ctx context.Context) error { if s.config.DrainURL == "" || s.config.DrainToken == "" { return fmt.Errorf("authenticated drain endpoint is required") } request, err := http.NewRequestWithContext(ctx, http.MethodPost, s.config.DrainURL, nil) if err != nil { return err } request.Header.Set("Authorization", "Bearer "+s.config.DrainToken) response, err := s.client.Do(request) if err != nil { return err } defer response.Body.Close() if response.StatusCode/100 != 2 { return fmt.Errorf("drain endpoint returned %s", response.Status) } return nil } // acknowledgeShutdown records the supervisor's planned termination after the // local game process has been told to drain. It uses the same bound workload // credential as registration. The idempotency key makes a repeated call safe. func (s *Supervisor) acknowledgeShutdown(ctx context.Context, reason string) error { if s.config.ControlPlaneURL == "" { return nil } matchID := s.matchID() if matchID == "" { return fmt.Errorf("shutdown acknowledgement has no match ID") } token, err := s.workloadToken() if err != nil { return err } body, err := json.Marshal(struct { Reason string `json:"reason"` }{reason}) if err != nil { return err } endpoint := strings.TrimRight(s.config.ControlPlaneURL, "/") + "/v1/servers/" + url.PathEscape(s.config.ServerID) + "/shutdown" 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) key := "supervisor-shutdown-" + s.config.ServerID + "-" + matchID + "-" + reason if len(key) > 128 { key = key[:128] } request.Header.Set("Idempotency-Key", key) 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 shutdown returned %s", response.Status) } return nil } func (s *Supervisor) assignedEndpoint(ctx context.Context) (int, string, error) { var server GameServer if err := s.sdkGet(ctx, "/gameserver", &server); err != nil { return 0, "", err } s.lastGameServer = server if len(server.Status.Ports) == 0 || strings.TrimSpace(server.Status.Address) == "" || strings.ContainsAny(server.Status.Address, " \t\r\n") { return 0, "", fmt.Errorf("Agones returned no assigned endpoint") } for _, port := range server.Status.Ports { if port.Port > 0 && port.Port <= 65535 && (port.Name == "game" || len(server.Status.Ports) == 1) { return port.Port, server.Status.Address, nil } } return 0, "", fmt.Errorf("Agones returned no usable game port") } func (s *Supervisor) waitReady(ctx context.Context) error { if s.config.ReadyURL == "" { return fmt.Errorf("allocated mode requires an explicit readiness URL") } deadline := time.NewTimer(s.config.ReadyTimeout) defer deadline.Stop() for { request, err := http.NewRequestWithContext(ctx, http.MethodGet, s.config.ReadyURL, nil) if err == nil { response, requestErr := s.client.Do(request) if requestErr == nil { _ = response.Body.Close() if response.StatusCode >= 200 && response.StatusCode < 300 { return nil } } } select { case <-ctx.Done(): return ctx.Err() case <-deadline.C: return fmt.Errorf("process-ready probe timed out") case <-time.After(s.config.PollInterval): } } } func (s *Supervisor) sdkGet(ctx context.Context, path string, target any) error { request, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(s.config.SDKBaseURL, "/")+path, nil) if err != nil { return err } response, err := s.client.Do(request) if err != nil { return err } defer response.Body.Close() if response.StatusCode/100 != 2 { return fmt.Errorf("Agones GET %s returned %s", path, response.Status) } return json.NewDecoder(response.Body).Decode(target) } func (s *Supervisor) sdkPost(ctx context.Context, path string) error { request, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(s.config.SDKBaseURL, "/")+path, nil) if err != nil { return err } response, err := s.client.Do(request) if err != nil { return err } defer response.Body.Close() if response.StatusCode/100 != 2 { return fmt.Errorf("Agones POST %s returned %s", path, response.Status) } return nil }