// 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 ( "context" "encoding/json" "fmt" "net/http" "os" "os/exec" "strconv" "strings" "time" ) type GameServer struct { 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 } type Supervisor struct { config Config client *http.Client cmd *exec.Cmd } 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.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 } return &Supervisor{config: config, client: config.HTTPClient}, 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)) } command := withPort(s.config.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 } return s.sdkPost(ctx, "/ready") } 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() } // 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 } 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 } 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 }