werkdock/internal/instance/supervise.go

package instance

import (
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"syscall"
	"time"

	"werkdock/internal/engine"
)

// StopTimeout is how long a forwarded signal gets before SIGKILL
// follows, docker stop's default.
const StopTimeout = 10 * time.Second

// SuperviseOptions is how the supervised command is wired to the world.
type SuperviseOptions struct {
	// Stdout and Stderr receive the command's output live (the terminal
	// of a foreground run); nil means nowhere.
	Stdout io.Writer
	Stderr io.Writer
	// Stdin is the command's input; nil means /dev/null.
	Stdin io.Reader
	// Record appends the output to the instance's stdout and stderr
	// files, for logs. A foreground --rm run has no reader for them
	// and skips the copy.
	Record bool
}

// Supervise runs the instance's command to completion under eng and
// keeps state.json current: running with the pids once started, exited
// with the code at the end, or exited with an Error if the start
// failed. The first SIGINT, SIGTERM, or SIGHUP the supervisor receives
// is forwarded to the command (which, as pid 1, sees it only if it
// handles it); a second one, or StopTimeout passing, kills it — docker
// stop's contract, applied to a foreground run as well. Returns the
// command's exit code. An --rm instance is removed afterwards.
func (m Manager) Supervise(inst *Instance, eng engine.Engine, opts SuperviseOptions) (int, error) {
	if inst.Running() {
		return 0, fmt.Errorf("instance %s is already running", inst.Config.Name)
	}
	stdout, stderr := opts.Stdout, opts.Stderr
	if opts.Record {
		outFile, err := os.OpenFile(inst.StdoutPath(), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
		if err != nil {
			return 0, err
		}
		defer outFile.Close()
		errFile, err := os.OpenFile(inst.StderrPath(), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
		if err != nil {
			return 0, err
		}
		defer errFile.Close()
		stdout, stderr = tee(stdout, outFile), tee(stderr, errFile)
	}
	if stdout == nil {
		stdout = io.Discard
	}
	if stderr == nil {
		stderr = io.Discard
	}
	b, ok := eng.(*engine.Bwrap)
	if ok {
		b.Stdout, b.Stderr, b.Stdin = stdout, stderr, opts.Stdin
	}
	if inst.Config.AutoRemove {
		defer m.remove(inst)
	}
	proc, err := eng.Start(inst.Config.Spec)
	if err != nil {
		inst.State = State{Status: StatusExited, ExitCode: ExitCodeLost, FinishedAt: time.Now(), Error: err.Error()}
		if saveErr := inst.SaveState(); saveErr != nil {
			return 0, saveErr
		}
		return 0, err
	}
	inst.State = State{Status: StatusRunning, PID: os.Getpid(), ChildPID: proc.ChildPID(), StartedAt: time.Now()}
	if err := inst.SaveState(); err != nil {
		_ = proc.Signal(syscall.SIGKILL)
		return 0, err
	}
	signals := make(chan os.Signal, 2)
	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
	defer signal.Stop(signals)
	done := make(chan struct{})
	go func() {
		var killAt <-chan time.Time
		for {
			select {
			case sig := <-signals:
				if killAt != nil {
					sig = syscall.SIGKILL
				} else {
					killAt = time.After(StopTimeout)
				}
				_ = proc.Signal(sig)
			case <-killAt:
				_ = proc.Signal(syscall.SIGKILL)
			case <-done:
				return
			}
		}
	}()
	code, err := proc.Wait()
	close(done)
	inst.State.Status = StatusExited
	inst.State.FinishedAt = time.Now()
	inst.State.ChildPID = 0
	if err != nil {
		inst.State.ExitCode = ExitCodeLost
		inst.State.Error = err.Error()
	} else {
		inst.State.ExitCode = code
	}
	if saveErr := inst.SaveState(); saveErr != nil && err == nil {
		err = saveErr
	}
	return code, err
}

func tee(live io.Writer, file io.Writer) io.Writer {
	if live == nil {
		return file
	}
	return io.MultiWriter(live, file)
}

// StartDetached launches the supervisor as a background process —
// `executable supervise ID`, in a session of its own, with the
// command's output going to the instance's files — and returns once
// it has recorded a state: running, or exited with the start error.
func (m Manager) StartDetached(inst *Instance, executable string) error {
	if inst.Running() {
		return fmt.Errorf("instance %s is already running", inst.Config.Name)
	}
	inst.State = State{Status: StatusCreated}
	if err := inst.SaveState(); err != nil {
		return err
	}
	// The supervisor's own messages (a bwrap that would not start) go
	// where logs will show them.
	errFile, err := os.OpenFile(inst.StderrPath(), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
	if err != nil {
		return err
	}
	defer errFile.Close()
	cmd := exec.Command(executable, "supervise", inst.Config.ID)
	cmd.Env = append(os.Environ(), "WERKDOCK_HOME="+m.Root)
	cmd.Stdin = nil
	cmd.Stdout = errFile
	cmd.Stderr = errFile
	cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
	if err := cmd.Start(); err != nil {
		return err
	}
	// Not our child to wait for; let it go.
	_ = cmd.Process.Release()
	deadline := time.Now().Add(30 * time.Second)
	for {
		if err := inst.Reload(); err != nil {
			return err
		}
		if inst.State.Status != StatusCreated {
			break
		}
		if time.Now().After(deadline) {
			return fmt.Errorf("instance %s: the supervisor did not report a start within 30s (see werkdock logs %s)", inst.Config.Name, inst.ShortID())
		}
		time.Sleep(50 * time.Millisecond)
	}
	if inst.State.Error != "" {
		return errors.New(inst.State.Error)
	}
	return nil
}

// Kill sends sig to the running command (docker kill). Not running is
// an error, as in docker.
func (m Manager) Kill(inst *Instance, sig syscall.Signal) error {
	if err := inst.Reload(); err != nil {
		return err
	}
	if !inst.Running() || !processAlive(inst.State.PID) {
		return fmt.Errorf("instance %s is not running", inst.Config.Name)
	}
	if err := syscall.Kill(inst.State.ChildPID, sig); err != nil && !errors.Is(err, syscall.ESRCH) {
		return fmt.Errorf("signalling instance %s: %w", inst.Config.Name, err)
	}
	return nil
}

// Stop is docker stop: SIGTERM, then SIGKILL after timeout. A stopped
// instance is left alone. Returns once the instance has exited.
func (m Manager) Stop(inst *Instance, timeout time.Duration) error {
	if err := inst.Reload(); err != nil {
		return err
	}
	if !inst.Running() {
		return nil
	}
	if err := m.Kill(inst, syscall.SIGTERM); err != nil {
		return err
	}
	if _, err := m.Wait(inst, timeout); err == nil {
		return nil
	}
	if err := m.Kill(inst, syscall.SIGKILL); err != nil {
		if inst.Running() {
			return err
		}
		return nil
	}
	_, err := m.Wait(inst, 10*time.Second)
	return err
}

// Wait blocks until the instance has exited and returns its exit code;
// timeout 0 waits forever. The supervisor is polled through
// state.json, which is what an API server on another host pid would
// do as well.
func (m Manager) Wait(inst *Instance, timeout time.Duration) (int, error) {
	deadline := time.Time{}
	if timeout > 0 {
		deadline = time.Now().Add(timeout)
	}
	for {
		reloaded, err := m.load(inst.Dir)
		if errors.Is(err, ErrNotFound) {
			// removed under us: an --rm instance that exited
			inst.State.Status = StatusExited
			return inst.State.ExitCode, nil
		}
		if err != nil {
			return 0, err
		}
		inst.State = reloaded.State
		if inst.State.Status == StatusExited {
			return inst.State.ExitCode, nil
		}
		if !deadline.IsZero() && time.Now().After(deadline) {
			return 0, fmt.Errorf("instance %s is still running after %s", inst.Config.Name, timeout)
		}
		time.Sleep(100 * time.Millisecond)
	}
}

// Logs copies the recorded stdout and stderr to the writers; with
// follow it keeps copying until the instance has exited and both files
// are drained, or until stop is closed (an API client that hung up;
// nil never stops).
func (m Manager) Logs(inst *Instance, stdout, stderr io.Writer, follow bool, stop <-chan struct{}) error {
	out, err := openOrEmpty(inst.StdoutPath())
	if err != nil {
		return err
	}
	defer out.Close()
	errs, err := openOrEmpty(inst.StderrPath())
	if err != nil {
		return err
	}
	defer errs.Close()
	for {
		n1, err := io.Copy(stdout, out)
		if err != nil {
			return err
		}
		n2, err := io.Copy(stderr, errs)
		if err != nil {
			return err
		}
		if !follow {
			return nil
		}
		select {
		case <-stop:
			return nil
		default:
		}
		if n1 == 0 && n2 == 0 {
			if err := inst.Reload(); err != nil {
				if os.IsNotExist(err) {
					return nil
				}
				return err
			}
			if !inst.Running() || !processAlive(inst.State.PID) {
				// drain what the last write left, then stop
				if _, err := io.Copy(stdout, out); err != nil {
					return err
				}
				_, err := io.Copy(stderr, errs)
				return err
			}
			select {
			case <-stop:
				return nil
			case <-time.After(200 * time.Millisecond):
			}
		}
	}
}

func openOrEmpty(path string) (*os.File, error) {
	f, err := os.Open(path)
	if os.IsNotExist(err) {
		return os.Open(os.DevNull)
	}
	return f, err
}

// InstancesDir is where instances live below root; exported for doctor
// and inspect output.
func InstancesDir(root string) string { return filepath.Join(root, "instances") }