mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
735 lines
25 KiB
Go
735 lines
25 KiB
Go
// 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, cosmic-clash.io/allocation-id, and
|
|
// allocator-selected compatibility fields.
|
|
// 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
|
|
DefaultHTTPTimeout = 10 * 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.Client{Timeout: DefaultHTTPTimeout}
|
|
}
|
|
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, err = withAllocatedCompatibility(command, s.lastGameServer)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
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
|
|
}
|
|
|
|
// withAllocatedCompatibility overlays fields selected by the allocator onto
|
|
// child flags. These values arrive through Agones allocation annotations after
|
|
// the pod was created, so static Fleet defaults must never win over them.
|
|
func withAllocatedCompatibility(command []string, gameServer GameServer) ([]string, error) {
|
|
values := map[string]string{}
|
|
annotations := gameServer.ObjectMeta.Annotations
|
|
if annotations == nil {
|
|
return command, nil
|
|
}
|
|
for annotation, flag := range map[string]string{
|
|
"cosmic-clash.io/arena-path": "arena-path",
|
|
"cosmic-clash.io/playlist": "playlist",
|
|
"cosmic-clash.io/region": "region",
|
|
"cosmic-clash.io/build": "client-build",
|
|
"cosmic-clash.io/protocol": "protocol-version",
|
|
"cosmic-clash.io/transport": "transport",
|
|
} {
|
|
value := annotations[annotation]
|
|
if value == "" {
|
|
continue
|
|
}
|
|
if strings.ContainsAny(value, "\r\n\t") {
|
|
return nil, fmt.Errorf("allocated annotation %q contains control characters", annotation)
|
|
}
|
|
values[flag] = value
|
|
}
|
|
return withAllocatedValues(command, values), nil
|
|
}
|
|
|
|
func withAllocatedValues(command []string, values map[string]string) []string {
|
|
result := append([]string(nil), command...)
|
|
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
|
|
}
|