package main
import (
"encoding/json"
"flag"
"fmt"
"net"
"os"
"slices"
"strings"
"text/tabwriter"
"time"
"github.com/ademidoff/supavisor/internal/api"
"github.com/ademidoff/supavisor/internal/version"
)
const (
tabPadding = 3
// Output formats. The table is for people and its columns are a
// presentation choice; json is the payload the daemon actually sent, for
// anything reading sctl rather than looking at it.
outputTable = "table"
outputJSON = "json"
// States a program has a live PID in, and the placeholders used when a
// column has nothing to report
stateRunning = "RUNNING"
stateStarting = "STARTING"
stateStopping = "STOPPING"
healthNone = "NONE"
notAvailable = "N/A"
// dialTimeout fails fast when the daemon is not listening.
dialTimeout = 5 * time.Second
// requestTimeout has to outlast a command's own work: stopping a process
// that ignores SIGINT takes the full graceful shutdown timeout before it is
// killed, and a start waits for the process to come up.
requestTimeout = 60 * time.Second
)
func main() {
var socketPath string
var output string
var showVersion bool
flag.StringVar(&socketPath, "s", "/tmp/supavisor.sock", "Path to supavisor socket")
flag.StringVar(&socketPath, "socket", "/tmp/supavisor.sock", "Path to supavisor socket")
flag.StringVar(&output, "o", outputTable, "Output format: table or json")
flag.StringVar(&output, "output", outputTable, "Output format: table or json")
flag.BoolVar(&showVersion, "version", false, "Print version information and exit")
flag.Usage = printUsage
flag.Parse()
if showVersion {
fmt.Println(version.String("sctl"))
return
}
if output != outputTable && output != outputJSON {
fmt.Fprintf(os.Stderr, "Error: unknown output format: %s (must be %s or %s)\n", output, outputTable, outputJSON)
os.Exit(1)
}
if flag.NArg() == 0 {
printUsage()
os.Exit(1)
}
command := flag.Arg(0)
args := flag.Args()[1:]
// Go's flag package stops parsing at the first non-flag argument, so an
// option placed after the command would be silently ignored and we would
// talk to the default socket instead of the requested one.
for _, arg := range args {
if strings.HasPrefix(arg, "-") {
fmt.Fprintf(os.Stderr, "Error: options must be given before the command: %s\n\n", arg)
printUsage()
os.Exit(1)
}
}
resp, err := sendRequest(socketPath, command, args)
if err != nil {
fmt.Fprintf(os.Stderr, "Fatal: %v\n", err)
os.Exit(1)
}
// JSON reports what the daemon sent, whether or not the command succeeded,
// so a caller reading sctl never has to scrape stderr to find out what went
// wrong. The exit code still carries the outcome.
if output == outputJSON {
if err := printJSON(*resp); err != nil {
fmt.Fprintf(os.Stderr, "Fatal: %v\n", err)
os.Exit(1)
}
if !resp.Success {
os.Exit(1)
}
return
}
// Handle response
if !resp.Success {
fmt.Fprintf(os.Stderr, "Error: %s\n", resp.Message)
os.Exit(1)
}
// Print response based on command
switch command {
case api.CommandStatus:
// A named program gets the detail view, which has room for why it is
// not running; the table stays narrow enough to read.
if len(args) > 0 {
printProcessDetail(*resp)
} else {
printStatus(*resp)
}
case api.CommandReload:
printReload(*resp)
default:
fmt.Println(resp.Message)
}
}
// printJSON writes the daemon's response as indented JSON. It is the response
// itself rather than a shape invented here, so the field names are the ones the
// protocol already defines and they are the same for every command.
func printJSON(resp api.Response) error {
encoded, err := json.MarshalIndent(resp, "", " ")
if err != nil {
return fmt.Errorf("failed to render the response as JSON: %w", err)
}
fmt.Println(string(encoded))
return nil
}
// printReload reports what the reload applied. Each category is named only when
// it has something in it, so a reload that changed one program says so in one
// line rather than printing two empty lists beside it.
func printReload(resp api.Response) {
fmt.Println(resp.Message)
applied, ok := reloadDiff(resp)
if !ok {
return
}
for _, category := range []struct {
label string
names []string
}{
{"added", applied.Added},
{"removed", applied.Removed},
{"changed", applied.Changed},
} {
if len(category.names) > 0 {
fmt.Printf(" %s: %s\n", category.label, strings.Join(category.names, ", "))
}
}
}
// reloadDiff recovers the typed diff from a response whose payload has already
// been decoded into the generic shape JSON gives us
func reloadDiff(resp api.Response) (api.ReloadResponse, bool) {
var applied api.ReloadResponse
raw, err := json.Marshal(resp.Data)
if err != nil {
return applied, false
}
if err := json.Unmarshal(raw, &applied); err != nil {
return applied, false
}
return applied, true
}
func printStatus(resp api.Response) {
rows, ok := statusRows(resp)
if !ok {
fmt.Println(resp.Message)
return
}
w := tabwriter.NewWriter(os.Stdout, 0, 0, tabPadding, ' ', 0)
fmt.Fprintln(w, "NAME\tSTATE\tDESIRED\tHEALTH\tPID\tEXIT_CODE\tRESTARTS\tUPTIME")
fmt.Fprintln(w, "----\t-----\t-------\t------\t---\t---------\t--------\t------")
for _, row := range rows {
state := getString(row, "state")
fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%d\t%d\t%s\n",
getString(row, "name"), state, getString(row, "desired"),
healthColumn(getString(row, "health")), pidColumn(state, getInt(row, "pid")),
getInt(row, "exit_code"), getInt(row, "restart_count"), getString(row, "uptime"))
}
_ = w.Flush()
}
// printProcessDetail reports on a single program, including why it is not
// running when it is wanted but is not
func printProcessDetail(resp api.Response) {
rows, ok := statusRows(resp)
if !ok || len(rows) == 0 {
fmt.Println(resp.Message)
return
}
row := rows[0]
state := getString(row, "state")
w := tabwriter.NewWriter(os.Stdout, 0, 0, tabPadding, ' ', 0)
fmt.Fprintf(w, "Name:\t%s\n", getString(row, "name"))
fmt.Fprintf(w, "State:\t%s\n", state)
fmt.Fprintf(w, "Desired:\t%s\n", getString(row, "desired"))
fmt.Fprintf(w, "Health:\t%s\n", healthColumn(getString(row, "health")))
fmt.Fprintf(w, "PID:\t%s\n", pidColumn(state, getInt(row, "pid")))
fmt.Fprintf(w, "Exit code:\t%d\n", getInt(row, "exit_code"))
fmt.Fprintf(w, "Restarts:\t%d\n", getInt(row, "restart_count"))
fmt.Fprintf(w, "Uptime:\t%s\n", getString(row, "uptime"))
// Only a program with something to explain has a reason: it is held back by
// a dependency, or it stopped trying on its own.
if reason := getString(row, "reason"); reason != "" {
fmt.Fprintf(w, "Reason:\t%s\n", reason)
}
_ = w.Flush()
}
// statusRows extracts the process list from a status response
func statusRows(resp api.Response) ([]map[string]any, bool) {
data, ok := resp.Data.(map[string]any)
if !ok {
return nil, false
}
processes, ok := data["processes"].([]any)
if !ok {
return nil, false
}
rows := make([]map[string]any, 0, len(processes))
for _, p := range processes {
row, ok := p.(map[string]any)
if !ok {
continue
}
rows = append(rows, row)
}
return rows, true
}
// healthColumn renders the health of a program. Programs without a health
// check, and programs that are not running, have nothing to report.
func healthColumn(health string) string {
if health == "" || health == healthNone {
return "-"
}
return health
}
// pidColumn renders the PID of a program that has one
func pidColumn(state string, pid int) string {
if slices.Contains([]string{stateRunning, stateStarting, stateStopping}, state) {
return fmt.Sprintf("%d", pid)
}
return notAvailable
}
func getString(m map[string]any, key string) string {
val, ok := m[key]
if !ok {
return ""
}
s, ok := val.(string)
if !ok {
return ""
}
return s
}
func getInt(m map[string]any, key string) int {
val, ok := m[key]
if !ok {
return 0
}
// JSON numbers are float64
f, ok := val.(float64)
if ok {
return int(f)
}
// Try int directly
i, ok := val.(int)
if ok {
return i
}
return 0
}
// sendRequest connects to supavisor, sends a request, and returns the response.
// It includes timeouts to prevent hanging when the daemon is not running.
func sendRequest(socketPath, command string, args []string) (*api.Response, error) {
// Connect to supavisor with timeout
dialer := net.Dialer{
Timeout: dialTimeout,
}
conn, err := dialer.Dial("unix", socketPath)
if err != nil {
return nil, fmt.Errorf("failed to connect to supavisor: %w\nMake sure the supavisor daemon is running", err)
}
defer conn.Close()
// Set read and write deadlines to prevent hanging
deadline := time.Now().Add(requestTimeout)
if err := conn.SetDeadline(deadline); err != nil {
return nil, fmt.Errorf("failed to set connection deadline: %w", err)
}
// Send request
req := api.Request{
Command: command,
Args: args,
}
encoder := json.NewEncoder(conn)
if err := encoder.Encode(&req); err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
// Receive response
decoder := json.NewDecoder(conn)
var resp api.Response
if err := decoder.Decode(&resp); err != nil {
return nil, fmt.Errorf("failed to receive response: %w\nMake sure the supavisor daemon is running and responding", err)
}
return &resp, nil
}
func printUsage() {
fmt.Println("Usage: sctl [OPTIONS] COMMAND [ARGS]")
fmt.Println()
fmt.Println("Commands:")
fmt.Println(" status Show status of all processes")
fmt.Println(" status <name> Show one process in detail, including why it is not running")
fmt.Println(" start <name> Start a process")
fmt.Println(" stop <name> Stop a process")
fmt.Println(" restart <name> Restart a process")
fmt.Println(" reload Reload configuration")
fmt.Println(" shutdown Shutdown supavisor")
fmt.Println()
fmt.Println("Options (must precede the command):")
fmt.Println(" -s, -socket PATH Path to supavisor socket (default: /tmp/supavisor.sock)")
fmt.Println(" -o, -output FORMAT Output format: table (default) or json")
fmt.Println(" -version Print version information and exit")
}
//nolint:goconst
package main
import (
"flag"
"fmt"
"io"
"log/slog"
"os"
"path/filepath"
"strings"
"golang.org/x/term"
"github.com/ademidoff/supavisor/internal/config"
"github.com/ademidoff/supavisor/internal/server"
"github.com/ademidoff/supavisor/internal/version"
)
// parseLogLevel converts a string log level to slog.Level
func parseLogLevel(level string) (slog.Level, error) {
switch strings.ToLower(strings.TrimSpace(level)) {
case "debug":
return slog.LevelDebug, nil
case "info":
return slog.LevelInfo, nil
case "warn":
return slog.LevelWarn, nil
case "error":
return slog.LevelError, nil
default:
return slog.LevelInfo, fmt.Errorf("invalid log level %s: must be debug, info, warn, or error", level)
}
}
func main() { //nolint:gocyclo,funlen
var configPath string
var logFilePath string
var showVersion bool
flag.StringVar(&configPath, "c", "/etc/supavisor/supavisor.yml", "Path to configuration file")
flag.StringVar(&configPath, "config", "/etc/supavisor/supavisor.yml", "Path to configuration file")
flag.StringVar(&logFilePath, "logfile", "", "Optional path to log file (logs go to stdout only in interactive mode)")
flag.BoolVar(&showVersion, "version", false, "Print version information and exit")
flag.Parse()
if showVersion {
fmt.Println(version.String("supavisor"))
return
}
if configPath == "" {
fmt.Fprintf(os.Stderr, "Error: configuration file path is required\n")
os.Exit(1)
}
// Parse configuration (loads supavisor.yml and any sibling supavisor.d/*.yml fragments)
cfg, err := config.ParseConfig(configPath)
if err != nil {
fmt.Fprintf(os.Stderr, "Error: failed to parse configuration: %v\n", err)
os.Exit(1)
}
// Use logfile from config if not specified via flag
if logFilePath == "" && cfg.Supavisor.LogFile != "" {
logFilePath = cfg.Supavisor.LogFile
}
// Setup logging
// Detect if stdout is a TTY (interactive terminal)
isTTY := term.IsTerminal(int(os.Stdout.Fd()))
var output io.Writer
// If a log file is specified, always use it (and only it)
// This is standard daemon behavior - logs go to the file, not stdout
if logFilePath != "" {
// Ensure log directory exists
if dir := filepath.Dir(logFilePath); dir != "" && dir != "." {
if err = os.MkdirAll(dir, 0o755); err != nil {
fmt.Fprintf(os.Stderr, "Error: failed to create log directory: %v\n", err)
os.Exit(1)
}
}
// Open log file for appending
var logFile *os.File
logFile, err = os.OpenFile(logFilePath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
if err != nil {
fmt.Fprintf(os.Stderr, "Error: failed to open log file: %v\n", err)
os.Exit(1)
}
// We don't close logFile here because it needs to stay open for the logger
// In a real daemon, we might want to handle rotation or closure on exit,
// but main() exit closes files anyway.
// Write logs only to the file (standard daemon behavior)
output = logFile
} else if isTTY {
// No logfile specified, use stdout only in interactive mode
output = os.Stdout
} else {
// No logfile and non-interactive mode, discard logs
output = io.Discard
}
replaceAttr := func(groups []string, a slog.Attr) slog.Attr {
if a.Key == slog.LevelKey {
level := a.Value.Any().(slog.Level)
a.Value = slog.StringValue(strings.ToLower(level.String()))
}
return a
}
// Parse log level
logLevel, err := parseLogLevel(cfg.Supavisor.LogLevel)
if err != nil {
fmt.Fprintf(os.Stderr, "Error: %v\n", err)
os.Exit(1)
}
var handler slog.Handler
opts := &slog.HandlerOptions{
Level: logLevel,
ReplaceAttr: replaceAttr,
}
switch cfg.Supavisor.LogFormat {
case "json":
handler = slog.NewJSONHandler(output, opts)
default:
handler = slog.NewTextHandler(output, opts)
}
// Create a logger with setup component
logger := slog.New(handler)
slog.SetDefault(logger)
l := logger.With("component", "setup")
l.Info("Setup completed.")
// Create supavisor with main component logger
sv, err := server.New(cfg, logger)
if err != nil {
l.Error("Failed to create supavisor", "error", err)
os.Exit(1)
}
// Start supavisor
if err := sv.Start(); err != nil {
l.Error("Failed to start supavisor", "error", err)
os.Exit(1)
}
// Wait forever (supavisor will handle signals)
select {}
}
// Package api contains the API messages for the supavisor and sctl.
package api
// Commands understood by the daemon. Both sides of the socket share these so
// the protocol is defined in one place.
const (
CommandStatus = "status"
CommandStart = "start"
CommandStop = "stop"
CommandRestart = "restart"
CommandReload = "reload"
CommandShutdown = "shutdown"
)
// Request represents a request from the CLI
type Request struct {
Command string `json:"command"`
Args []string `json:"args"`
}
// Response represents a response from the daemon
type Response struct {
Data interface{} `json:"data,omitempty"`
Message string `json:"message,omitempty"`
Success bool `json:"success"`
}
// ProcessStatus represents the status of a process
type ProcessStatus struct {
Name string `json:"name"`
State string `json:"state"`
Desired string `json:"desired"`
Health string `json:"health"`
Uptime string `json:"uptime"`
// Reason is why a program that is wanted is not running yet, and is absent
// whenever nothing is holding it back.
Reason string `json:"reason,omitempty"`
PID int `json:"pid"`
ExitCode int `json:"exit_code"`
RestartCount int `json:"restart_count"`
}
// StatusResponse represents a status response
type StatusResponse struct {
Processes []ProcessStatus `json:"processes"`
}
// ReloadResponse reports what a reload did. A caller driving supavisor over the
// socket needs this to tell an applied change from a no-op, and a program that
// reloads its own supervisor needs it to know whether it was in Changed and is
// therefore about to be stopped and replaced.
type ReloadResponse struct {
Added []string `json:"added,omitempty"`
Removed []string `json:"removed,omitempty"`
Changed []string `json:"changed,omitempty"`
}
// Empty reports whether the reload left every program alone.
func (r ReloadResponse) Empty() bool {
return len(r.Added) == 0 && len(r.Removed) == 0 && len(r.Changed) == 0
}
package config
import (
"bytes"
"errors"
"fmt"
"io"
"maps"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"syscall"
"gopkg.in/yaml.v3"
)
// RestartPolicy represents the restart behavior for a process
type RestartPolicy string
const (
RestartAlways RestartPolicy = "always"
RestartNever RestartPolicy = "never"
RestartUnexpected RestartPolicy = "unexpected"
)
const (
defaultLogFileMaxBytes = 50 * 1024 * 1024
// defaultStopWaitSecs is how long a process gets to exit on its stop signal
// before it is killed.
defaultStopWaitSecs = 10
defaultLogFileBackups = 10
defaultStartSecs = 1
defaultMaxRestarts = 3
defaultPriority = 999
// maxSocketPathLen is the smallest sockaddr_un limit across the platforms
// supavisor runs on (104 on macOS, 108 on Linux), including the terminator.
maxSocketPathLen = 104
)
// intOrDefault returns the configured value, treating an absent setting rather
// than a zero one as "not configured".
func intOrDefault(configured *int, fallback int) int {
if configured == nil {
return fallback
}
return *configured
}
// SupavisorConfig represents the main supavisor configuration
type SupavisorConfig struct {
LogFile string
PidFile string
Socket string
SocketGroup string
LogFormat string
LogLevel string
}
// ProgramConfig represents configuration for a single program
type ProgramConfig struct {
Environment map[string]string
// HealthCheck is nil for a program that does not declare one, in which case
// nothing about the program can be said beyond whether it is running.
HealthCheck *HealthCheck
Name string
Command string
Directory string
Autorestart RestartPolicy
StdoutLogfile string
StderrLogfile string
User string
DependsOn []Dependency
Autostart bool
Priority int
StartSecs int
StopSignal syscall.Signal
StopWaitSecs int
MaxRestarts int
StdoutLogfileMaxBytes int64
StdoutLogfileBackups int
StdoutLogfileMaxAge int // days
StderrLogfileMaxBytes int64
StderrLogfileBackups int
StderrLogfileMaxAge int // days
}
// DependencyNames returns the programs this one depends on, without the
// condition each has to reach. The dependency graph is about ordering only.
func (p *ProgramConfig) DependencyNames() []string {
names := make([]string, 0, len(p.DependsOn))
for _, dep := range p.DependsOn {
names = append(names, dep.Name)
}
return names
}
// Config represents the complete configuration
type Config struct {
Programs map[string]*ProgramConfig
// SourcePath is the main config file this was read from, so that a reload
// knows where to look. It is empty for a configuration built in code.
SourcePath string
Supavisor SupavisorConfig
}
// configFile represents the YAML config file structure
type configFile struct {
Programs map[string]*programFile `yaml:"programs"`
Supavisor supavisorFile `yaml:"supavisor"`
}
type supavisorFile struct {
LogFile string `yaml:"logfile"`
PidFile string `yaml:"pidfile"`
Socket string `yaml:"socket"`
SocketGroup string `yaml:"socket_group"`
LogFormat string `yaml:"log_format"`
LogLevel string `yaml:"log_level"`
}
type programFile struct {
Environment map[string]string `yaml:"environment"`
HealthCheck *healthCheckFile `yaml:"health_check"`
Autostart *bool `yaml:"autostart"`
// Pointers so an explicit 0 is distinguishable from an absent setting:
// max_restarts: 0 means never retry, not "use the default of 3".
Priority *int `yaml:"priority"`
StartSecs *int `yaml:"startsecs"`
StopWaitSecs *int `yaml:"stopwaitsecs"`
MaxRestarts *int `yaml:"max_restarts"`
StdoutLogfileBackups *int `yaml:"stdout_logfile_backups"`
StdoutLogfileMaxAge *int `yaml:"stdout_logfile_maxage"`
StderrLogfileBackups *int `yaml:"stderr_logfile_backups"`
StderrLogfileMaxAge *int `yaml:"stderr_logfile_maxage"`
Command string `yaml:"command"`
Directory string `yaml:"directory"`
Autorestart string `yaml:"autorestart"`
StopSignal string `yaml:"stopsignal"`
StdoutLogfile string `yaml:"stdout_logfile"`
StderrLogfile string `yaml:"stderr_logfile"`
StdoutLogfileMaxBytes string `yaml:"stdout_logfile_maxbytes"`
StderrLogfileMaxBytes string `yaml:"stderr_logfile_maxbytes"`
User string `yaml:"user"`
DependsOn dependsOnFile `yaml:"depends_on"`
}
// ParseConfigFile parses a single YAML configuration file. It does not look for
// fragment files. Use ParseConfig for the full main-file + supavisor.d/ behavior.
func ParseConfigFile(path string) (*Config, error) {
cfg, err := parseConfigFileRaw(path)
if err != nil {
return nil, err
}
config := &Config{
Supavisor: SupavisorConfig{
LogFile: cfg.Supavisor.LogFile,
PidFile: defaultString(cfg.Supavisor.PidFile, "/var/run/supavisor.pid"),
Socket: defaultString(cfg.Supavisor.Socket, "/tmp/supavisor.sock"),
SocketGroup: cfg.Supavisor.SocketGroup,
LogFormat: defaultString(cfg.Supavisor.LogFormat, "text"),
LogLevel: defaultString(cfg.Supavisor.LogLevel, "info"),
},
Programs: make(map[string]*ProgramConfig),
}
config.SourcePath = path
if err := mergePrograms(config.Programs, cfg.Programs, path, map[string]string{}); err != nil {
return nil, err
}
return config, nil
}
// ParseConfig parses the main config file and merges any fragment files found
// in the sibling directory <basename-no-ext>.d/ (e.g. /etc/supavisor/supavisor.yml
// -> /etc/supavisor/supavisor.d/). Fragments are loaded in lexical order and may
// only define the programs section. Duplicate program names across files are a
// hard error.
func ParseConfig(mainPath string) (*Config, error) {
cfg, err := ParseConfigFile(mainPath)
if err != nil {
return nil, err
}
dropDir := fragmentDir(mainPath)
fragments, err := listFragmentFiles(dropDir)
if err != nil {
return nil, err
}
origins := make(map[string]string, len(cfg.Programs))
for name := range cfg.Programs {
origins[name] = mainPath
}
for _, fragPath := range fragments {
frag, err := parseConfigFileRaw(fragPath)
if err != nil {
return nil, err
}
if !isSupavisorSectionEmpty(&frag.Supavisor) {
return nil, fmt.Errorf("fragment %s: must not define a supavisor section; daemon settings belong in the main config file", fragPath)
}
if err := mergePrograms(cfg.Programs, frag.Programs, fragPath, origins); err != nil {
return nil, err
}
}
return cfg, nil
}
// fragmentDir returns the sibling drop-in directory for a given main config path.
// For /etc/supavisor/supavisor.yml it returns /etc/supavisor/supavisor.d.
func fragmentDir(mainPath string) string {
dir := filepath.Dir(mainPath)
base := filepath.Base(mainPath)
ext := filepath.Ext(base)
stem := strings.TrimSuffix(base, ext)
return filepath.Join(dir, stem+".d")
}
// listFragmentFiles returns *.yml and *.yaml files in dir sorted lexically.
// A missing directory is not an error.
func listFragmentFiles(dir string) ([]string, error) {
entries, err := os.ReadDir(dir)
if err != nil {
if os.IsNotExist(err) {
return nil, nil
}
return nil, fmt.Errorf("failed to read fragment directory %s: %w", dir, err)
}
var paths []string
for _, e := range entries {
if e.IsDir() {
continue
}
ext := strings.ToLower(filepath.Ext(e.Name()))
if ext != ".yml" && ext != ".yaml" {
continue
}
paths = append(paths, filepath.Join(dir, e.Name()))
}
sort.Strings(paths)
return paths, nil
}
func parseConfigFileRaw(path string) (*configFile, error) {
data, err := os.ReadFile(path)
if err != nil {
return nil, fmt.Errorf("failed to read config file %s: %w", path, err)
}
// Reject unknown keys rather than ignoring them: a typo in a setting that
// governs restarts or log rotation would otherwise be silently replaced by
// a default, and the program would run with behavior nobody asked for.
decoder := yaml.NewDecoder(bytes.NewReader(data))
decoder.KnownFields(true)
var cfg configFile
if err := decoder.Decode(&cfg); err != nil {
if errors.Is(err, io.EOF) {
return &cfg, nil
}
return nil, fmt.Errorf("failed to parse config file %s: %w", path, err)
}
return &cfg, nil
}
func isSupavisorSectionEmpty(s *supavisorFile) bool {
return *s == (supavisorFile{})
}
// mergePrograms converts and merges raw program entries into dst. origins tracks
// which file each program was first defined in so duplicate errors can name both
// sources. A nil origins map disables tracking (single-file callers).
func mergePrograms(dst map[string]*ProgramConfig, src map[string]*programFile, srcPath string, origins map[string]string) error {
for name, prog := range src {
if prog == nil {
continue
}
if existingPath, exists := origins[name]; exists {
return fmt.Errorf("duplicate program %s: defined in %s and %s", name, existingPath, srcPath)
}
programConfig, err := convertProgram(name, prog)
if err != nil {
return fmt.Errorf("program %s (%s): %w", name, srcPath, err)
}
dst[name] = programConfig
if origins != nil {
origins[name] = srcPath
}
}
return nil
}
func defaultString(s, def string) string {
if s == "" {
return def
}
return s
}
func convertProgram(name string, raw *programFile) (*ProgramConfig, error) {
if raw.Command == "" {
return nil, fmt.Errorf("command is required")
}
autostart := true
if raw.Autostart != nil {
autostart = *raw.Autostart
}
restartPolicy := defaultString(raw.Autorestart, "unexpected")
var autorestart RestartPolicy
switch restartPolicy {
case "always":
autorestart = RestartAlways
case "never":
autorestart = RestartNever
case "unexpected":
autorestart = RestartUnexpected
default:
return nil, fmt.Errorf("invalid autorestart policy: %s (must be always, never, or unexpected)", restartPolicy)
}
// Accepting this and running the program as the daemon's user anyway would
// be a silent security surprise: a config asking for an unprivileged user
// would run as root.
if strings.TrimSpace(raw.User) != "" {
return nil, fmt.Errorf("user is not implemented: remove it, or run supavisor as %s", raw.User)
}
stopSignal, err := parseSignal(raw.StopSignal)
if err != nil {
return nil, err
}
startSecs := intOrDefault(raw.StartSecs, defaultStartSecs)
stopWaitSecs := intOrDefault(raw.StopWaitSecs, defaultStopWaitSecs)
maxRestarts := intOrDefault(raw.MaxRestarts, defaultMaxRestarts)
priority := intOrDefault(raw.Priority, defaultPriority)
for field, value := range map[string]int{
"startsecs": startSecs, "stopwaitsecs": stopWaitSecs, "max_restarts": maxRestarts,
} {
if value < 0 {
return nil, fmt.Errorf("invalid %s: %d (must not be negative)", field, value)
}
}
env := make(map[string]string)
if len(raw.Environment) > 0 {
maps.Copy(env, raw.Environment)
}
logging, err := convertLogging(raw)
if err != nil {
return nil, err
}
healthCheck, err := convertHealthCheck(raw.HealthCheck)
if err != nil {
return nil, err
}
return &ProgramConfig{
Name: name,
Command: raw.Command,
Directory: raw.Directory,
Environment: env,
Autostart: autostart,
Autorestart: autorestart,
DependsOn: raw.DependsOn,
HealthCheck: healthCheck,
Priority: priority,
StartSecs: startSecs,
StopSignal: stopSignal,
StopWaitSecs: stopWaitSecs,
MaxRestarts: maxRestarts,
StdoutLogfile: raw.StdoutLogfile,
StderrLogfile: raw.StderrLogfile,
StdoutLogfileMaxBytes: logging.stdoutMaxBytes,
StdoutLogfileBackups: logging.stdoutBackups,
StdoutLogfileMaxAge: intOrDefault(raw.StdoutLogfileMaxAge, 0),
StderrLogfileMaxBytes: logging.stderrMaxBytes,
StderrLogfileBackups: logging.stderrBackups,
StderrLogfileMaxAge: intOrDefault(raw.StderrLogfileMaxAge, 0),
User: raw.User,
}, nil
}
// loggingDefaults holds the resolved log rotation settings for a program
type loggingDefaults struct {
stdoutMaxBytes int64
stderrMaxBytes int64
stdoutBackups int
stderrBackups int
}
func convertLogging(raw *programFile) (loggingDefaults, error) {
stdoutMaxBytes, err := parseBytes("stdout_logfile_maxbytes", raw.StdoutLogfileMaxBytes)
if err != nil {
return loggingDefaults{}, err
}
stderrMaxBytes, err := parseBytes("stderr_logfile_maxbytes", raw.StderrLogfileMaxBytes)
if err != nil {
return loggingDefaults{}, err
}
l := loggingDefaults{
stdoutMaxBytes: stdoutMaxBytes,
stderrMaxBytes: stderrMaxBytes,
stdoutBackups: intOrDefault(raw.StdoutLogfileBackups, defaultLogFileBackups),
stderrBackups: intOrDefault(raw.StderrLogfileBackups, defaultLogFileBackups),
}
return l, nil
}
// stopSignals are the signals a program may be configured to stop on
var stopSignals = map[string]syscall.Signal{
"TERM": syscall.SIGTERM,
"INT": syscall.SIGINT,
"QUIT": syscall.SIGQUIT,
"HUP": syscall.SIGHUP,
"USR1": syscall.SIGUSR1,
"USR2": syscall.SIGUSR2,
"KILL": syscall.SIGKILL,
}
// parseSignal resolves a configured stop signal name, with or without the SIG
// prefix. An empty name means SIGTERM, which is what a daemon expects to be
// asked to shut down with.
func parseSignal(name string) (syscall.Signal, error) {
if strings.TrimSpace(name) == "" {
return syscall.SIGTERM, nil
}
key := strings.TrimPrefix(strings.ToUpper(strings.TrimSpace(name)), "SIG")
sig, ok := stopSignals[key]
if !ok {
names := make([]string, 0, len(stopSignals))
for known := range stopSignals {
names = append(names, known)
}
sort.Strings(names)
return 0, fmt.Errorf("invalid stopsignal: %s (must be one of %s)", name, strings.Join(names, ", "))
}
return sig, nil
}
// parseBytes parses a byte string like "10MB", "1GB", "500KB" into bytes. An
// empty value takes the default; anything unparseable is an error rather than a
// silent fallback, so that a typo in a size limit cannot quietly become 50MB.
func parseBytes(field, value string) (int64, error) {
trimmed := strings.TrimSpace(value)
if trimmed == "" {
return defaultLogFileMaxBytes, nil
}
upper := strings.ToUpper(trimmed)
var multiplier int64 = 1
switch {
case strings.HasSuffix(upper, "KB"):
multiplier = 1024
upper = strings.TrimSuffix(upper, "KB")
case strings.HasSuffix(upper, "MB"):
multiplier = 1024 * 1024
upper = strings.TrimSuffix(upper, "MB")
case strings.HasSuffix(upper, "GB"):
multiplier = 1024 * 1024 * 1024
upper = strings.TrimSuffix(upper, "GB")
case strings.HasSuffix(upper, "B"):
upper = strings.TrimSuffix(upper, "B")
}
val, err := strconv.ParseInt(strings.TrimSpace(upper), 10, 64)
if err != nil {
return 0, fmt.Errorf("invalid %s: %s (expected a byte count, optionally suffixed with KB, MB or GB)", field, value)
}
if val < 0 {
return 0, fmt.Errorf("invalid %s: %s (must not be negative)", field, value)
}
return val * multiplier, nil
}
// Validate validates the configuration
func (c *Config) Validate() error {
// Check for circular dependencies
visited := make(map[string]bool)
recStack := make(map[string]bool)
for name := range c.Programs {
if !visited[name] {
if err := c.checkCircularDependency(name, visited, recStack); err != nil {
return err
}
}
}
// Check that all dependencies exist, and that the condition each is waited
// on for is one the dependency can actually reach
for _, name := range sortedProgramNames(c.Programs) {
for _, dep := range c.Programs[name].DependsOn {
target, exists := c.Programs[dep.Name]
if !exists {
return fmt.Errorf("program %s depends on %s which does not exist", name, dep.Name)
}
if dep.Condition == ConditionHealthy && target.HealthCheck == nil {
return fmt.Errorf("program %s waits for %s to be healthy, but %s has no health_check", name, dep.Name, dep.Name)
}
// A program that is always restarted is never left in EXITED, so
// waiting for it to complete would never resolve.
if dep.Condition == ConditionCompleted && target.Autorestart == RestartAlways {
return fmt.Errorf(
"program %s waits for %s to complete, but %s has autorestart: always and is restarted instead of completing",
name, dep.Name, dep.Name)
}
}
}
if err := c.validateLogPaths(); err != nil {
return err
}
return c.validateSocketPath()
}
// validateLogPaths rejects two programs writing to the same log file.
//
// Supavisor owns the log descriptor so that rotation works, which means two
// programs sharing a path would rotate the same files independently and
// destroy each other's output.
func (c *Config) validateLogPaths() error {
owners := make(map[string]string)
for _, name := range sortedProgramNames(c.Programs) {
prog := c.Programs[name]
for _, path := range []string{prog.StdoutLogfile, prog.StderrLogfile} {
if path == "" {
continue
}
// One program pointing both its streams at one file is fine: they
// share a single writer.
if owner, taken := owners[path]; taken && owner != name {
return fmt.Errorf("programs %s and %s both log to %s: each log file must belong to one program", owner, name, path)
}
owners[path] = name
}
}
return nil
}
// validateSocketPath rejects a socket path the kernel cannot bind.
//
// The sockaddr_un limit is around 104 bytes, and exceeding it surfaces as a
// bare "bind: invalid argument" from the listener with nothing to point at.
func (c *Config) validateSocketPath() error {
if len(c.Supavisor.Socket) >= maxSocketPathLen {
return fmt.Errorf("socket path is %d bytes, which exceeds the %d byte limit for a unix socket: %s",
len(c.Supavisor.Socket), maxSocketPathLen-1, c.Supavisor.Socket)
}
return nil
}
func sortedProgramNames(programs map[string]*ProgramConfig) []string {
names := make([]string, 0, len(programs))
for name := range programs {
names = append(names, name)
}
sort.Strings(names)
return names
}
func (c *Config) checkCircularDependency(name string, visited, recStack map[string]bool) error {
visited[name] = true
recStack[name] = true
prog, exists := c.Programs[name]
if !exists {
return nil
}
for _, dep := range prog.DependsOn {
if !visited[dep.Name] {
if err := c.checkCircularDependency(dep.Name, visited, recStack); err != nil {
return err
}
} else if recStack[dep.Name] {
return fmt.Errorf("circular dependency detected: %s -> %s", name, dep.Name)
}
}
recStack[name] = false
return nil
}
// EnsureLogDirectories creates directories for log files if they don't exist
func (c *Config) EnsureLogDirectories() error {
dirs := make(map[string]bool)
for _, prog := range c.Programs {
if prog.StdoutLogfile != "" {
dir := getDir(prog.StdoutLogfile)
if dir != "" {
dirs[dir] = true
}
}
if prog.StderrLogfile != "" {
dir := getDir(prog.StderrLogfile)
if dir != "" {
dirs[dir] = true
}
}
}
// Create supavisor log directory
if c.Supavisor.LogFile != "" {
dir := getDir(c.Supavisor.LogFile)
if dir != "" {
dirs[dir] = true
}
}
for dir := range dirs {
if err := os.MkdirAll(dir, 0o755); err != nil {
return fmt.Errorf("failed to create log directory %s: %w", dir, err)
}
}
return nil
}
func getDir(path string) string {
idx := strings.LastIndex(path, "/")
if idx == -1 {
return ""
}
return path[:idx]
}
package config
import (
"fmt"
"gopkg.in/yaml.v3"
)
// DependencyCondition is what a dependency has to reach before a program that
// depends on it may start.
type DependencyCondition string
const (
// ConditionStarted is satisfied once the dependency is RUNNING, which says
// its process is alive but nothing about whether it can serve yet.
ConditionStarted DependencyCondition = "started"
// ConditionHealthy also requires the dependency's health check to pass,
// which is what readiness means for a program that initializes after its
// process is up.
ConditionHealthy DependencyCondition = "healthy"
// ConditionCompleted is satisfied once the dependency has exited with
// status 0, which is what "the work is done" means for a migration, an init
// task or anything else that runs once rather than staying up. It stays
// satisfied while the dependency sits in EXITED, so a dependent started
// long afterwards still starts.
ConditionCompleted DependencyCondition = "completed"
)
// Dependency is one entry of a program's depends_on
type Dependency struct {
Name string
Condition DependencyCondition
}
// dependsOnFile accepts both depends_on forms: a list of program names, which
// waits for each of them to be RUNNING, and a mapping of program name to
// condition, which can additionally wait for a health check.
type dependsOnFile []Dependency
// UnmarshalYAML decodes either depends_on form
func (d *dependsOnFile) UnmarshalYAML(node *yaml.Node) error {
if node.Kind == yaml.SequenceNode {
var names []string
if err := node.Decode(&names); err != nil {
return fmt.Errorf("invalid depends_on: %w", err)
}
deps := make([]Dependency, 0, len(names))
for _, name := range names {
deps = append(deps, Dependency{Name: name, Condition: ConditionStarted})
}
*d = deps
return nil
}
if node.Kind == yaml.MappingNode {
deps, err := parseDependencyMapping(node)
if err != nil {
return err
}
*d = deps
return nil
}
return fmt.Errorf("invalid depends_on: expected a list of program names, or a mapping of program name to condition")
}
// parseDependencyMapping reads the mapping form. The nodes are walked by hand
// rather than decoded into a struct so that an unknown key is still rejected:
// a yaml.Node decodes without the strict mode the config decoder was given.
func parseDependencyMapping(node *yaml.Node) ([]Dependency, error) {
deps := make([]Dependency, 0, len(node.Content)/2)
for i := 0; i+1 < len(node.Content); i += 2 {
key, value := node.Content[i], node.Content[i+1]
dep := Dependency{Name: key.Value, Condition: ConditionStarted}
switch {
// A bare name with nothing under it keeps the default condition
case value.Tag == "!!null":
case value.Kind == yaml.MappingNode:
condition, err := parseDependencyCondition(key.Value, value)
if err != nil {
return nil, err
}
dep.Condition = condition
default:
return nil, fmt.Errorf("invalid depends_on entry %s: expected a mapping with a condition key", key.Value)
}
deps = append(deps, dep)
}
return deps, nil
}
// parseDependencyCondition reads the settings of one mapping-form entry
func parseDependencyCondition(name string, node *yaml.Node) (DependencyCondition, error) {
condition := ConditionStarted
for i := 0; i+1 < len(node.Content); i += 2 {
key, value := node.Content[i], node.Content[i+1]
if key.Value != "condition" {
return "", fmt.Errorf("invalid depends_on entry %s: unknown key %s", name, key.Value)
}
switch DependencyCondition(value.Value) {
case ConditionStarted, ConditionHealthy, ConditionCompleted:
condition = DependencyCondition(value.Value)
default:
return "", fmt.Errorf("invalid depends_on condition for %s: %s (must be started, healthy or completed)", name, value.Value)
}
}
return condition, nil
}
package config
import (
"fmt"
"net"
"net/url"
"sort"
"strings"
"time"
)
const (
// defaultProbeInterval is how often a health check runs. It is short
// because a dependent waiting on the check pays it as startup latency.
defaultProbeInterval = 2 * time.Second
// defaultProbeTimeout bounds a single attempt, so that a probe which hangs
// holds up the next one rather than the checker for good.
defaultProbeTimeout = 5 * time.Second
// defaultProbeRetries is how many consecutive failures mark a program that
// was healthy as unhealthy.
defaultProbeRetries = 3
// probeHTTP names the http probe kind, which is also the url scheme it
// accepts alongside https.
probeHTTP = "http"
)
// HealthCheck describes how to tell whether a program is ready to serve, as
// opposed to merely running. Exactly one of Exec, TCP and HTTP is set.
type HealthCheck struct {
Exec string
TCP string
HTTP string
Interval time.Duration
Timeout time.Duration
StartPeriod time.Duration
Retries int
}
type healthCheckFile struct {
Retries *int `yaml:"retries"`
Exec string `yaml:"exec"`
TCP string `yaml:"tcp"`
HTTP string `yaml:"http"`
Interval string `yaml:"interval"`
Timeout string `yaml:"timeout"`
StartPeriod string `yaml:"start_period"`
}
// convertHealthCheck resolves a raw health_check block, or returns nil for a
// program that does not have one
func convertHealthCheck(raw *healthCheckFile) (*HealthCheck, error) {
if raw == nil {
return nil, nil
}
check := &HealthCheck{
Exec: strings.TrimSpace(raw.Exec),
TCP: strings.TrimSpace(raw.TCP),
HTTP: strings.TrimSpace(raw.HTTP),
Retries: intOrDefault(raw.Retries, defaultProbeRetries),
}
if err := validateProbeTarget(check); err != nil {
return nil, err
}
var err error
if check.Interval, err = parseDuration("health_check.interval", raw.Interval, defaultProbeInterval); err != nil {
return nil, err
}
if check.Timeout, err = parseDuration("health_check.timeout", raw.Timeout, defaultProbeTimeout); err != nil {
return nil, err
}
if check.StartPeriod, err = parseDuration("health_check.start_period", raw.StartPeriod, 0); err != nil {
return nil, err
}
switch {
case check.Interval <= 0:
return nil, fmt.Errorf("invalid health_check.interval: %s (must be positive)", check.Interval)
case check.Timeout <= 0:
return nil, fmt.Errorf("invalid health_check.timeout: %s (must be positive)", check.Timeout)
case check.StartPeriod < 0:
return nil, fmt.Errorf("invalid health_check.start_period: %s (must not be negative)", check.StartPeriod)
case check.Retries < 1:
return nil, fmt.Errorf("invalid health_check.retries: %d (must be at least 1)", check.Retries)
}
return check, nil
}
// validateProbeTarget checks that exactly one kind of probe is configured and
// that it is usable, so that a typo fails at startup rather than on every
// attempt for the lifetime of the program.
func validateProbeTarget(check *HealthCheck) error {
kinds := map[string]string{"exec": check.Exec, "tcp": check.TCP, probeHTTP: check.HTTP}
configured := make([]string, 0, len(kinds))
for kind, value := range kinds {
if value != "" {
configured = append(configured, kind)
}
}
sort.Strings(configured)
switch {
case len(configured) == 0:
return fmt.Errorf("health_check needs one of exec, tcp or http")
case len(configured) > 1:
return fmt.Errorf("health_check sets %s: only one of exec, tcp or http may be used", strings.Join(configured, " and "))
}
if check.TCP != "" {
if _, _, err := net.SplitHostPort(check.TCP); err != nil {
return fmt.Errorf("invalid health_check.tcp: %s (expected host:port)", check.TCP)
}
}
if check.HTTP != "" {
parsed, err := url.Parse(check.HTTP)
if err != nil {
return fmt.Errorf("invalid health_check.http: %s (%w)", check.HTTP, err)
}
if (parsed.Scheme != probeHTTP && parsed.Scheme != "https") || parsed.Host == "" {
return fmt.Errorf("invalid health_check.http: %s (expected an http:// or https:// url)", check.HTTP)
}
}
return nil
}
// parseDuration parses a duration such as 2s or 500ms. An empty value takes the
// default; anything unparseable is an error rather than a silent fallback, for
// the same reason a mistyped log size is.
func parseDuration(field, value string, def time.Duration) (time.Duration, error) {
trimmed := strings.TrimSpace(value)
if trimmed == "" {
return def, nil
}
parsed, err := time.ParseDuration(trimmed)
if err != nil {
return 0, fmt.Errorf("invalid %s: %s (expected a duration such as 2s or 500ms)", field, value)
}
return parsed, nil
}
package dependency
import (
"fmt"
"slices"
"sort"
)
// Graph represents a directed graph of process dependencies
type Graph struct {
nodes map[string]*Node
}
// Node represents a node in the dependency graph
type Node struct {
Name string
Dependencies []string
Dependents []string
}
// NewGraph creates a new dependency graph
func NewGraph() *Graph {
return &Graph{
nodes: make(map[string]*Node),
}
}
// AddNode adds a node to the graph
func (g *Graph) AddNode(name string, dependencies []string) {
// If node already exists, update its dependencies
if existingNode, exists := g.nodes[name]; exists {
// Remove old dependent relationships
for _, oldDep := range existingNode.Dependencies {
if depNode, exists := g.nodes[oldDep]; exists {
// Remove name from oldDep's dependents
newDependents := make([]string, 0, len(depNode.Dependents))
for _, d := range depNode.Dependents {
if d != name {
newDependents = append(newDependents, d)
}
}
depNode.Dependents = newDependents
}
}
existingNode.Dependencies = dependencies
} else {
node := &Node{
Name: name,
Dependencies: dependencies,
Dependents: make([]string, 0),
}
g.nodes[name] = node
}
// Update dependents for each dependency
node := g.nodes[name]
for _, dep := range dependencies {
if depNode, exists := g.nodes[dep]; exists {
// Check if name is already in dependents
found := slices.Contains(depNode.Dependents, name)
if !found {
depNode.Dependents = append(depNode.Dependents, name)
}
}
}
// Also check if any existing nodes depend on this newly added node
for otherName, otherNode := range g.nodes {
if otherName == name {
continue
}
for _, dep := range otherNode.Dependencies {
if dep == name {
// otherNode depends on name, so name should have otherName as dependent
found := slices.Contains(node.Dependents, otherName)
if !found {
node.Dependents = append(node.Dependents, otherName)
}
}
}
}
}
// TopologicalSort returns a topological ordering of nodes
// Returns an error if a circular dependency is detected
func (g *Graph) TopologicalSort() ([]string, error) {
// Calculate in-degrees
inDegree := make(map[string]int)
for name, node := range g.nodes {
inDegree[name] = 0
for _, dep := range node.Dependencies {
if _, exists := g.nodes[dep]; exists {
inDegree[name]++
}
}
}
// Find all nodes with in-degree 0
queue := make([]string, 0, len(g.nodes))
for name, degree := range inDegree {
if degree == 0 {
queue = append(queue, name)
}
}
result := []string{}
processed := 0
for len(queue) > 0 {
current := queue[0]
queue = queue[1:]
result = append(result, current)
processed++
// Reduce in-degree for all dependents
node := g.nodes[current]
for _, dependent := range node.Dependents {
inDegree[dependent]--
if inDegree[dependent] == 0 {
queue = append(queue, dependent)
}
}
}
// If we didn't process all nodes, there's a cycle
if processed != len(g.nodes) {
return nil, fmt.Errorf("circular dependency detected")
}
return result, nil
}
// Tiers groups nodes so that everything in a tier depends only on nodes in
// earlier tiers. Startup works forwards through the tiers, shutdown backwards,
// and everything within one tier is independent of its peers.
// Returns an error if a circular dependency is detected.
func (g *Graph) Tiers() ([][]string, error) {
inDegree := make(map[string]int, len(g.nodes))
for name, node := range g.nodes {
inDegree[name] = 0
for _, dep := range node.Dependencies {
if _, exists := g.nodes[dep]; exists {
inDegree[name]++
}
}
}
current := make([]string, 0, len(g.nodes))
for name, degree := range inDegree {
if degree == 0 {
current = append(current, name)
}
}
tiers := [][]string{}
placed := 0
for len(current) > 0 {
sort.Strings(current)
tiers = append(tiers, current)
placed += len(current)
next := make([]string, 0, len(g.nodes))
for _, name := range current {
for _, dependent := range g.nodes[name].Dependents {
inDegree[dependent]--
if inDegree[dependent] == 0 {
next = append(next, dependent)
}
}
}
current = next
}
if placed != len(g.nodes) {
return nil, fmt.Errorf("circular dependency detected")
}
return tiers, nil
}
// GetDependencies returns all processes that the given process depends on
func (g *Graph) GetDependencies(name string) []string {
node, exists := g.nodes[name]
if !exists {
return []string{}
}
return node.Dependencies
}
// Package logrotate writes captured process output to a log file, rotating it
// by size and pruning old backups.
package logrotate
import (
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"time"
)
// Writer is an io.WriteCloser that appends to a log file and rotates it once it
// would grow past maxBytes.
//
// The writer owns the file descriptor. Rotation closes, renames and reopens,
// which only works because the process producing the output writes to a pipe
// rather than to this file: a child holding its own descriptor would keep
// writing to the renamed inode and the new file would stay empty forever.
type Writer struct {
file *os.File
path string
maxBytes int64
size int64
backups int
maxAge int
mu sync.Mutex
}
// NewWriter opens path for appending, creating its directory if needed
func NewWriter(path string, maxBytes int64, backups, maxAge int) (*Writer, error) {
if dir := filepath.Dir(path); dir != "" && dir != "." {
if err := os.MkdirAll(dir, 0o755); err != nil {
return nil, fmt.Errorf("failed to create log directory %s: %w", dir, err)
}
}
file, err := openAppend(path)
if err != nil {
return nil, err
}
info, err := file.Stat()
if err != nil {
_ = file.Close()
return nil, fmt.Errorf("failed to stat log file %s: %w", path, err)
}
w := &Writer{
file: file,
path: path,
maxBytes: maxBytes,
size: info.Size(),
backups: backups,
maxAge: maxAge,
}
w.prune()
return w, nil
}
// Path returns the log file being written
func (w *Writer) Path() string {
return w.path
}
// Write appends p, rotating first if it would take the file past maxBytes
func (w *Writer) Write(p []byte) (int, error) {
w.mu.Lock()
defer w.mu.Unlock()
if w.file == nil {
return 0, os.ErrClosed
}
// Only rotate when there is content to preserve, so that a single write
// larger than maxBytes still lands somewhere instead of rotating forever.
if w.maxBytes > 0 && w.size > 0 && w.size+int64(len(p)) > w.maxBytes {
// A rotation that fails must not stop the process from logging, so the
// error is only fatal when it left us without an open file.
if err := w.rotate(); err != nil && w.file == nil {
return 0, err
}
}
n, err := w.file.Write(p)
w.size += int64(n)
return n, err
}
// Close releases the log file
func (w *Writer) Close() error {
w.mu.Lock()
defer w.mu.Unlock()
if w.file == nil {
return nil
}
err := w.file.Close()
w.file = nil
return err
}
// rotate shifts the backup chain and reopens an empty log. Must hold mu.
func (w *Writer) rotate() error {
if err := w.file.Close(); err != nil {
return fmt.Errorf("failed to close log before rotation: %w", err)
}
w.file = nil
shiftErr := w.shiftBackups()
// Reopen whatever happened above: losing the descriptor would silently end
// this process's logging, which is worse than a failed rotation.
file, err := openAppend(w.path)
if err != nil {
return err
}
w.file = file
w.size = 0
if info, statErr := file.Stat(); statErr == nil {
w.size = info.Size()
}
w.prune()
return shiftErr
}
// shiftBackups renames log.N to log.N+1 and the current log to log.1
func (w *Writer) shiftBackups() error {
if w.backups <= 0 {
if err := os.Remove(w.path); err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to discard current log: %w", err)
}
return nil
}
for i := w.backups - 1; i >= 1; i-- {
oldPath := backupPath(w.path, i)
if _, err := os.Stat(oldPath); err != nil {
continue
}
if err := os.Rename(oldPath, backupPath(w.path, i+1)); err != nil {
return fmt.Errorf("failed to rotate backup %d: %w", i, err)
}
}
if err := os.Rename(w.path, backupPath(w.path, 1)); err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to rotate current log: %w", err)
}
return nil
}
// prune removes numbered backups beyond the configured count and, when maxAge
// is set, those older than the limit. Must hold mu.
func (w *Writer) prune() {
dir := filepath.Dir(w.path)
base := filepath.Base(w.path)
entries, err := os.ReadDir(dir)
if err != nil {
return
}
var cutoff time.Time
if w.maxAge > 0 {
cutoff = time.Now().AddDate(0, 0, -w.maxAge)
}
for _, entry := range entries {
if entry.IsDir() || !strings.HasPrefix(entry.Name(), base+".") {
continue
}
num, ok := backupNumber(strings.TrimPrefix(entry.Name(), base+"."))
if !ok {
continue
}
expired := false
if w.maxAge > 0 {
info, infoErr := entry.Info()
expired = infoErr == nil && info.ModTime().Before(cutoff)
}
if num > w.backups || expired {
_ = os.Remove(filepath.Join(dir, entry.Name()))
}
}
}
func openAppend(path string) (*os.File, error) {
file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
if err != nil {
return nil, fmt.Errorf("failed to open log file %s: %w", path, err)
}
return file, nil
}
func backupPath(path string, num int) string {
return fmt.Sprintf("%s.%d", path, num)
}
// backupNumber parses the numeric suffix of a rotated log file
func backupNumber(s string) (int, bool) {
if s == "" {
return 0, false
}
num := 0
for _, r := range s {
if r < '0' || r > '9' {
return 0, false
}
num = num*10 + int(r-'0')
}
if num == 0 {
return 0, false
}
return num, true
}
package process
import (
"context"
"fmt"
"io"
"log/slog"
"net"
"net/http"
"os"
"os/exec"
"strings"
"syscall"
"time"
"github.com/ademidoff/supavisor/internal/config"
)
const (
// maxProbeOutputBytes bounds how much of a failing exec probe's output is
// carried into the log line.
maxProbeOutputBytes = 200
// probeWaitDelay bounds how long a failing probe's output is waited for
// after it exits, so that a probe leaking a background child, which holds
// the pipe open behind it, cannot hold the checker.
probeWaitDelay = time.Second
)
// Health is what a program's health check says about it. It is separate from
// the process state, because a program can be RUNNING and not yet able to
// serve: a database in its bootstrap phase is alive but refuses connections.
type Health string
const (
// HealthNone is reported when no health check is configured, and when the
// program is not running, where readiness would mean nothing.
HealthNone Health = "NONE"
// HealthStarting is a configured check that has not passed yet
HealthStarting Health = "STARTING"
// HealthHealthy is a check that passed on its last attempt
HealthHealthy Health = "HEALTHY"
// HealthUnhealthy is a check that has failed retries times in a row
HealthUnhealthy Health = "UNHEALTHY"
)
// probeFunc runs one health check attempt. A nil error means the program is
// ready to serve.
type probeFunc func(ctx context.Context) error
// SetHealthChangeCallback sets a callback for health changes
func (p *Process) SetHealthChangeCallback(fn func(name string, prevHealth, health Health)) {
p.onHealthChange = fn
}
// GetHealth returns what the health check last reported. It is HealthNone for a
// program without a health check, and for one that is not running.
func (p *Process) GetHealth() Health {
p.mu.RLock()
defer p.mu.RUnlock()
if p.health == "" {
return HealthNone
}
return p.health
}
// setHealth records a health result and calls the callback
func (p *Process) setHealth(health Health) {
p.mu.Lock()
prevHealth := p.health
if prevHealth == "" {
prevHealth = HealthNone
}
p.health = health
p.mu.Unlock()
if prevHealth == health {
return
}
// The server logs the transition too, so this is the detail behind it
p.logger.Debug("Health changed", "prev_health", prevHealth, "health", health)
if p.onHealthChange != nil {
p.onHealthChange(p.config.Name, prevHealth, health)
}
}
// startHealthChecks begins probing this run. It is a no-op for a program
// without a health check, which stays at HealthNone.
func (p *Process) startHealthChecks(ctx context.Context) {
check := p.config.HealthCheck
if check == nil {
return
}
probe, err := newProbe(p.config, p.logger)
if err != nil {
// Configuration validation rejects this, so reaching it means the
// program was built in code rather than parsed from a file.
p.logger.Error("Health check is unusable, not probing", "error", err)
return
}
healthCtx, cancel := context.WithCancel(ctx)
done := make(chan struct{})
p.mu.Lock()
p.healthCancel = cancel
p.healthDone = done
p.mu.Unlock()
// Set before returning rather than from the goroutine, so that a dependent
// looking immediately after the start cannot see the previous run's result.
p.setHealth(HealthStarting)
go func() {
defer close(done)
p.probeLoop(healthCtx, check, probe)
}()
}
// stopHealthChecks ends probing and clears the health, which means nothing for
// a program that is not running. It is safe to call more than once.
func (p *Process) stopHealthChecks() {
p.mu.Lock()
cancel, done := p.healthCancel, p.healthDone
p.healthCancel, p.healthDone = nil, nil
p.mu.Unlock()
if cancel == nil {
return
}
cancel()
<-done
p.setHealth(HealthNone)
}
// probeLoop checks the program until the run ends.
//
// A failing probe never stops or restarts the program: supavisor reports what
// it observes, and the restart policy stays tied to the process actually
// exiting. What health does decide is whether dependents waiting on it may
// start.
func (p *Process) probeLoop(ctx context.Context, check *config.HealthCheck, probe probeFunc) {
ticker := time.NewTicker(check.Interval)
defer ticker.Stop()
// Probing straight away rather than after one interval keeps a program that
// is ready immediately from paying the interval as startup latency.
graceEnd := time.Now().Add(check.StartPeriod)
everHealthy := false
failures := 0
for {
attemptCtx, cancelAttempt := context.WithTimeout(ctx, check.Timeout)
err := probe(attemptCtx)
cancelAttempt()
switch {
case ctx.Err() != nil:
return
case err == nil:
failures = 0
everHealthy = true
p.setHealth(HealthHealthy)
default:
failures++
p.logger.Debug("Health check failed", "error", err, "consecutive_failures", failures)
// Inside start_period a program that has never been healthy is
// still starting up, which is the whole point of that window.
if failures >= check.Retries && (everHealthy || !time.Now().Before(graceEnd)) {
p.setHealth(HealthUnhealthy)
}
}
select {
case <-ticker.C:
case <-ctx.Done():
return
}
}
}
// newProbe builds the probe for a program's health check
func newProbe(cfg *config.ProgramConfig, logger *slog.Logger) (probeFunc, error) {
check := cfg.HealthCheck
switch {
case check.Exec != "":
return execProbe(cfg, logger), nil
case check.TCP != "":
return tcpProbe(check.TCP), nil
case check.HTTP != "":
return httpProbe(check.HTTP), nil
}
return nil, fmt.Errorf("health check has no exec command, tcp address or http url")
}
// execProbe runs a command and treats a zero exit status as ready. It runs
// where the program runs and with the program's environment, so that a check
// like pg_isready sees the same settings the program was given.
func execProbe(cfg *config.ProgramConfig, logger *slog.Logger) probeFunc {
return func(ctx context.Context) error {
parts := parseCommand(cfg.HealthCheck.Exec)
if len(parts) == 0 {
return fmt.Errorf("invalid health check command: %s", cfg.HealthCheck.Exec)
}
// Neither CommandContext nor CombinedOutput, because both of them wait
// for the child and the reaper is the only thing allowed to do that. A
// second waiter loses statuses, and a lost status here reads as a
// failed probe: a healthy program intermittently marked UNHEALTHY, and
// its dependents held back with it.
//nolint:gosec,noctx // see above; the context is honored below instead
cmd := exec.Command(parts[0], parts[1:]...)
cmd.Dir = cfg.Directory
cmd.Env = processEnv(cfg)
// Its own group, so a timed-out probe can be killed along with anything
// it spawned rather than just the command itself.
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
readEnd, writeEnd, err := os.Pipe()
if err != nil {
return fmt.Errorf("failed to create probe pipe: %w", err)
}
// Closing the read end unblocks the drain goroutine below, whatever
// happens to the probe itself.
defer func() { _ = readEnd.Close() }()
// An *os.File is handed to the child as-is. Any other writer makes
// os/exec start a copier that only Wait() can finish.
cmd.Stdout = writeEnd
cmd.Stderr = writeEnd
reaper := reaperFor(logger)
var pid int
exited, err := reaper.spawn(func() (int, error) {
startErr := cmd.Start()
if startErr != nil {
return 0, startErr
}
pid = cmd.Process.Pid
return pid, nil
})
// The child holds its own copy now. Ours would keep the read below from
// ever reaching EOF.
_ = writeEnd.Close()
if err != nil {
return err
}
// Nothing waits on this os.Process, so the handle Start allocated has to
// be given back by hand. A probe runs every interval for the life of the
// program, so leaking one pidfd per attempt exhausts the daemon's
// descriptors far quicker than the main process path would.
defer func() {
releaseErr := cmd.Process.Release()
if releaseErr != nil {
logger.Debug("Failed to release the probe handle", "error", releaseErr)
}
}()
output := make(chan []byte, 1)
go func() {
// A read error means the probe is gone or the pipe was closed under
// us; either way whatever arrived is all the output there is.
data, _ := io.ReadAll(readEnd) //nolint:errcheck
output <- data
}()
select {
case status := <-exited:
if exitCodeOfStatus(status) == 0 {
return nil
}
return probeFailure(ctx, status, output)
case <-ctx.Done():
// Only while the reaper still holds it: once a PID has been
// collected the kernel may have handed it to something else, and
// kill(-pid) would land on a stranger. This is the same hazard that
// ruled out exec.CommandContext in buildCommand.
signalErr := reaper.signalGroupIfUnreaped(pid, syscall.SIGKILL)
if signalErr != nil {
logger.Debug("Failed to kill a timed-out probe", "error", signalErr)
}
return fmt.Errorf("health check timed out: %w", ctx.Err())
}
}
}
// probeFailure describes a probe that exited non-zero, quoting its output when
// it produced any.
//
// The wait is bounded twice over: a probe that leaked a background child leaves
// the pipe open behind it, and the attempt's own deadline may pass while we are
// waiting. Neither may hold the checker, which stopHealthChecks blocks on when
// a program is stopping.
func probeFailure(ctx context.Context, status syscall.WaitStatus, output <-chan []byte) error {
err := exitErrorOf(status)
timer := time.NewTimer(probeWaitDelay)
defer timer.Stop()
select {
case data := <-output:
if line := firstLine(data); line != "" {
return fmt.Errorf("%w: %s", err, line)
}
case <-timer.C:
case <-ctx.Done():
}
return err
}
// tcpProbe connects to an address and treats a completed connection as ready
func tcpProbe(address string) probeFunc {
return func(ctx context.Context) error {
var dialer net.Dialer
conn, err := dialer.DialContext(ctx, "tcp", address)
if err != nil {
return err
}
return conn.Close()
}
}
// httpProbe issues a GET and treats any non-error status as ready
func httpProbe(target string) probeFunc {
return func(ctx context.Context) error {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, http.NoBody)
if err != nil {
return err
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode >= http.StatusBadRequest {
return fmt.Errorf("health check returned status %d", resp.StatusCode)
}
return nil
}
}
// firstLine reduces probe output to something that fits in a log line
func firstLine(out []byte) string {
text := strings.TrimSpace(string(out))
if text == "" {
return ""
}
if idx := strings.IndexByte(text, '\n'); idx != -1 {
text = text[:idx]
}
if len(text) > maxProbeOutputBytes {
text = text[:maxProbeOutputBytes] + "..."
}
return text
}
package process
import (
"bufio"
"context"
"errors"
"fmt"
"io"
"log/slog"
"os"
"os/exec"
"sync"
"syscall"
"time"
"github.com/ademidoff/supavisor/internal/config"
"github.com/ademidoff/supavisor/internal/logrotate"
)
const (
maxBackoffSeconds = 30
restartWaitInterval = 100 * time.Millisecond
// logDrainTimeout bounds how long we wait for a pipe to reach EOF after the
// process exits. A grandchild that inherited the descriptor keeps it open,
// so the wait cannot be unbounded.
logDrainTimeout = 2 * time.Second
// maxLogLineBytes bounds how much of a newline-free run of output is held in
// memory before it is written out in pieces.
maxLogLineBytes = 64 * 1024
// maxBackoffShift caps the exponent used for restart backoff. Without it
// 1<<(restartCount-1) overflows int once max_restarts grows past ~63 and
// yields a zero backoff, turning the restart policy into a tight loop.
maxBackoffShift = 5
// reapGrace bounds the wait for a monitor to report an exit after SIGKILL.
// A process in an uninterruptible wait does not die on SIGKILL, and one
// program in that state must not be able to hold up the daemon's shutdown.
reapGrace = 5 * time.Second
// healthyUptime is how long a process must stay up for the run to count as
// successful. Reaching it resets the consecutive-restart counter so that
// max_restarts bounds crash loops rather than restarts over the whole
// lifetime of the daemon.
healthyUptime = 60 * time.Second
)
// Process represents a managed process
type Process struct {
config *config.ProgramConfig
logger *slog.Logger
// onStateChange is set once before the process is started
onStateChange func(name string, prevState, newState State)
// onHealthChange is set once before the process is started
onHealthChange func(name string, prevHealth, health Health)
cmd *exec.Cmd
cancel context.CancelFunc
healthCancel context.CancelFunc
monitorDone chan struct{}
stopRequested chan struct{}
healthDone chan struct{}
lastError error
startTime time.Time
stopTime time.Time
state State
health Health
streams []*logStream
// mu guards all mutable run state, from cmd down to stoppedExternally.
// The monitor goroutine writes it while the IPC path reads it.
mu sync.RWMutex
pid int
exitCode int
restartCount int
completed bool
stoppedExternally bool
}
// logStream captures one child output stream. The child writes into a pipe and
// a drain goroutine copies it into the rotating log file, so that supavisor
// rather than the child owns the log descriptor.
type logStream struct {
sink *logrotate.Writer
readEnd *os.File
childEnd *os.File
done chan struct{}
}
// NewProcess creates a new process instance
func NewProcess(cfg *config.ProgramConfig, logger *slog.Logger) *Process {
return &Process{
config: cfg,
logger: logger.With("component", "process", "process", cfg.Name),
state: StateStopped,
health: HealthNone,
}
}
// SetStateChangeCallback sets a callback for state changes
func (p *Process) SetStateChangeCallback(fn func(name string, prevState, newState State)) {
p.onStateChange = fn
}
// GetState returns the current state
func (p *Process) GetState() State {
p.mu.RLock()
defer p.mu.RUnlock()
return p.state
}
// setState sets the state and calls the callback
func (p *Process) setState(newState State) {
p.mu.Lock()
prevState := p.state
p.state = newState
p.mu.Unlock()
if p.onStateChange != nil && prevState != newState {
p.onStateChange(p.config.Name, prevState, newState)
}
}
// compareAndSetState transitions from one state to another only if the process
// is still in the expected state, and reports whether it did.
func (p *Process) compareAndSetState(from, to State) bool {
p.mu.Lock()
if p.state != from {
p.mu.Unlock()
return false
}
p.state = to
p.mu.Unlock()
if p.onStateChange != nil && from != to {
p.onStateChange(p.config.Name, from, to)
}
return true
}
// GetPID returns the process ID
func (p *Process) GetPID() int {
p.mu.RLock()
defer p.mu.RUnlock()
return p.pid
}
// GetExitCode returns the exit code
func (p *Process) GetExitCode() int {
p.mu.RLock()
defer p.mu.RUnlock()
return p.exitCode
}
// GetStartTime returns the start time
func (p *Process) GetStartTime() time.Time {
p.mu.RLock()
defer p.mu.RUnlock()
return p.startTime
}
// GetRestartCount returns the number of restarts
func (p *Process) GetRestartCount() int {
p.mu.RLock()
defer p.mu.RUnlock()
return p.restartCount
}
// HasCompleted reports whether the program has finished its work successfully:
// a run that ended on its own with status 0.
//
// It stays true while the program sits in EXITED, which is what lets a one-off
// be depended on. Starting the program again clears it, so the answer is always
// about the most recent run rather than about any run there has ever been.
func (p *Process) HasCompleted() bool {
p.mu.RLock()
defer p.mu.RUnlock()
return p.completed
}
// Start starts the process
func (p *Process) Start() error {
p.mu.Lock()
if p.state == StateRunning || p.state == StateStarting {
p.mu.Unlock()
return fmt.Errorf("process %s is already running or starting", p.config.Name)
}
// Running again puts whatever the last run achieved back in question, so
// anything waiting for this program to complete waits for this run instead.
p.completed = false
// Every run gets a fresh context, canceled by Stop(). It drives this run's
// health checks and restart backoff, so reusing a canceled one would leave
// a restarted process with neither.
if p.cancel != nil {
p.cancel()
}
ctx, cancel := context.WithCancel(context.Background())
p.cancel = cancel
p.mu.Unlock()
p.logger.Info("Setting state to STARTING")
p.setState(StateStarting)
p.logger.Info("Setting up log capture")
stdout, stderr, err := p.startLogging()
if err != nil {
p.stopLogging()
cancel()
p.setState(StateFatal)
return fmt.Errorf("failed to set up log capture: %w", err)
}
cmd, err := p.buildCommand(stdout, stderr)
if err != nil {
p.stopLogging()
cancel()
p.setState(StateFatal)
return err
}
p.logger.Info("Executing command", "command", cmd.String())
exited, err := p.spawn(cmd)
if err != nil {
p.logger.Error("Failed to start", "error", err)
p.stopLogging()
cancel()
p.setState(StateFatal)
return fmt.Errorf("failed to start process: %w", err)
}
// The child has its own copies now. Until ours are closed the pipes never
// reach EOF and the drain goroutines would never finish.
p.closeChildEnds()
monitorDone := make(chan struct{})
stopRequested := make(chan struct{})
// Closed as soon as the monitor sees the exit, so the start check can tell
// a process that fell over from one that is still up without probing a PID
// that may already have been reused.
runExited := make(chan struct{})
pid := p.recordRun(cmd, monitorDone, stopRequested)
p.logger.Info("Started process", "pid", pid)
// Before the monitor: it tears the checker down when the process exits, and
// a process that exits straight away would otherwise leave one running.
p.startHealthChecks(ctx)
go p.monitor(ctx, cmd, exited, runExited, monitorDone, stopRequested)
go p.waitForStartSuccess(ctx, runExited)
return nil
}
// recordRun publishes the state of a run that has just started, and returns its
// PID. Everything here is read by the IPC path while the monitor writes it.
func (p *Process) recordRun(cmd *exec.Cmd, monitorDone, stopRequested chan struct{}) int {
p.mu.Lock()
defer p.mu.Unlock()
p.cmd = cmd
p.pid = cmd.Process.Pid
p.startTime = time.Now()
p.lastError = nil
p.monitorDone = monitorDone
p.stopRequested = stopRequested
p.stoppedExternally = false
return p.pid
}
// spawn starts the command through the reaper, which registers the PID before
// it can exit. Starting it any other way would let a program that fails
// immediately have its status collected with nobody yet listening for it.
func (p *Process) spawn(cmd *exec.Cmd) (<-chan syscall.WaitStatus, error) {
return reaperFor(p.logger).spawn(func() (int, error) {
err := cmd.Start()
if err != nil {
return 0, err
}
return cmd.Process.Pid, nil
})
}
// buildCommand assembles the exec.Cmd for a single run. A nil stdout or stderr
// descriptor leaves the stream connected to /dev/null.
func (p *Process) buildCommand(stdout, stderr *os.File) (*exec.Cmd, error) {
p.logger.Debug("Parsing command", "command", p.config.Command)
parts := parseCommand(p.config.Command)
if len(parts) == 0 {
return nil, fmt.Errorf("invalid command: %s", p.config.Command)
}
p.logger.Debug("Creating command", "command_parts", parts)
// Deliberately not CommandContext: its context watcher is wired to Wait,
// which the reaper now owns, so it would never be released. Its Cancel hook
// would also fire against a PID that has already been reaped and may have
// been recycled by then, killing an unrelated process group. Stop signals
// the group explicitly, so nothing is lost.
//nolint:gosec,noctx // noctx wants CommandContext; see the comment above for why it cannot be used here
cmd := exec.Command(parts[0], parts[1:]...)
// Give the child its own process group, which everything it spawns
// inherits. Signaling the group is the only way to reach the workload
// behind a wrapper script: killing just the direct child leaves its
// children running and reparented to init.
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
if p.config.Directory != "" {
p.logger.Info("Setting working directory", "directory", p.config.Directory)
cmd.Dir = p.config.Directory
}
if len(p.config.Environment) > 0 {
p.logger.Info("Setting environment variables", "count", len(p.config.Environment))
}
cmd.Env = processEnv(p.config)
// Leave Stdout/Stderr nil when there is no capture pipe: assigning a nil
// *os.File would hand the child a closed descriptor instead of /dev/null.
if stdout != nil {
cmd.Stdout = stdout
}
if stderr != nil {
cmd.Stderr = stderr
}
return cmd, nil
}
// processEnv returns the environment a program runs with, which its health
// check probe also inherits
func processEnv(cfg *config.ProgramConfig) []string {
env := os.Environ()
for k, v := range cfg.Environment {
env = append(env, fmt.Sprintf("%s=%s", k, v))
}
return env
}
// waitForStartSuccess promotes the process to RUNNING once it has stayed alive
// for startsecs.
func (p *Process) waitForStartSuccess(ctx context.Context, runExited <-chan struct{}) {
p.logger.Info("Waiting before checking start success", "seconds", p.config.StartSecs)
select {
case <-time.After(time.Duration(p.config.StartSecs) * time.Second):
case <-ctx.Done():
return
case <-runExited:
// The monitor owns the state from here: it has the exit code and the
// restart policy, and will settle on BACKOFF, EXITED or FATAL
// accordingly. Setting a state here as well would publish a start
// failure for a program that exited cleanly, and would race the
// monitor for a transition it is about to make correctly.
return
}
if p.compareAndSetState(StateStarting, StateRunning) {
p.logger.Info("Start successful, setting state to RUNNING")
}
}
// Stop stops the process
func (p *Process) Stop() error {
// Marking the stop before anything else tells the monitor goroutine not to
// act on a restart it may already have queued.
p.mu.Lock()
p.stoppedExternally = true
state := p.state
cmd := p.cmd
cancel := p.cancel
monitorDone := p.monitorDone
pid := p.pid
// Taking the channel makes sure it is closed exactly once, however many
// times Stop is called.
stopRequested := p.stopRequested
p.stopRequested = nil
p.mu.Unlock()
// Wake anything waiting out a restart backoff before doing anything else,
// so a stop is not held up by a delay that is about to be abandoned.
if stopRequested != nil {
close(stopRequested)
}
// A probe must not outlive the process it is checking, and readiness means
// nothing for a program that is going away.
p.stopHealthChecks()
// Nothing is running in these states, but the monitor may be sitting in a
// restart backoff. Canceling the context is what actually stops it.
if state.IsStopped() || state == StateBackoff {
p.logger.Info("Process is not running, canceling any pending restart")
if cancel != nil {
cancel()
}
p.stopLogging()
p.setState(StateStopped)
return nil
}
p.logger.Info("Stopping process", "pid", pid)
p.setState(StateStopping)
if cmd != nil && cmd.Process != nil {
// Check the group is still alive before signaling: the process may have
// exited already, for instance on the parent's own SIGTERM.
if err := SignalGroup(pid, syscall.Signal(0)); err == nil {
p.logger.Info("Signaling the process group for graceful shutdown", "signal", p.config.StopSignal)
if err := SignalGroup(pid, p.config.StopSignal); err != nil {
p.logger.Warn("Failed to send stop signal", "signal", p.config.StopSignal, "error", err)
}
}
if monitorDone != nil {
select {
case <-monitorDone:
p.logger.Info("Process exited gracefully")
case <-time.After(time.Duration(p.config.StopWaitSecs) * time.Second):
p.logger.Info("Graceful shutdown timeout, sending SIGKILL to the process group")
if err := SignalGroup(pid, syscall.SIGKILL); err != nil {
p.logger.Warn("Failed to send SIGKILL", "error", err)
}
// Bounded, because SIGKILL is not always the end of it: a
// process wedged in an uninterruptible wait does not die on it,
// and waiting forever would take the whole shutdown down with
// this one program. Giving up here leaves the run unfinished,
// which the reconciler will see, rather than hanging the daemon.
if awaitClose(monitorDone, reapGrace) {
p.logger.Info("Force killed")
} else {
p.logger.Error("Process did not report its exit after SIGKILL, giving up on it",
"pid", pid, "waited", reapGrace)
}
}
}
}
if cancel != nil {
cancel()
}
p.logger.Info("Closing process log files")
p.stopLogging()
p.logger.Info("Process stopped successfully")
return nil
}
// SignalGroup sends sig to the process group led by pid. Children inherit their
// parent's group, so this reaches the whole tree rather than just the process
// supavisor started.
func SignalGroup(pid int, sig syscall.Signal) error {
if pid <= 0 {
return fmt.Errorf("invalid pid %d", pid)
}
return syscall.Kill(-pid, sig)
}
// killLingeringGroup kills anything left in the process group after the process
// itself has exited
func (p *Process) killLingeringGroup(pid int) {
// The leader has been reaped by the time this runs, so its PID is free and
// the group id is only ours for as long as nothing else claims that number.
// A live process holding it means it has been handed out again, and
// signaling the group could then reach a stranger's tree rather than our
// leftovers. Refusing to clean up in that case leaks whatever is left in
// the old group, which is the lesser of the two: it is bounded by the
// program's own children, where the alternative is unbounded.
if err := syscall.Kill(pid, syscall.Signal(0)); err == nil {
p.logger.Warn("Not clearing the process group: its id belongs to a live process now", "pgid", pid)
return
}
if err := SignalGroup(pid, syscall.Signal(0)); err != nil {
return
}
p.logger.Info("Process group still has members after exit, killing them", "pgid", pid)
if err := SignalGroup(pid, syscall.SIGKILL); err != nil {
p.logger.Warn("Failed to kill lingering process group", "pgid", pid, "error", err)
}
}
// awaitClose waits for done to be closed, and reports whether it was before the
// timeout elapsed.
func awaitClose(done <-chan struct{}, timeout time.Duration) bool {
timer := time.NewTimer(timeout)
defer timer.Stop()
select {
case <-done:
return true
case <-timer.C:
return false
}
}
// Restart restarts the process
func (p *Process) Restart() error {
p.logger.Info("Restarting process")
if err := p.Stop(); err != nil {
p.logger.Error("Error during stop phase of restart", "error", err)
return err
}
p.logger.Debug("Waiting 100ms before restart")
time.Sleep(restartWaitInterval)
return p.Start()
}
// monitor waits for the process to exit and applies the restart policy
func (p *Process) monitor(
ctx context.Context,
cmd *exec.Cmd,
exited <-chan syscall.WaitStatus,
runExited, done, stopRequested chan struct{},
) {
defer close(done)
// The status comes from the reaper rather than cmd.Wait(): one waiter for
// the whole daemon is what keeps orphan reaping from stealing it.
status := <-exited
close(runExited)
// Read before releasing the handle below, which sets Process.Pid to -1.
pid := cmd.Process.Pid
// The process is reaped, but anything it spawned is still in its group and
// would survive as an orphan. Clear the group first: those grandchildren
// also hold the log pipe open, so removing them lets the drain finish
// instead of timing out.
p.killLingeringGroup(pid)
// Nothing waited on the os.Process, so the handle Start allocated is still
// open. Releasing it closes the pidfd, which would otherwise leak one
// descriptor per run.
releaseErr := cmd.Process.Release()
if releaseErr != nil {
p.logger.Debug("Failed to release the process handle", "error", releaseErr)
}
// Probing a process that has exited would keep reporting on whatever else
// answers at that address until something stops the checker.
p.stopHealthChecks()
// Flush whatever the process left in the pipes before recording the exit.
p.stopLogging()
p.mu.Lock()
p.exitCode = exitCodeOfStatus(status)
p.stopTime = time.Now()
p.lastError = exitErrorOf(status)
// A run that lasted long enough is treated as successful, so max_restarts
// bounds consecutive crashes rather than the lifetime restart count.
if p.stopTime.Sub(p.startTime) >= healthyUptime {
p.restartCount = 0
}
stoppedExternally := p.stoppedExternally
exitCode := p.exitCode
p.mu.Unlock()
switch currentState := p.GetState(); {
case currentState == StateStopping && stoppedExternally:
p.logger.Info("Process stopped", "exit_code", exitCode)
p.setState(StateStopped)
case currentState == StateStopping:
p.logger.Info("Process exited during stop", "exit_code", exitCode)
p.setState(StateExited)
case stoppedExternally:
p.logger.Info("Process exited before it was stopped", "exit_code", exitCode)
p.setState(StateStopped)
default:
p.logger.Info("Process exited", "exit_code", exitCode)
// A clean exit nobody asked for is the program finishing its work.
// Latched before the state change, because that change is what wakes
// anything waiting for this program to complete.
if exitCode == 0 {
p.mu.Lock()
p.completed = true
p.mu.Unlock()
}
p.setState(StateExited)
p.maybeRestart(ctx, stopRequested, exitCode)
}
}
// maybeRestart applies the autorestart policy after an unsupervised exit
func (p *Process) maybeRestart(ctx context.Context, stopRequested chan struct{}, exitCode int) {
shouldRestart := false
switch p.config.Autorestart {
case config.RestartAlways:
shouldRestart = true
p.logger.Debug("Autorestart policy is 'always', will restart")
case config.RestartUnexpected:
shouldRestart = exitCode != 0
p.logger.Debug("Autorestart policy is 'unexpected'", "exit_code", exitCode, "will_restart", shouldRestart)
case config.RestartNever:
p.logger.Debug("Autorestart policy is 'never', will not restart")
}
if !shouldRestart {
return
}
p.mu.Lock()
if p.restartCount >= p.config.MaxRestarts {
p.mu.Unlock()
p.logger.Error("Exceeded maximum restart attempts, setting state to FATAL", "max_restarts", p.config.MaxRestarts)
p.setState(StateFatal)
return
}
p.restartCount++
attempt := p.restartCount
p.mu.Unlock()
backoff := backoffDuration(attempt)
p.logger.Info("Restart attempt", "attempt", attempt, "max_restarts", p.config.MaxRestarts, "backoff", backoff)
select {
case <-time.After(backoff):
case <-ctx.Done():
p.logger.Info("Restart canceled")
return
case <-stopRequested:
p.logger.Info("Restart canceled")
return
}
p.mu.RLock()
canceled := p.stoppedExternally || p.state == StateStopping
p.mu.RUnlock()
if canceled {
p.logger.Info("Restart canceled")
return
}
p.logger.Info("Attempting restart after backoff")
p.setState(StateBackoff)
if err := p.Start(); err != nil {
p.logger.Error("Restart failed", "error", err)
p.setState(StateFatal)
}
}
// backoffDuration returns the delay before restart attempt n, counting from 1
func backoffDuration(attempt int) time.Duration {
shift := attempt - 1
if shift < 0 {
shift = 0
}
if shift > maxBackoffShift {
shift = maxBackoffShift
}
return min(time.Duration(1<<uint(shift))*time.Second, maxBackoffSeconds*time.Second)
}
// exitErrorOf describes a non-successful exit, and is nil for a clean one. It
// stands in for the error cmd.Wait() used to return, which callers surface as
// the program's last error.
func exitErrorOf(status syscall.WaitStatus) error {
switch {
case status.Signaled():
return fmt.Errorf("signal: %s", status.Signal())
case status.ExitStatus() != 0:
return fmt.Errorf("exit status %d", status.ExitStatus())
default:
return nil
}
}
// startLogging creates this run's capture pipes and returns the descriptors to
// hand to the child. A nil descriptor means the stream is discarded.
func (p *Process) startLogging() (stdout, stderr *os.File, err error) {
stdoutPath := p.config.StdoutLogfile
stderrPath := p.config.StderrLogfile
var stream *logStream
// One file means one pipe, so the two streams interleave in write order
// instead of racing two writers against the same path.
if stdoutPath != "" && stdoutPath == stderrPath {
stream, err = p.addLogStream(
stdoutPath,
max(p.config.StdoutLogfileMaxBytes, p.config.StderrLogfileMaxBytes),
max(p.config.StdoutLogfileBackups, p.config.StderrLogfileBackups),
max(p.config.StdoutLogfileMaxAge, p.config.StderrLogfileMaxAge),
)
if err != nil {
return nil, nil, err
}
return stream.childEnd, stream.childEnd, nil
}
if stdoutPath != "" {
stream, err = p.addLogStream(
stdoutPath,
p.config.StdoutLogfileMaxBytes,
p.config.StdoutLogfileBackups,
p.config.StdoutLogfileMaxAge,
)
if err != nil {
return nil, nil, err
}
stdout = stream.childEnd
}
if stderrPath != "" {
stream, err = p.addLogStream(
stderrPath,
p.config.StderrLogfileMaxBytes,
p.config.StderrLogfileBackups,
p.config.StderrLogfileMaxAge,
)
if err != nil {
return nil, nil, err
}
stderr = stream.childEnd
}
return stdout, stderr, nil
}
// addLogStream opens a rotating log file and starts draining a pipe into it
func (p *Process) addLogStream(path string, maxBytes int64, backups, maxAge int) (*logStream, error) {
sink, err := logrotate.NewWriter(path, maxBytes, backups, maxAge)
if err != nil {
return nil, err
}
readEnd, childEnd, err := os.Pipe()
if err != nil {
_ = sink.Close()
return nil, fmt.Errorf("failed to create log pipe for %s: %w", path, err)
}
stream := &logStream{
sink: sink,
readEnd: readEnd,
childEnd: childEnd,
done: make(chan struct{}),
}
p.mu.Lock()
p.streams = append(p.streams, stream)
p.mu.Unlock()
go func() {
defer close(stream.done)
if err := drainStream(stream.sink, stream.readEnd); err != nil {
p.logger.Warn("Log capture ended with an error", "path", path, "error", err)
}
}()
return stream, nil
}
// drainStream copies the pipe into the log one line at a time. Copying in bulk
// would hand the writer chunks far larger than the rotation threshold, so the
// log would overshoot maxbytes by a whole read buffer on every rotation.
func drainStream(sink io.Writer, pipe io.Reader) error {
reader := bufio.NewReaderSize(pipe, maxLogLineBytes)
for {
line, err := reader.ReadSlice('\n')
if len(line) > 0 {
if _, writeErr := sink.Write(line); writeErr != nil {
return writeErr
}
}
switch {
case err == nil, errors.Is(err, bufio.ErrBufferFull):
// A line longer than the buffer is written in pieces
continue
case errors.Is(err, io.EOF), errors.Is(err, os.ErrClosed):
return nil
default:
return err
}
}
}
// closeChildEnds releases the parent's copy of each pipe write end, which the
// child inherited during Start
func (p *Process) closeChildEnds() {
p.mu.RLock()
streams := p.streams
p.mu.RUnlock()
for _, stream := range streams {
if stream.childEnd != nil {
_ = stream.childEnd.Close()
stream.childEnd = nil
}
}
}
// stopLogging drains what the process left in the pipes and closes the log
// files. It is safe to call more than once.
func (p *Process) stopLogging() {
p.mu.Lock()
streams := p.streams
p.streams = nil
p.mu.Unlock()
for _, stream := range streams {
if stream.childEnd != nil {
_ = stream.childEnd.Close()
stream.childEnd = nil
}
// A pipe only reaches EOF once every writer has closed it, and a
// grandchild that outlived the process still holds one, so the drain
// gets a bounded window before we take the read end away from it.
select {
case <-stream.done:
case <-time.After(logDrainTimeout):
p.logger.Warn("Log capture still open after exit, closing it", "path", stream.sink.Path())
_ = stream.readEnd.Close()
<-stream.done
}
_ = stream.readEnd.Close()
if err := stream.sink.Close(); err != nil {
p.logger.Warn("Failed to close log file", "path", stream.sink.Path(), "error", err)
}
}
}
// parseCommand splits a command string into its arguments.
//
// This is not a shell: there is no expansion, globbing or substitution. It
// handles quoting and backslash escapes so that arguments containing spaces,
// quotes or an empty string can be expressed.
func parseCommand(cmd string) []string {
parts := []string{}
scan := commandScanner{}
for i := 0; i < len(cmd); i++ {
// A backslash escapes the next character, except inside single quotes
// where a shell treats it literally too.
if cmd[i] == '\\' && scan.quoteChar != '\'' && i+1 < len(cmd) {
i++
scan.current = append(scan.current, cmd[i])
continue
}
if arg, complete := scan.step(cmd[i]); complete {
parts = append(parts, arg)
}
}
if arg, complete := scan.finish(); complete {
parts = append(parts, arg)
}
return parts
}
// commandScanner tracks quoting while splitting a command string
type commandScanner struct {
current []byte
quoteChar byte
inQuotes bool
// quoted records that this argument was written with quotes, so that ""
// produces an empty argument rather than nothing at all.
quoted bool
}
// step consumes one character, returning an argument when one is complete
func (s *commandScanner) step(char byte) (arg string, complete bool) {
switch {
case (char == '"' || char == '\'') && !s.inQuotes:
s.inQuotes = true
s.quoteChar = char
s.quoted = true
case s.inQuotes && char == s.quoteChar:
s.inQuotes = false
s.quoteChar = 0
case (char == ' ' || char == '\t') && !s.inQuotes:
return s.finish()
default:
s.current = append(s.current, char)
}
return "", false
}
// finish closes off the argument being scanned, if there is one
func (s *commandScanner) finish() (arg string, complete bool) {
if len(s.current) == 0 && !s.quoted {
return "", false
}
arg = string(s.current)
s.current = s.current[:0]
s.quoted = false
return arg, true
}
// Signal sends a signal to the process
func (p *Process) Signal(sig os.Signal) error {
p.mu.RLock()
cmd := p.cmd
p.mu.RUnlock()
if cmd == nil || cmd.Process == nil {
return fmt.Errorf("process is not running")
}
return cmd.Process.Signal(sig)
}
package process
import (
"errors"
"log/slog"
"os"
"os/signal"
"sync"
"syscall"
"time"
)
// reapBackstop bounds how long an exit can go unnoticed if a SIGCHLD is missed.
// Signals coalesce, so several children exiting at once can produce one
// notification; the reap loop drains everything it can each time, and this only
// covers the case where a signal is lost entirely.
const reapBackstop = time.Second
var (
reaperOnce sync.Once
reaper *childReaper
)
// childReaper owns every wait4 in the daemon.
//
// Waiting is a property of the process, not of any one Process object:
// wait4(-1) takes whatever has exited, so a second waiter steals statuses from
// the first. os/exec's Wait is exactly such a second waiter, which is why
// nothing here calls cmd.Wait() and every exit arrives through this type.
//
// Reaping anything, rather than only what we started, is also what makes
// supavisor safe to run as PID 1: a grandchild orphaned by a wrapper script is
// reparented to us, and if nobody waits for it, it stays a zombie forever.
type childReaper struct {
logger *slog.Logger
waiters map[int]chan syscall.WaitStatus
// forking excludes reaping, so a status cannot be collected for a PID that
// a start has not registered yet. It is read-held across the fork, which
// lets starts run in parallel with each other: holding one exclusive lock
// across every fork made a slow exec stall every other program's start and
// all exit delivery with it.
forking sync.RWMutex
// mu guards waiters alone, and is never held across a fork or a syscall
// that can block.
mu sync.Mutex
}
// StartReaping starts the daemon's reaper. Call it once, from daemon startup,
// with the root logger.
//
// Doing it here rather than on the first fork is what makes the PID 1 guarantee
// unconditional: a container whose programs are all autostart: false, or one
// still working through dependencies, would otherwise have no wait4 loop at all
// and would accumulate any orphan reparented onto it in the meantime.
func StartReaping(logger *slog.Logger) {
reaperFor(logger)
}
// reaperFor returns the daemon's reaper, starting it if StartReaping has not
// run. The fallback keeps a Process usable on its own, in tests and anywhere
// else that builds one directly, at the cost of the reaper inheriting that
// caller's logger.
func reaperFor(logger *slog.Logger) *childReaper {
reaperOnce.Do(func() {
reaper = &childReaper{
logger: logger.With("component", "reaper"),
waiters: make(map[int]chan syscall.WaitStatus),
}
go reaper.run()
})
return reaper
}
// spawn starts a child and registers it in one step, so an exit cannot be
// reaped before we know who it belongs to. start does the fork, and returns the
// PID it produced.
//
// # Why the lock spans the whole fork, and what it costs
//
// wait4 is destructive: it reports which child exited only by collecting it. So
// a status for a PID nobody has registered cannot be set aside for a moment and
// identified later, and it cannot be left uncollected either. Excluding
// reaping for the duration of the fork is what makes registration and
// collection mutually exclusive.
//
// The cost is real and known: a start wedged in exec, on a hung mount or
// anything else that blocks in the loader, stalls exit delivery for every other
// program until it returns. The exposure is a stalled daemon, which the
// reconciler and an operator can both see.
//
// The obvious way to avoid that is to drop the barrier and hold statuses for
// unregistered PIDs while any start is in flight, letting each start claim its
// own on the way past. It does not work, and the failure is worse than the
// stall: an in-flight count is global, so it cannot say that a given PID
// belongs to a start. As PID 1 every orphan exits into that holding area, and
// collecting one frees its PID for reuse. A later start handed the same number
// takes the orphan's status, reports a running program as exited, kills its
// process group and then tracks nothing. A time bound narrows that window; it
// does not close it. This was tried in #22 and withdrawn.
//
// Closing it properly needs a non-destructive look at what has exited —
// waitid(P_ALL, WEXITED|WNOHANG|WNOWAIT), which names the PID and leaves the
// zombie in place. Go does not expose it: x/sys's Siginfo stops at Signo, Errno
// and Code with the union left as opaque bytes, so si_pid cannot be read, and
// the standard library has no Waitid at all. It would mean laying out the
// siginfo ABI by hand, in this file. Worth revisiting if that changes.
func (r *childReaper) spawn(start func() (int, error)) (<-chan syscall.WaitStatus, error) {
// Read-held across the fork: reapAll takes it exclusively, so it cannot
// collect a status for a PID that is still being registered, while other
// starts are free to proceed alongside this one.
r.forking.RLock()
defer r.forking.RUnlock()
pid, err := start()
if err != nil {
return nil, err
}
// Buffered so the reaper never blocks on a caller that is not listening yet.
exited := make(chan syscall.WaitStatus, 1)
r.mu.Lock()
defer r.mu.Unlock()
r.waiters[pid] = exited
return exited, nil
}
func (r *childReaper) run() {
// Buffered: a SIGCHLD arriving while we are already reaping must not be
// dropped, or the exit it refers to waits for the backstop.
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGCHLD)
ticker := time.NewTicker(reapBackstop)
defer ticker.Stop()
for {
select {
case <-sigChan:
case <-ticker.C:
}
r.reapAll()
}
}
// reapAll collects every child that has exited, dispatching each status to
// whoever started it. It drains rather than taking one, because signals
// coalesce and several children can exit between two notifications.
func (r *childReaper) reapAll() {
// No fork may be in flight while statuses are being collected. See spawn
// for why this has to cover the whole fork, and what it costs.
r.forking.Lock()
defer r.forking.Unlock()
for {
var status syscall.WaitStatus
pid, err := syscall.Wait4(-1, &status, syscall.WNOHANG, nil)
if pid <= 0 {
// 0: children exist, none have exited. ECHILD: none at all.
if err != nil && !errors.Is(err, syscall.ECHILD) {
r.logger.Debug("wait4 failed", "error", err)
}
return
}
r.mu.Lock()
exited := r.waiters[pid]
delete(r.waiters, pid)
r.mu.Unlock()
if exited == nil {
// Nobody started this: an orphan reparented to us because we are
// PID 1. Reaping it is the whole point; there is no one to tell.
r.logger.Debug("Reaped an orphan", "pid", pid)
continue
}
exited <- status
close(exited)
}
}
// signalGroupIfUnreaped signals pid's process group, but only while the reaper
// still holds that PID. Once wait4 has collected it the kernel is free to hand
// the number to something else, and kill(-pid) would then reach an unrelated
// process group. Held under the same lock reapAll uses to drop the waiter, so
// the check cannot go stale between here and the signal.
func (r *childReaper) signalGroupIfUnreaped(pid int, sig syscall.Signal) error {
if pid <= 0 {
return nil
}
r.mu.Lock()
defer r.mu.Unlock()
if _, waiting := r.waiters[pid]; !waiting {
return nil
}
return SignalGroup(pid, sig)
}
// exitCodeSignalled is what a process killed by a signal reports, matching
// os.ProcessState.ExitCode.
const exitCodeSignalled = -1
// exitCodeOfStatus reports the exit code a wait status carries.
//
// A signaled process reports -1 rather than 128+signal. That is what
// os.ProcessState.ExitCode gives, which is what this used to be built on, and
// it reaches operators through the exit_code field of the status API. The shell
// convention would be more informative, but changing a documented field is a
// deliberate act rather than a side effect of moving where the status is read.
func exitCodeOfStatus(status syscall.WaitStatus) int {
if status.Signaled() {
return exitCodeSignalled
}
return status.ExitStatus()
}
package process
// State represents the state of a process
type State string
const (
StateStopped State = "STOPPED"
StateStarting State = "STARTING"
StateRunning State = "RUNNING"
StateBackoff State = "BACKOFF" // Failed to start, waiting before retry
StateStopping State = "STOPPING"
StateExited State = "EXITED"
StateFatal State = "FATAL" // Failed to start after all retries
StateUnknown State = "UNKNOWN"
)
// String returns the string representation of the state
func (s State) String() string {
return string(s)
}
// IsRunning returns true if the process is in a running state
func (s State) IsRunning() bool {
return s == StateRunning
}
// IsStopped returns true if the process is stopped
func (s State) IsStopped() bool {
return s == StateStopped || s == StateExited || s == StateFatal
}
//go:build linux
package server
import (
"fmt"
"os"
"strings"
)
// bootIDPath identifies the current boot. A process start time on Linux is
// measured in ticks since boot, so it means nothing on its own once the machine
// has restarted.
const bootIDPath = "/proc/sys/kernel/random/boot_id"
// startTimeField is the index of the process start time within /proc/<pid>/stat
// once the fields before and including the command name have been dropped. The
// start time is field 22 and the slice begins at field 3, so it sits at 19.
const startTimeField = 19
// processStartToken returns an opaque identifier for the current run of pid.
//
// A PID on its own does not identify a process: the kernel reuses PIDs, so a
// PID recorded before a crash may belong to something entirely unrelated by the
// time supavisor starts again. Field 22 of /proc/<pid>/stat is the start time in
// clock ticks since boot, which is fixed for the life of the process and
// distinguishes it from a later one that inherited its number.
func processStartToken(pid int) (string, error) {
data, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid))
if err != nil {
return "", fmt.Errorf("failed to read process info for pid %d: %w", pid, err)
}
// The second field is the command name in parentheses and may itself
// contain spaces and parentheses, so fields are counted from the last ')'.
stat := string(data)
end := strings.LastIndex(stat, ")")
if end < 0 {
return "", fmt.Errorf("unrecognized /proc/%d/stat format", pid)
}
fields := strings.Fields(stat[end+1:])
if len(fields) <= startTimeField {
return "", fmt.Errorf("unrecognized /proc/%d/stat format", pid)
}
return fields[startTimeField], nil
}
// bootID identifies the running boot, so that process start times recorded
// before a restart are recognizable as belonging to a different one.
func bootID() (string, error) {
data, err := os.ReadFile(bootIDPath)
if err != nil {
return "", fmt.Errorf("failed to read boot id: %w", err)
}
id := strings.TrimSpace(string(data))
if id == "" {
return "", fmt.Errorf("%s is empty", bootIDPath)
}
return id, nil
}
package server
import (
"encoding/json"
"fmt"
"io"
"log/slog"
"net"
"os"
"os/user"
"strconv"
"sync"
"syscall"
"time"
"github.com/ademidoff/supavisor/internal/api"
)
const (
msgProcessNameRequired = "process name required"
// socketMode keeps the control socket off-limits to arbitrary local users.
// Anyone who can write to it can stop supervised processes and, through
// start, cause configured commands to run as the daemon's user.
socketMode = 0o660
// maxConnections bounds how many clients can occupy the daemon at once, so
// a local process cannot exhaust its goroutines or descriptors.
maxConnections = 32
// maxRequestBytes bounds a single request. Without it a client could stream
// an unbounded body into the decoder and exhaust memory.
maxRequestBytes = 64 * 1024
// connectionTimeout drops a client that connects and then does nothing,
// rather than holding the slot open indefinitely.
connectionTimeout = 2 * time.Minute
// acceptRetryDelay backs off after an accept error that is not fatal, such
// as running out of descriptors. Retrying flat out would spin on the CPU.
acceptRetryDelay = 50 * time.Millisecond
)
// IPCServer handles communication with the CLI tool
type IPCServer struct {
listener net.Listener
server *Server
stopChan chan struct{}
// accepting is closed once the accept loop has returned, or immediately if
// it never started. Only that loop adds to connections, so waiting for it
// is what lets Stop wait on the group safely.
accepting chan struct{}
slots chan struct{}
socketInfo os.FileInfo
socketPath string
socketGroup string
connections sync.WaitGroup
}
// NewIPCServer creates a new IPC server
func NewIPCServer(socketPath, socketGroup string, server *Server) *IPCServer {
return &IPCServer{
socketPath: socketPath,
socketGroup: socketGroup,
server: server,
stopChan: make(chan struct{}),
accepting: make(chan struct{}),
slots: make(chan struct{}, maxConnections),
}
}
// Start starts the IPC server
func (s *IPCServer) Start() error {
// Safe to clear a leftover socket: the caller holds the PID file lock, so
// no other daemon can be listening on this path.
if err := os.Remove(s.socketPath); err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to remove existing socket: %w", err)
}
listener, err := net.Listen("unix", s.socketPath) //nolint:noctx
if err != nil {
return fmt.Errorf("failed to listen on socket: %w", err)
}
// Unlink on our own terms. Go removes the socket path on Close even when a
// newer daemon has already replaced it, which would leave that daemon
// listening on a socket nothing can reach.
if unixListener, ok := listener.(*net.UnixListener); ok {
unixListener.SetUnlinkOnClose(false)
}
s.listener = listener
info, err := os.Stat(s.socketPath)
if err != nil {
return s.startFailed(fmt.Errorf("failed to stat socket: %w", err))
}
s.socketInfo = info
if err := s.restrictSocket(); err != nil {
return s.startFailed(err)
}
go func() {
defer close(s.accepting)
s.acceptConnections()
}()
return nil
}
// startFailed reports a failure from Start, releasing what Stop would otherwise
// wait for: the accept loop never began, so nothing else will close this.
func (s *IPCServer) startFailed(err error) error {
close(s.accepting)
return err
}
// restrictSocket narrows who can talk to the daemon
func (s *IPCServer) restrictSocket() error {
if s.socketGroup != "" {
gid, err := lookupGroupID(s.socketGroup)
if err != nil {
return err
}
if err := os.Chown(s.socketPath, -1, gid); err != nil {
return fmt.Errorf("failed to give socket to group %s: %w", s.socketGroup, err)
}
}
if err := os.Chmod(s.socketPath, socketMode); err != nil {
return fmt.Errorf("failed to set socket permissions: %w", err)
}
return nil
}
// lookupGroupID resolves a group name or numeric id
func lookupGroupID(group string) (int, error) {
if gid, err := strconv.Atoi(group); err == nil {
return gid, nil
}
grp, err := user.LookupGroup(group)
if err != nil {
return 0, fmt.Errorf("failed to look up socket_group %s: %w", group, err)
}
gid, err := strconv.Atoi(grp.Gid)
if err != nil {
return 0, fmt.Errorf("group %s has an unusable gid %s: %w", group, grp.Gid, err)
}
return gid, nil
}
// Stop stops the IPC server
func (s *IPCServer) Stop() error {
close(s.stopChan)
if s.listener == nil {
return nil
}
err := s.listener.Close()
// Before waiting on the group: the accept loop is the only thing that adds
// to it, and an Add that starts while the counter is zero and a Wait is
// already running is a WaitGroup misuse, which the race detector reports as
// a data race. Closing the listener is what ends the loop, so this returns
// promptly.
<-s.accepting
s.connections.Wait()
if s.socketInfo != nil {
if rmErr := removeIfSame(s.socketPath, s.socketInfo); rmErr != nil {
slog.Warn("failed to remove socket", "path", s.socketPath, "error", rmErr)
}
}
return err
}
// acceptConnections accepts incoming connections
func (s *IPCServer) acceptConnections() {
for {
conn, err := s.listener.Accept()
if err != nil {
select {
case <-s.stopChan:
return
default:
}
// Retrying immediately on a persistent error, such as running out
// of descriptors, would spin on the CPU for as long as it lasts.
slog.Warn("failed to accept connection", "error", err)
time.Sleep(acceptRetryDelay)
continue
}
// Hold the client rather than dropping it, so that a burst queues
// instead of failing, but never run more than maxConnections at once.
select {
case s.slots <- struct{}{}:
case <-s.stopChan:
_ = conn.Close()
return
}
s.connections.Add(1)
go func() {
defer func() {
<-s.slots
s.connections.Done()
}()
s.handleConnection(conn)
}()
}
}
// handleConnection handles a single connection
func (s *IPCServer) handleConnection(conn net.Conn) {
defer conn.Close()
// A client that connects and then goes quiet must not hold its slot for
// ever. The deadline is extended for each request it actually sends.
encoder := json.NewEncoder(conn)
for {
if err := conn.SetDeadline(time.Now().Add(connectionTimeout)); err != nil {
break
}
// A new decoder per request keeps the size limit per request rather
// than over the lifetime of the connection.
decoder := json.NewDecoder(io.LimitReader(conn, maxRequestBytes))
var req api.Request
if err := decoder.Decode(&req); err != nil {
break
}
resp := s.handleRequest(&req)
if err := encoder.Encode(resp); err != nil {
break
}
}
}
// handleRequest handles a request and returns a response
func (s *IPCServer) handleRequest(req *api.Request) *api.Response {
switch req.Command {
case api.CommandStatus:
return s.handleStatus(req.Args)
case api.CommandStart:
if len(req.Args) == 0 {
return &api.Response{Success: false, Message: msgProcessNameRequired}
}
return s.handleStart(req.Args[0])
case api.CommandStop:
if len(req.Args) == 0 {
return &api.Response{Success: false, Message: msgProcessNameRequired}
}
return s.handleStop(req.Args[0])
case api.CommandRestart:
if len(req.Args) == 0 {
return &api.Response{Success: false, Message: msgProcessNameRequired}
}
return s.handleRestart(req.Args[0])
case api.CommandReload:
return s.handleReload()
case api.CommandShutdown:
return s.handleShutdown()
default:
return &api.Response{Success: false, Message: fmt.Sprintf("unknown command: %s", req.Command)}
}
}
// handleStatus returns the status of every process, or of one named process
func (s *IPCServer) handleStatus(args []string) *api.Response {
statuses := s.server.GetStatus()
if len(args) > 0 {
statuses = selectProcess(statuses, args[0])
if len(statuses) == 0 {
return &api.Response{Success: false, Message: fmt.Sprintf("process %s not found", args[0])}
}
}
processStatuses := make([]api.ProcessStatus, 0, len(statuses))
for _, status := range statuses {
processStatuses = append(processStatuses, api.ProcessStatus{
Name: status.Name,
State: string(status.State),
Desired: string(status.Desired),
Health: string(status.Health),
Reason: status.Reason,
PID: status.PID,
ExitCode: status.ExitCode,
RestartCount: status.RestartCount,
Uptime: status.Uptime,
})
}
return &api.Response{
Success: true,
Data: map[string]any{"processes": processStatuses},
}
}
// selectProcess narrows a status list to one program, or to nothing if that
// program is not configured
func selectProcess(statuses []ProcessStatusInfo, name string) []ProcessStatusInfo {
for _, status := range statuses {
if status.Name == name {
return []ProcessStatusInfo{status}
}
}
return nil
}
// handleStart starts a process
func (s *IPCServer) handleStart(name string) *api.Response {
if err := s.server.StartProcess(name); err != nil {
return &api.Response{Success: false, Message: err.Error()}
}
return &api.Response{Success: true, Message: fmt.Sprintf("process %s started", name)}
}
// handleStop stops a process
func (s *IPCServer) handleStop(name string) *api.Response {
if err := s.server.StopProcess(name); err != nil {
return &api.Response{Success: false, Message: err.Error()}
}
return &api.Response{Success: true, Message: fmt.Sprintf("process %s stopped", name)}
}
// handleRestart restarts a process
func (s *IPCServer) handleRestart(name string) *api.Response {
if err := s.server.RestartProcess(name); err != nil {
return &api.Response{Success: false, Message: err.Error()}
}
return &api.Response{Success: true, Message: fmt.Sprintf("process %s restarted", name)}
}
// handleReload reloads the configuration and reports what it applied
func (s *IPCServer) handleReload() *api.Response {
applied, err := s.server.Reload()
if err != nil {
return &api.Response{Success: false, Message: err.Error()}
}
if applied.Empty() {
return &api.Response{Success: true, Message: "configuration reloaded, nothing changed"}
}
return &api.Response{Success: true, Message: "configuration reloaded", Data: applied}
}
// handleShutdown shuts down the supavisor
func (s *IPCServer) handleShutdown() *api.Response {
go func() {
// Send SIGTERM to ourselves
pid := os.Getpid()
proc, err := os.FindProcess(pid)
if err != nil {
slog.Error("failed to find process", "pid", pid, "error", err)
return
}
err = proc.Signal(syscall.SIGTERM)
if err != nil {
slog.Error("failed to send SIGTERM", "error", err)
}
}()
return &api.Response{Success: true, Message: "shutdown initiated"}
}
package server
import (
"fmt"
"os"
"path/filepath"
"strconv"
"strings"
"syscall"
)
// pidLock is an exclusive lock held on the daemon's PID file for as long as the
// daemon runs.
//
// The lock, not the file contents, is what keeps two daemons from running at
// once. Reading the file and checking whether that PID is alive is a race, and
// it cannot tell a live daemon from a crashed one, which is why a crash used to
// leave behind a file that an operator had to remove by hand before supavisor
// would start again. A flock is released by the kernel when the holder dies, so
// a crashed daemon leaves nothing to clean up.
type pidLock struct {
file *os.File
path string
}
// maxPIDFileBytes bounds how much of a pid file is read when reporting which
// daemon holds the lock.
const maxPIDFileBytes = 32
// acquirePIDLock locks path and records the current PID in it
func acquirePIDLock(path string) (*pidLock, error) {
if dir := filepath.Dir(path); dir != "" && dir != "." {
if err := os.MkdirAll(dir, 0o755); err != nil {
return nil, fmt.Errorf("failed to create pid file directory %s: %w", dir, err)
}
}
file, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o644)
if err != nil {
return nil, fmt.Errorf("failed to open pid file %s: %w", path, err)
}
if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
owner := readPID(file)
_ = file.Close()
if owner > 0 {
return nil, fmt.Errorf("supavisor is already running (PID: %d)", owner)
}
return nil, fmt.Errorf("pid file %s is held by another process: %w", path, err)
}
if err := writePID(file, os.Getpid()); err != nil {
_ = file.Close()
return nil, err
}
return &pidLock{file: file, path: path}, nil
}
// Release removes the PID file and drops the lock
func (l *pidLock) Release() error {
// Unlink only our own file. A shutdown can take several seconds and overlap
// with the next daemon's startup, and removing by path alone would delete
// the incoming daemon's PID file, leaving it running and unfindable.
var rmErr error
if info, err := l.file.Stat(); err == nil {
rmErr = removeIfSame(l.path, info)
}
// Closing the descriptor releases the flock.
if err := l.file.Close(); err != nil {
return err
}
return rmErr
}
func writePID(file *os.File, pid int) error {
if err := file.Truncate(0); err != nil {
return fmt.Errorf("failed to truncate pid file: %w", err)
}
if _, err := file.Seek(0, 0); err != nil {
return fmt.Errorf("failed to rewind pid file: %w", err)
}
if _, err := fmt.Fprintf(file, "%d\n", pid); err != nil {
return fmt.Errorf("failed to write pid file: %w", err)
}
return file.Sync()
}
// readPID reads the PID recorded in an already-open pid file
func readPID(file *os.File) int {
if _, err := file.Seek(0, 0); err != nil {
return 0
}
buf := make([]byte, maxPIDFileBytes)
n, err := file.Read(buf)
if n == 0 || (err != nil && n == 0) {
return 0
}
pid, err := strconv.Atoi(strings.TrimSpace(string(buf[:n])))
if err != nil {
return 0
}
return pid
}
// removeIfSame removes path only if it still refers to the same file we created.
// Anything else means a newer daemon has already replaced it and it is not ours
// to delete.
func removeIfSame(path string, want os.FileInfo) error {
got, err := os.Stat(path)
if err != nil {
if os.IsNotExist(err) {
return nil
}
return err
}
if !os.SameFile(want, got) {
return nil
}
return os.Remove(path)
}
package server
import (
"fmt"
"sort"
"time"
"github.com/ademidoff/supavisor/internal/config"
"github.com/ademidoff/supavisor/internal/process"
)
// DesiredState is what supavisor has been asked to keep a program at, which is
// separate from what the program happens to be doing right now.
type DesiredState string
const (
DesiredRunning DesiredState = "RUNNING"
DesiredStopped DesiredState = "STOPPED"
)
// reconcileInterval is the slowest supavisor will notice that a program has
// drifted from its desired state. Most of the time it does not wait for the
// tick: every process state change asks for a pass immediately.
const reconcileInterval = time.Second
// reconcileLoop drives programs towards their desired state until shutdown.
//
// This exists because starting a program is not a one-shot action: a program
// whose dependencies were not ready yet has to be started later, when they are.
// Without a loop that keeps looking, one dependency that was slow to come up
// left everything behind it stopped for good.
func (s *Server) reconcileLoop() {
defer close(s.reconcileDone)
ticker := time.NewTicker(reconcileInterval)
defer ticker.Stop()
for {
s.reconcile()
select {
case <-s.reconcileNow:
case <-ticker.C:
case <-s.stopChan:
return
}
}
}
// requestReconcile asks for a pass without waiting for the next tick. It never
// blocks, so it is safe to call from process state change callbacks.
func (s *Server) requestReconcile() {
select {
case s.reconcileNow <- struct{}{}:
default:
}
}
// reconcile moves every program one step towards its desired state.
//
// It does no work itself: anything that can take time runs on its own
// goroutine, so a program that takes seconds to stop does not hold up the rest.
func (s *Server) reconcile() {
for _, name := range s.programNames() {
s.processMutex.RLock()
desired := s.desired[name]
proc := s.processes[name]
busy := s.inflight[name]
s.processMutex.RUnlock()
if busy || proc == nil {
continue
}
state := proc.GetState()
switch {
// Only STOPPED is started. A process that exited or gave up on its own
// is left alone: restarts within a run are the autorestart policy's
// business, and starting a FATAL process here would defeat
// max_restarts entirely.
case desired == DesiredRunning && state == process.StateStopped:
ready, reason := s.dependenciesSatisfied(name)
if !ready {
s.noteBlocked(name, reason)
continue
}
s.noteBlocked(name, "")
s.logger.Info("Starting process", "process", name)
s.runAction(name, proc.Start)
// A program that has exited or given up is not running, but it is not
// released either: its monitor may be sitting in a restart backoff that
// only Stop() cancels. Leaving it alone let a stopped program spawn a
// run nobody was asking for, one backoff after the stop was reported
// as done.
case desired == DesiredStopped && state != process.StateStopped && state != process.StateStopping:
s.logger.Info("Stopping process", "process", name)
s.runAction(name, proc.Stop)
}
}
}
// noteBlocked reports why a program cannot start yet, once per distinct reason.
// An empty reason forgets what was last reported, so that a program blocked
// again later says so again.
//
// The reconciler runs every second and a dependency can be down for a long
// time, so reporting every pass would bury everything else in the log; saying
// nothing left a program that never starts with no explanation at all.
func (s *Server) noteBlocked(name, reason string) {
s.processMutex.Lock()
changed := s.blockedReason[name] != reason
if reason == "" {
delete(s.blockedReason, name)
} else {
s.blockedReason[name] = reason
}
s.processMutex.Unlock()
if changed && reason != "" {
s.logger.Info("Not starting yet", "process", name, "reason", reason)
}
}
// runAction performs a transition off the reconcile loop and marks the program
// busy so that the next pass does not start a second one
func (s *Server) runAction(name string, action func() error) {
s.processMutex.Lock()
s.inflight[name] = true
s.processMutex.Unlock()
s.actions.Add(1)
go func() {
defer s.actions.Done()
if err := action(); err != nil {
s.logger.Error("Failed to reconcile process", "process", name, "error", err)
}
s.processMutex.Lock()
delete(s.inflight, name)
s.processMutex.Unlock()
s.requestReconcile()
}()
}
// dependenciesSatisfied reports whether every program this one depends on has
// reached the condition it is waited on for, and if not, which one is holding
// it back
func (s *Server) dependenciesSatisfied(name string) (satisfied bool, reason string) {
for _, dep := range s.dependenciesOf(name) {
s.processMutex.RLock()
depProc, exists := s.processes[dep.Name]
s.processMutex.RUnlock()
if !exists {
return false, fmt.Sprintf("dependency %s is not configured", dep.Name)
}
// A one-off is waited on for having finished rather than for being up,
// so RUNNING is the wrong thing to ask of it: it is only running while
// the work is still in flight.
if dep.Condition == config.ConditionCompleted {
if !depProc.HasCompleted() {
return false, fmt.Sprintf("dependency %s has not completed, it is %s", dep.Name, depProc.GetState())
}
continue
}
if state := depProc.GetState(); state != process.StateRunning {
return false, fmt.Sprintf("dependency %s is %s", dep.Name, state)
}
// Running only means the process is alive. A program that initializes
// after startup is not ready to be depended on until its check passes.
if dep.Condition == config.ConditionHealthy {
if health := depProc.GetHealth(); health != process.HealthHealthy {
return false, fmt.Sprintf("dependency %s is running but its health check is %s", dep.Name, health)
}
}
}
return true, ""
}
// dependenciesOf returns what a program depends on, with the condition each
// dependency has to reach. The graph carries the ordering, the configuration
// carries the conditions.
func (s *Server) dependenciesOf(name string) []config.Dependency {
s.processMutex.RLock()
defer s.processMutex.RUnlock()
prog := s.config.Programs[name]
if prog == nil {
return nil
}
return prog.DependsOn
}
// programNames returns configured program names ordered by priority, lowest
// first, with names breaking ties so the order is stable.
//
// Dependencies still decide what may start; priority only settles the order
// among programs that are all ready at the same moment.
func (s *Server) programNames() []string {
s.processMutex.RLock()
defer s.processMutex.RUnlock()
names := make([]string, 0, len(s.processes))
for name := range s.processes {
names = append(names, name)
}
sort.Slice(names, func(i, j int) bool {
a, b := s.config.Programs[names[i]], s.config.Programs[names[j]]
if a != nil && b != nil && a.Priority != b.Priority {
return a.Priority < b.Priority
}
return names[i] < names[j]
})
return names
}
// process returns a managed process by name
func (s *Server) process(name string) *process.Process {
s.processMutex.RLock()
defer s.processMutex.RUnlock()
return s.processes[name]
}
// setDesired records what a program should be doing
func (s *Server) setDesired(name string, desired DesiredState) error {
s.processMutex.Lock()
defer s.processMutex.Unlock()
if _, exists := s.processes[name]; !exists {
return fmt.Errorf("process %s not found", name)
}
s.desired[name] = desired
// Nothing is holding back a program nobody is asking for, so a reason
// recorded while it was wanted must not outlive the request.
if desired == DesiredStopped {
delete(s.blockedReason, name)
}
return nil
}
// awaitState waits for the reconciler to bring a program to a settled state, so
// that a caller still learns what actually happened rather than only that the
// request was recorded.
func (s *Server) awaitState(name string, done func(process.State) bool) error {
proc := s.process(name)
if proc == nil {
return fmt.Errorf("process %s not found", name)
}
deadline := time.After(actionTimeout)
ticker := time.NewTicker(pollInterval)
defer ticker.Stop()
for {
if done(proc.GetState()) {
return nil
}
// Waiting out the timeout is only worth it while the outcome can still
// change. If a dependency is not even meant to come up, say so now.
//
// Only for a program that is wanted running: dependencies decide what
// may start, never what may stop, and a stop that was proceeding
// perfectly well used to report that the program could not start.
if s.wantsToRun(name) {
if blocked, reason := s.blockedIndefinitely(name); blocked {
return fmt.Errorf("process %s cannot start: %s", name, reason)
}
}
select {
case <-ticker.C:
case <-s.stopChan:
return fmt.Errorf("supavisor is shutting down")
case <-deadline:
return s.awaitTimeoutError(name, proc.GetState())
}
}
}
// wantsToRun reports whether a program is currently meant to be running
func (s *Server) wantsToRun(name string) bool {
s.processMutex.RLock()
defer s.processMutex.RUnlock()
return s.desired[name] == DesiredRunning
}
// blockedIndefinitely reports a dependency that is not merely down but has no
// prospect of coming up, so that waiting for it would never resolve.
//
// The walk is transitive: a dependency that is itself waiting on something that
// will never start is just as final as one that is stopped outright.
func (s *Server) blockedIndefinitely(name string) (blocked bool, reason string) {
return s.blockedByDependency(name, make(map[string]bool))
}
func (s *Server) blockedByDependency(name string, seen map[string]bool) (blocked bool, reason string) {
if seen[name] {
return false, ""
}
seen[name] = true
for _, dep := range s.dependenciesOf(name) {
s.processMutex.RLock()
depProc := s.processes[dep.Name]
depDesired := s.desired[dep.Name]
s.processMutex.RUnlock()
if depProc == nil {
return true, fmt.Sprintf("dependency %s is not configured", dep.Name)
}
state := depProc.GetState()
waitsForCompletion := dep.Condition == config.ConditionCompleted
switch {
// Work that is done stays done: what the program is left sitting in
// afterwards, and whether anyone means to run it again, no longer
// decides whether a dependent may start.
case waitsForCompletion && depProc.HasCompleted():
continue
case state == process.StateFatal:
return true, fmt.Sprintf("dependency %s gave up starting", dep.Name)
case depDesired == DesiredStopped && state.IsStopped():
return true, fmt.Sprintf("dependency %s is stopped and is not set to start", dep.Name)
case waitsForCompletion && s.exitedForGood(dep.Name, state):
return true, fmt.Sprintf("dependency %s exited with status %d and will not be run again",
dep.Name, depProc.GetExitCode())
case state == process.StateRunning:
continue
}
if blockedDeep, deepReason := s.blockedByDependency(dep.Name, seen); blockedDeep {
return true, deepReason
}
}
return false, ""
}
// exitedForGood reports a program that has exited and whose restart policy
// declined to run it again.
//
// Unlike a FATAL program, which gave up on its way somewhere, this one has
// settled: waiting for it to complete would never resolve. Only autorestart:
// never can leave it here, since the other policies either restart an
// unsuccessful exit or eventually reach FATAL.
func (s *Server) exitedForGood(name string, state process.State) bool {
if state != process.StateExited {
return false
}
s.processMutex.RLock()
defer s.processMutex.RUnlock()
return s.autorestartPolicy(name) == config.RestartNever
}
// awaitTimeoutError explains why a program never reached the expected state
func (s *Server) awaitTimeoutError(name string, state process.State) error {
if ready, reason := s.dependenciesSatisfied(name); !ready {
return fmt.Errorf("process %s is not running: %s", name, reason)
}
return fmt.Errorf("timed out waiting for process %s, it is %s", name, state)
}
package server
import (
"fmt"
"maps"
"reflect"
"slices"
"sync"
"github.com/ademidoff/supavisor/internal/api"
"github.com/ademidoff/supavisor/internal/config"
"github.com/ademidoff/supavisor/internal/dependency"
"github.com/ademidoff/supavisor/internal/process"
)
// Reload re-reads the configuration files and applies what changed, and reports
// what it applied.
//
// Programs that are unchanged keep running untouched. Programs that were
// removed are stopped and forgotten, programs that were added are picked up by
// the reconciler, and programs whose definition changed are stopped and
// replaced so they come back on the new definition.
func (s *Server) Reload() (api.ReloadResponse, error) {
// One reload at a time: two of them interleaving would compute their plans
// against the same starting point and then apply both.
s.reloadMutex.Lock()
defer s.reloadMutex.Unlock()
var none api.ReloadResponse
if s.config.SourcePath == "" {
return none, fmt.Errorf("no configuration file to reload")
}
newCfg, err := config.ParseConfig(s.config.SourcePath)
if err != nil {
return none, fmt.Errorf("failed to reload configuration: %w", err)
}
if err := newCfg.Validate(); err != nil {
return none, fmt.Errorf("invalid configuration: %w", err)
}
if _, err := buildDependencyGraph(newCfg).TopologicalSort(); err != nil {
return none, fmt.Errorf("dependency graph validation failed: %w", err)
}
if setting := changedDaemonSetting(&s.config.Supavisor, &newCfg.Supavisor); setting != "" {
return none, fmt.Errorf("%s cannot be changed while running: restart supavisor to apply it", setting)
}
if err := newCfg.EnsureLogDirectories(); err != nil {
return none, fmt.Errorf("failed to create log directories: %w", err)
}
added, removed, changed := diffPrograms(s.config.Programs, newCfg.Programs)
applied := api.ReloadResponse{Added: added, Removed: removed, Changed: changed}
if applied.Empty() {
s.logger.Info("Configuration reloaded, nothing changed")
return applied, nil
}
s.logger.Info("Reloading configuration", "added", added, "removed", removed, "changed", changed)
// Whatever an operator asked for survives the reload: a program that was
// deliberately stopped must not come back just because its definition moved.
s.processMutex.RLock()
previous := maps.Clone(s.desired)
s.processMutex.RUnlock()
// A program that is going away, or is about to be redefined, has to stop
// before it can be dropped or replaced.
s.stopForReload(slices.Concat(removed, changed))
s.processMutex.Lock()
s.applyPrograms(newCfg, previous, added, removed, changed)
s.processMutex.Unlock()
s.requestReconcile()
s.markStateDirty()
s.logger.Info("Configuration reloaded")
return applied, nil
}
// stopForReload stops programs that reload is about to drop or replace
func (s *Server) stopForReload(names []string) {
var wg sync.WaitGroup
for _, name := range names {
if err := s.setDesired(name, DesiredStopped); err != nil {
continue
}
wg.Go(func() {
err := s.awaitState(name, func(state process.State) bool {
return state == process.StateStopped
})
if err != nil {
s.logger.Warn("failed to stop process for reload", "process", name, "error", err)
}
})
}
s.requestReconcile()
wg.Wait()
}
// applyPrograms swaps in the new configuration. Must be called with
// processMutex held.
func (s *Server) applyPrograms(newCfg *config.Config, previous map[string]DesiredState, added, removed, changed []string) {
for _, name := range removed {
delete(s.processes, name)
delete(s.desired, name)
delete(s.inflight, name)
delete(s.blockedReason, name)
}
for _, name := range changed {
s.processes[name] = s.newProcess(newCfg.Programs[name])
s.desired[name] = previous[name]
// The program is being replaced, so whatever was reported about the
// previous definition should be reported again for this one.
delete(s.blockedReason, name)
}
for _, name := range added {
s.processes[name] = s.newProcess(newCfg.Programs[name])
s.desired[name] = DesiredStopped
if newCfg.Programs[name].Autostart {
s.desired[name] = DesiredRunning
}
}
s.config = newCfg
s.dependencyGraph = buildDependencyGraph(newCfg)
}
// diffPrograms reports which programs a new configuration adds, drops and
// redefines
func diffPrograms(old, updated map[string]*config.ProgramConfig) ([]string, []string, []string) {
var added, removed, changed []string
for name, newProg := range updated {
oldProg, exists := old[name]
switch {
case !exists:
added = append(added, name)
case !reflect.DeepEqual(oldProg, newProg):
changed = append(changed, name)
}
}
for name := range old {
if _, exists := updated[name]; !exists {
removed = append(removed, name)
}
}
slices.Sort(added)
slices.Sort(removed)
slices.Sort(changed)
return added, removed, changed
}
// changedDaemonSetting names a daemon-level setting that differs, or an empty
// string if they match. These are bound at startup: the PID file is locked and
// the socket is already listening, so changing them means restarting.
func changedDaemonSetting(old, updated *config.SupavisorConfig) string {
switch {
case old.PidFile != updated.PidFile:
return "pidfile"
case old.Socket != updated.Socket:
return "socket"
case old.SocketGroup != updated.SocketGroup:
return "socket_group"
case old.LogFile != updated.LogFile:
return "logfile"
case old.LogFormat != updated.LogFormat:
return "log_format"
case old.LogLevel != updated.LogLevel:
return "log_level"
}
return ""
}
// buildDependencyGraph builds the dependency graph for a configuration
func buildDependencyGraph(cfg *config.Config) *dependency.Graph {
graph := dependency.NewGraph()
for name, progConfig := range cfg.Programs {
graph.AddNode(name, progConfig.DependencyNames())
}
return graph
}
// newProcess creates a managed process wired to this server's callbacks
func (s *Server) newProcess(cfg *config.ProgramConfig) *process.Process {
proc := process.NewProcess(cfg, s.processLogger)
proc.SetStateChangeCallback(s.onProcessStateChange)
proc.SetHealthChangeCallback(s.onProcessHealthChange)
return proc
}
package server
import (
"fmt"
"log/slog"
"os"
"os/signal"
"sort"
"sync"
"syscall"
"time"
"github.com/ademidoff/supavisor/internal/config"
"github.com/ademidoff/supavisor/internal/dependency"
"github.com/ademidoff/supavisor/internal/process"
)
const (
// actionTimeout bounds how long a start or stop request waits for the
// reconciler to produce a settled outcome before reporting back.
actionTimeout = 30 * time.Second
pollInterval = 100 * time.Millisecond
// signalBuffer leaves room for a second stop signal to be noticed while the
// first one is being acted on.
signalBuffer = 4
exitOK = 0
exitFailed = 1
exitInterrupted = 130
)
// ProcessStatusInfo contains status information about a process
type ProcessStatusInfo struct {
Name string
State process.State
Desired DesiredState
Health process.Health
Uptime string
// Reason is why a program is not running, and is empty whenever there is
// nothing to explain.
Reason string
PID int
ExitCode int
RestartCount int
}
// Server manages all processes
type Server struct {
config *config.Config
logger *slog.Logger
processLogger *slog.Logger
processes map[string]*process.Process
desired map[string]DesiredState
inflight map[string]bool
blockedReason map[string]string
dependencyGraph *dependency.Graph
ipcServer *IPCServer
pidLock *pidLock
stateDirty chan struct{}
reconcileNow chan struct{}
reconcileDone chan struct{}
stopChan chan struct{}
stateFile string
bootID string
actions sync.WaitGroup
reloadMutex sync.Mutex
processMutex sync.RWMutex
running bool
}
// New creates a new server instance
func New(cfg *config.Config, logger *slog.Logger) (*Server, error) {
if err := cfg.Validate(); err != nil {
return nil, fmt.Errorf("invalid configuration: %w", err)
}
if err := cfg.EnsureLogDirectories(); err != nil {
return nil, fmt.Errorf("failed to create log directories: %w", err)
}
graph := buildDependencyGraph(cfg)
// Verify no circular dependencies
if _, err := graph.TopologicalSort(); err != nil {
return nil, fmt.Errorf("dependency graph validation failed: %w", err)
}
s := &Server{
config: cfg,
logger: logger.With("component", "main"),
processLogger: logger,
processes: make(map[string]*process.Process, len(cfg.Programs)),
desired: make(map[string]DesiredState, len(cfg.Programs)),
inflight: make(map[string]bool),
blockedReason: make(map[string]string),
dependencyGraph: graph,
stopChan: make(chan struct{}),
stateDirty: make(chan struct{}, 1),
reconcileNow: make(chan struct{}, 1),
reconcileDone: make(chan struct{}),
stateFile: stateFilePath(cfg.Supavisor.PidFile),
}
// Recorded without failing startup over it: an unidentifiable boot only
// means orphan reaping is skipped, which is the safe direction.
boot, err := bootID()
if err != nil {
s.logger.Warn("Cannot identify this boot, processes surviving a crash will not be reaped", "error", err)
}
s.bootID = boot
// Every configured program gets a process up front, running or not, so that
// the reconciler and the status command can see the whole set rather than
// only what has been started so far.
for name, progConfig := range cfg.Programs {
s.processes[name] = s.newProcess(progConfig)
s.desired[name] = DesiredStopped
if progConfig.Autostart {
s.desired[name] = DesiredRunning
}
}
return s, nil
}
// Start starts the supavisor
func (s *Server) Start() error {
if s.running {
return fmt.Errorf("supavisor is already running")
}
// Holding this for the lifetime of the daemon is what keeps a second
// instance from starting, and it is dropped by the kernel on a crash.
if err := s.lockPIDFile(); err != nil {
return err
}
// Before anything is started, so that an orphan reparented onto us is
// collected even while no program is running yet.
process.StartReaping(s.processLogger)
// Holding the lock means no other daemon is running, so anything still
// alive from a previous one is an orphan of a crash.
s.reapOrphans()
s.running = true
// Start IPC server
s.ipcServer = NewIPCServer(s.config.Supavisor.Socket, s.config.Supavisor.SocketGroup, s)
if err := s.ipcServer.Start(); err != nil {
s.releasePIDFile()
s.running = false
return fmt.Errorf("failed to start IPC server: %w", err)
}
// Setup signal handling
s.setupSignalHandling()
go s.recordStateChanges()
s.logger.Info("IPC server started", "socket", s.config.Supavisor.Socket)
s.logger.Info("Starting processes...")
go s.reconcileLoop()
s.logger.Info("Supavisor started successfully")
return nil
}
// Stop stops the supavisor and all processes
func (s *Server) Stop() error {
if !s.running {
return nil
}
s.logger.Info("Stopping supavisor...")
s.running = false
close(s.stopChan)
// Let the reconciler and anything it started settle first, or it would
// bring processes straight back up as they are stopped.
<-s.reconcileDone
s.actions.Wait()
s.stopAllProcesses()
// Stop IPC server
if s.ipcServer != nil {
s.logger.Info("Stopping IPC server")
if err := s.ipcServer.Stop(); err != nil {
s.logger.Error("failed to stop IPC server", "error", err)
}
}
// Everything was stopped deliberately, so there is nothing for the next
// daemon to reap.
if s.stateFile != "" {
s.clearStateFile()
}
s.releasePIDFile()
s.logger.Info("Supavisor daemon stopped")
return nil
}
// stopAllProcesses stops every process, working from the outermost dependents
// inwards and stopping each tier in parallel.
//
// Stopping serially cost stopwaitsecs for every process that does not exit on
// its stop signal, so a handful of them was enough to exceed systemd's
// TimeoutStopSec and have the daemon killed with its processes still running.
// Order matters as well: stopping a database before the programs using it is
// the wrong way round.
func (s *Server) stopAllProcesses() {
tiers, err := s.dependencyGraph.Tiers()
if err != nil {
s.logger.Warn("Failed to order shutdown, stopping everything at once", "error", err)
tiers = [][]string{s.programNames()}
}
for tier := len(tiers) - 1; tier >= 0; tier-- {
names := tiers[tier]
s.logger.Info("Stopping processes", "tier", tier, "count", len(names), "processes", names)
var wg sync.WaitGroup
for _, name := range names {
proc := s.process(name)
if proc == nil {
continue
}
wg.Go(func() {
if stopErr := proc.Stop(); stopErr != nil {
s.logger.Warn("failed to stop process", "process", name, "error", stopErr)
}
})
}
wg.Wait()
}
}
// StartProcess marks a process as wanted and reports whether it came up
func (s *Server) StartProcess(name string) error {
s.logger.Info("Start requested", "process", name)
proc := s.process(name)
if proc == nil {
return fmt.Errorf("process %s not found", name)
}
if proc.GetState() == process.StateRunning {
return fmt.Errorf("process %s is already running", name)
}
// A process that gave up, or exited on its own, has to be released back to
// STOPPED before the reconciler will consider starting it again.
if proc.GetState().IsStopped() {
if err := proc.Stop(); err != nil {
s.logger.Warn("failed to reset process before start", "process", name, "error", err)
}
}
if err := s.setDesired(name, DesiredRunning); err != nil {
return err
}
s.requestReconcile()
return s.awaitState(name, func(state process.State) bool {
// A one-off can finish its work inside startsecs and so never reach
// RUNNING. Waiting only for RUNNING would report a program that did
// exactly what it was asked to as a failure. The reset above put it in
// STOPPED, so the completion seen here can only be this run's.
return state == process.StateRunning ||
(state == process.StateExited && proc.HasCompleted())
})
}
// StopProcess marks a process as unwanted and waits for it to stop
func (s *Server) StopProcess(name string) error {
s.logger.Info("Stop requested", "process", name)
if err := s.setDesired(name, DesiredStopped); err != nil {
return err
}
s.requestReconcile()
return s.awaitState(name, func(state process.State) bool {
return state == process.StateStopped
})
}
// RestartProcess stops a process and starts it again
func (s *Server) RestartProcess(name string) error {
s.logger.Info("Restart requested", "process", name)
if err := s.StopProcess(name); err != nil {
return err
}
return s.StartProcess(name)
}
// GetStatus returns the status of all processes
func (s *Server) GetStatus() []ProcessStatusInfo {
s.processMutex.RLock()
defer s.processMutex.RUnlock()
statuses := make([]ProcessStatusInfo, 0, len(s.processes))
for name, proc := range s.processes {
state := proc.GetState()
pid := proc.GetPID()
exitCode := proc.GetExitCode()
restartCount := proc.GetRestartCount()
var uptime string
if state == process.StateRunning {
startTime := proc.GetStartTime()
duration := time.Since(startTime)
uptime = formatDuration(duration)
} else {
uptime = "N/A"
}
statuses = append(statuses, ProcessStatusInfo{
Name: name,
State: state,
Desired: s.desired[name],
Health: proc.GetHealth(),
Reason: s.notRunningReason(name, state, restartCount, exitCode),
PID: pid,
ExitCode: exitCode,
RestartCount: restartCount,
Uptime: uptime,
})
}
// Sort by process name alphabetically
sort.Slice(statuses, func(i, j int) bool {
return statuses[i].Name < statuses[j].Name
})
return statuses
}
// notRunningReason explains why a program is not running, or returns an empty
// string when there is nothing to explain.
//
// A program can be held back by a dependency, it can have given up, or it can
// have exited under a policy that leaves it alone. All three leave it not
// running while it may still be wanted, and the state alone does not say which.
// Must be called with processMutex held.
func (s *Server) notRunningReason(name string, state process.State, restartCount, exitCode int) string {
if reason := s.blockedReason[name]; reason != "" {
return reason
}
if state == process.StateFatal {
return fmt.Sprintf("gave up after %s", restartPhrase(restartCount))
}
if state == process.StateExited {
return exitedReason(s.autorestartPolicy(name), exitCode)
}
return ""
}
// reasonCleanExit is the whole explanation for a zero exit under the default
// policy: nothing went wrong, so there is nothing to restart.
const reasonCleanExit = "exited cleanly; autorestart is unexpected"
// exitedReason explains why a program that exited on its own was left that way.
// It names the policy that decided it, which is the setting to change.
func exitedReason(policy config.RestartPolicy, exitCode int) string {
switch {
case policy == config.RestartNever:
return fmt.Sprintf("exited with status %d; autorestart is never", exitCode)
case policy == config.RestartUnexpected && exitCode == 0:
return reasonCleanExit
}
// Every other combination would have been restarted, so getting here means
// the restart was abandoned rather than declined.
return fmt.Sprintf("exited with status %d", exitCode)
}
// autorestartPolicy returns the restart policy of a program. Must be called
// with processMutex held.
func (s *Server) autorestartPolicy(name string) config.RestartPolicy {
prog := s.config.Programs[name]
if prog == nil {
return ""
}
return prog.Autorestart
}
// restartPhrase renders a restart count as something that reads in a sentence
func restartPhrase(restarts int) string {
unit := "restarts"
if restarts == 1 {
unit = "restart"
}
return fmt.Sprintf("%d %s", restarts, unit)
}
// onProcessStateChange is called when a process state changes
func (s *Server) onProcessStateChange(name string, prevState, newState process.State) {
if prevState != newState {
s.logger.Info("Process state changed", "process", name, "prev_state", prevState, "new_state", newState)
}
s.markStateDirty()
// A dependency reaching RUNNING is what unblocks everything behind it, so
// reconcile now rather than waiting up to a tick for it.
s.requestReconcile()
}
// onProcessHealthChange is called when a program's health check result changes
func (s *Server) onProcessHealthChange(name string, prevHealth, health process.Health) {
s.logger.Info("Process health changed", "process", name, "prev_health", prevHealth, "new_health", health)
// A dependency becoming healthy unblocks whatever was waiting for it, in
// the same way reaching RUNNING does.
s.requestReconcile()
}
// setupSignalHandling installs the daemon's signal handlers
func (s *Server) setupSignalHandling() {
sigChan := make(chan os.Signal, signalBuffer)
signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM, syscall.SIGHUP)
go s.handleSignals(sigChan)
}
// handleSignals runs for the life of the daemon.
//
// It keeps reading rather than acting once and exiting, because installing a
// handler replaces the default disposition: a second SIGTERM sent to a
// shutdown that is taking too long would otherwise be swallowed, leaving
// SIGKILL as the only way out.
func (s *Server) handleSignals(sigChan chan os.Signal) {
shuttingDown := false
for sig := range sigChan {
switch sig {
case syscall.SIGHUP:
s.logger.Info("Received SIGHUP, reloading configuration")
_, err := s.Reload()
if err != nil {
s.logger.Error("failed to reload configuration", "error", err)
}
default:
if shuttingDown {
s.logger.Warn("Received a second stop signal, exiting immediately", "signal", sig.String())
os.Exit(exitInterrupted)
}
shuttingDown = true
s.logger.Info("Received signal to stop supavisor", "signal", sig.String())
go func() {
code := exitOK
if err := s.Stop(); err != nil {
s.logger.Error("failed to stop supavisor", "error", err)
code = exitFailed
}
os.Exit(code)
}()
}
}
}
// lockPIDFile takes the exclusive PID file lock for this daemon
func (s *Server) lockPIDFile() error {
if s.config.Supavisor.PidFile == "" {
s.logger.Warn("No pidfile configured: nothing prevents a second instance from starting")
return nil
}
lock, err := acquirePIDLock(s.config.Supavisor.PidFile)
if err != nil {
return err
}
s.pidLock = lock
return nil
}
// releasePIDFile drops the PID file lock and removes the file
func (s *Server) releasePIDFile() {
if s.pidLock == nil {
return
}
if err := s.pidLock.Release(); err != nil {
s.logger.Warn("failed to release PID file", "error", err)
}
s.pidLock = nil
}
// formatDuration formats a duration as a human-readable string
func formatDuration(d time.Duration) string {
if d < time.Minute {
return fmt.Sprintf("%ds", int(d.Seconds()))
} else if d < time.Hour {
return fmt.Sprintf("%dm %ds", int(d.Minutes()), int(d.Seconds())%60)
} else {
hours := int(d.Hours())
minutes := int(d.Minutes()) % 60
seconds := int(d.Seconds()) % 60
return fmt.Sprintf("%dh %dm %ds", hours, minutes, seconds)
}
}
package server
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"syscall"
"time"
"github.com/ademidoff/supavisor/internal/process"
)
const (
// orphanStopTimeout is how long an orphan gets to exit on SIGTERM before it
// is killed.
orphanStopTimeout = 5 * time.Second
// orphanPollInterval is how often orphans are checked for exit during that
// window.
orphanPollInterval = 100 * time.Millisecond
)
// processState is what a daemon records about the processes it owns, so that a
// daemon starting after a crash can recognize what outlived it.
type processState struct {
// BootID ties the records below to one boot. On Linux a process start time
// is measured in ticks since boot, so a recorded PID could otherwise match
// an unrelated process that happened to start at the same offset of a later
// boot, and supavisor would kill it.
BootID string `json:"boot_id"`
Children []childRecord `json:"children"`
}
// childRecord identifies one managed process well enough for a later daemon to
// decide whether the process now holding that PID is the one we started.
type childRecord struct {
Name string `json:"name"`
StartToken string `json:"start_token"`
PID int `json:"pid"`
}
// stateFilePath returns the sibling state file for a given pid file, so that
// /var/run/supavisor.pid is tracked by /var/run/supavisor.state.
func stateFilePath(pidFile string) string {
if pidFile == "" {
return ""
}
return strings.TrimSuffix(pidFile, filepath.Ext(pidFile)) + ".state"
}
// writeStateFile replaces the state file atomically, so a crash mid-write can
// never leave a half-written record behind for the next daemon to act on.
func writeStateFile(path, boot string, records []childRecord) error {
data, err := json.Marshal(processState{BootID: boot, Children: records})
if err != nil {
return fmt.Errorf("failed to encode process state: %w", err)
}
tmp := path + ".tmp"
if err := os.WriteFile(tmp, data, 0o600); err != nil {
return fmt.Errorf("failed to write process state: %w", err)
}
if err := os.Rename(tmp, path); err != nil {
_ = os.Remove(tmp)
return fmt.Errorf("failed to replace process state: %w", err)
}
return nil
}
// readStateFile returns what a previous daemon recorded
func readStateFile(path string) (*processState, error) {
data, err := os.ReadFile(path)
if err != nil {
if os.IsNotExist(err) {
return nil, nil
}
return nil, fmt.Errorf("failed to read process state: %w", err)
}
var state processState
if err := json.Unmarshal(data, &state); err != nil {
return nil, fmt.Errorf("failed to decode process state %s: %w", path, err)
}
return &state, nil
}
// saveState records the processes supavisor currently owns, so that a daemon
// starting after a crash can recognize what outlived it
func (s *Server) saveState() {
if s.stateFile == "" {
return
}
s.processMutex.RLock()
records := make([]childRecord, 0, len(s.processes))
for name, proc := range s.processes {
pid := proc.GetPID()
if pid <= 0 || proc.GetState().IsStopped() {
continue
}
token, err := processStartToken(pid)
if err != nil {
continue
}
records = append(records, childRecord{Name: name, PID: pid, StartToken: token})
}
s.processMutex.RUnlock()
if err := writeStateFile(s.stateFile, s.bootID, records); err != nil {
s.logger.Warn("failed to record process state", "error", err)
}
}
// markStateDirty asks for the state file to be refreshed.
//
// This is called from process state change callbacks, which run while the
// caller may hold processMutex for writing, so the write has to happen on
// another goroutine rather than inline.
func (s *Server) markStateDirty() {
select {
case s.stateDirty <- struct{}{}:
default:
}
}
// recordStateChanges rewrites the state file whenever processes change,
// coalescing bursts into a single write
func (s *Server) recordStateChanges() {
for {
select {
case <-s.stateDirty:
s.saveState()
case <-s.stopChan:
return
}
}
}
// reapOrphans stops processes left running by a previous daemon that did not
// shut down cleanly. Supavisor cannot adopt them: they are not its children, so
// it can neither wait on them nor learn how they exit.
func (s *Server) reapOrphans() {
if s.stateFile == "" {
return
}
state, err := readStateFile(s.stateFile)
if err != nil {
s.logger.Warn("failed to read previous process state", "error", err)
return
}
if state == nil || len(state.Children) == 0 {
return
}
// A PID recorded before a restart says nothing about what holds it now, and
// on Linux neither does the start time, since it is measured from boot.
// Killing on that basis could take out an unrelated process.
if s.bootID == "" || state.BootID != s.bootID {
s.logger.Info("Recorded processes belong to an earlier boot, leaving them alone",
"recorded_boot", state.BootID, "count", len(state.Children))
s.clearStateFile()
return
}
orphans := s.identifyOrphans(state.Children)
if len(orphans) == 0 {
s.clearStateFile()
return
}
for _, orphan := range orphans {
s.logger.Warn("Stopping a process left behind by a previous supavisor",
"process", orphan.Name, "pid", orphan.PID)
if err := process.SignalGroup(orphan.PID, syscall.SIGTERM); err != nil {
s.logger.Warn("failed to signal orphan", "process", orphan.Name, "pid", orphan.PID, "error", err)
}
}
s.waitForOrphans(orphans)
s.clearStateFile()
}
// identifyOrphans filters recorded processes down to those still running under
// the same identity
func (s *Server) identifyOrphans(records []childRecord) []childRecord {
orphans := make([]childRecord, 0, len(records))
for _, rec := range records {
token, err := processStartToken(rec.PID)
if err != nil {
// Either the process is gone, or this platform cannot prove
// identity, in which case killing by PID alone is not safe.
continue
}
if token != rec.StartToken {
s.logger.Info("Recorded pid now belongs to an unrelated process, leaving it alone",
"process", rec.Name, "pid", rec.PID)
continue
}
orphans = append(orphans, rec)
}
return orphans
}
// waitForOrphans gives orphans a chance to exit, then kills what is left
func (s *Server) waitForOrphans(orphans []childRecord) {
deadline := time.Now().Add(orphanStopTimeout)
for time.Now().Before(deadline) {
if !anyGroupAlive(orphans) {
s.logger.Info("Orphaned processes exited")
return
}
time.Sleep(orphanPollInterval)
}
for _, orphan := range orphans {
if err := process.SignalGroup(orphan.PID, syscall.Signal(0)); err != nil {
continue
}
s.logger.Warn("Orphan did not exit, killing it", "process", orphan.Name, "pid", orphan.PID)
if err := process.SignalGroup(orphan.PID, syscall.SIGKILL); err != nil {
s.logger.Warn("failed to kill orphan", "process", orphan.Name, "pid", orphan.PID, "error", err)
}
}
}
func anyGroupAlive(records []childRecord) bool {
for _, rec := range records {
if err := process.SignalGroup(rec.PID, syscall.Signal(0)); err == nil {
return true
}
}
return false
}
func (s *Server) clearStateFile() {
if err := os.Remove(s.stateFile); err != nil && !os.IsNotExist(err) {
s.logger.Warn("failed to remove process state file", "error", err)
}
}
// Package version reports the build identity of a supavisor binary.
package version
import "fmt"
// Overridden at link time with -ldflags -X by the release pipeline. The
// defaults are what a plain `go build` from a working tree reports.
var (
Version = "dev"
Commit = "none"
Date = "unknown"
)
// String renders the one-line version banner for the named binary
func String(name string) string {
return fmt.Sprintf("%s %s (%s, %s)", name, Version, Commit, Date)
}