johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package hostprocess

import (
	"cmp"
	"errors"
	"fmt"
	"strconv"
	"syscall"
	"time"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostsvc"
)

type OpKind string

const (
	OpStart OpKind = "start"
	OpStop  OpKind = "stop"
)

type Op struct {
	Kind OpKind
	Spec hostsvc.UnitSpec
	Unit string
}

type Observed struct {
	Object     v1alpha1.HostProcess
	Backend    hostsvc.Backend
	Unit       hostsvc.UnitStatus
	Inspect    error
	Check      *hostsvc.Observation
	Exec       string
	Prefix     string
	Deleting   bool
	Restarted  bool
	StartedKey string
}

func RestartedInPlace(o v1alpha1.HostProcess, st hostsvc.UnitStatus) bool {
	rt := o.Status.Runtime
	return rt != nil && rt.InvocationID != "" && o.Spec.RestartPolicy != "" && o.Spec.RestartPolicy != v1alpha1.HostProcessRestartNever &&
		(st.State == hostsvc.Activating || st.State == hostsvc.Active && st.MainPID > 0 && st.InvocationID != "" && st.InvocationID != rt.InvocationID)
}

type Result struct {
	Phase      v1alpha1.HostProcessPhase
	LostReason v1alpha1.HostProcessLostReason
	Message    string
	Ops        []Op
	Identify   bool
	Unknown    bool
}

var ErrNotOurs = errors.New("hostprocess: a supervisor places only the units under its own prefix")

func restart(p v1alpha1.HostProcessRestartPolicy) hostsvc.Restart {
	switch p {
	case v1alpha1.HostProcessRestartAlways:
		return hostsvc.RestartAlways
	case v1alpha1.HostProcessRestartOnFailure:
		return hostsvc.RestartOnFailure
	}
	return hostsvc.RestartNo
}

func signal(s string) syscall.Signal {
	switch s {
	case "SIGINT", "INT":
		return syscall.SIGINT
	case "SIGKILL", "KILL":
		return syscall.SIGKILL
	case "SIGQUIT", "QUIT":
		return syscall.SIGQUIT
	}
	return syscall.SIGTERM
}

const DefaultMemoryMax = 2 << 30

func UnitSpec(o v1alpha1.HostProcess, exec string) (hostsvc.UnitSpec, error) {
	s := o.Spec
	exec = cmp.Or(exec, s.Exec)
	if exec == "" {
		return hostsvc.UnitSpec{}, fmt.Errorf("hostprocess %s: neither an exec path nor a resolvable asset", o.Name)
	}
	desc := "HostProcess " + o.Name
	if d := s.Description; d != "" {
		desc = d
	}
	spec := hostsvc.UnitSpec{Name: s.Unit, Description: desc, Exec: append([]string{exec}, s.Argv...),
		WorkDir: s.WorkDir, Slice: s.Slice, Restart: restart(s.RestartPolicy), After: s.After, Wants: s.Wants,
		PrivateHostname: s.PrivateHostname, FDStore: int(s.FdStore), KillSignal: syscall.SIGTERM}
	for _, e := range s.Env {
		spec.Env = append(spec.Env, e.Name+"="+e.Value)
	}
	spec.EnvFiles = append([]string(nil), s.EnvFiles...)
	if s.Output != nil {
		spec.Output = hostsvc.Output{StdoutFile: s.Output.Stdout, StderrFile: s.Output.Stderr}
	}
	if s.Stop != nil {
		spec.KillSignal = signal(s.Stop.Signal)
		spec.StopTimeout = time.Duration(s.Stop.GraceSeconds) * time.Second
	}
	spec.Limits.MemoryMax = DefaultMemoryMax
	if s.Limits != nil {
		spec.Limits = hostsvc.Limits{MemoryMax: DefaultMemoryMax, CPUWeight: uint64(s.Limits.CpuWeight), CPUQuotaPercent: int(s.Limits.CpuQuotaPercent),
			TasksMax: uint64(s.Limits.TasksMax)}
		if s.Limits.MemoryMax != "" {
			n, err := strconv.ParseUint(s.Limits.MemoryMax, 10, 64)
			if err != nil {
				return hostsvc.UnitSpec{}, fmt.Errorf("hostprocess %s: limits.memoryMax %q is not a byte count", o.Name, s.Limits.MemoryMax)
			}
			spec.Limits.MemoryMax = n
		}
	}
	return spec, hostsvc.ValidateSpec(spec)
}

func lost(r hostsvc.LostReason) v1alpha1.HostProcessLostReason {
	switch r {
	case hostsvc.LostHostRebooted:
		return v1alpha1.HostProcessLostHostRebooted
	case hostsvc.LostUnitFailed, hostsvc.LostUnitExited:
		return v1alpha1.HostProcessLostUnitFailed
	case hostsvc.LostOOMKilled:
		return v1alpha1.HostProcessLostOOMKilled
	case hostsvc.LostIdentityMismatch:
		return v1alpha1.HostProcessLostIdentityMismatch
	}
	return v1alpha1.HostProcessLostUnitGone
}

func phaseOf(st hostsvc.UnitStatus) v1alpha1.HostProcessPhase {
	switch st.State {
	case hostsvc.Active:
		return v1alpha1.HostProcessPhaseRunning
	case hostsvc.Activating, hostsvc.Reloading:
		return v1alpha1.HostProcessPhaseStarting
	case hostsvc.Deactivating:
		return v1alpha1.HostProcessPhaseStopping
	case hostsvc.Failed:
		return v1alpha1.HostProcessPhaseFailed
	}
	return v1alpha1.HostProcessPhaseStopped
}

func Reconcile(o Observed) Result {
	obj := o.Object
	rt := obj.Status.Runtime
	if o.Inspect != nil {
		return Result{Phase: obj.Status.Phase, LostReason: obj.Status.LostReason, Unknown: true,
			Message: "the service manager is unreachable; Unknown, never Lost: " + o.Inspect.Error()}
	}
	if rt != nil && rt.Backend != "" && rt.Backend != string(o.Backend) {
		return Result{Phase: v1alpha1.HostProcessPhaseLost, LostReason: v1alpha1.HostProcessLostIdentityMismatch,
			Message: fmt.Sprintf("placed by the %s backend and this supervisor runs %s: never adopted across backends", rt.Backend, o.Backend)}
	}
	if obj.Spec.Placement == v1alpha1.HostProcessPlacementBootstrap {
		if o.Unit.State == hostsvc.NotFound || o.Unit.State == "" {
			return Result{Phase: v1alpha1.HostProcessPhaseLost, LostReason: v1alpha1.HostProcessLostUnitGone,
				Message: "the bootstrap unit is not loaded; its supervisor (systemd or devsup) owns it, this is a mirror"}
		}
		return Result{Phase: phaseOf(o.Unit), Identify: o.Unit.State == hostsvc.Active,
			Message: "a read-only mirror of a bootstrap-layer unit"}
	}
	if o.Prefix != "" && (len(obj.Spec.Unit) < len(o.Prefix) || obj.Spec.Unit[:len(o.Prefix)] != o.Prefix) {
		return Result{Phase: v1alpha1.HostProcessPhaseFailed, Message: fmt.Sprintf("%v: %s is not under %s", ErrNotOurs, obj.Spec.Unit, o.Prefix)}
	}
	if o.Deleting {
		if o.Unit.State == hostsvc.Active || o.Unit.State == hostsvc.Activating {
			return Result{Phase: v1alpha1.HostProcessPhaseStopping, Ops: []Op{{Kind: OpStop, Unit: obj.Spec.Unit}}}
		}
		return Result{Phase: v1alpha1.HostProcessPhaseStopped}
	}
	if rt != nil && rt.InvocationID != "" {
		if obj.Status.Phase == v1alpha1.HostProcessPhaseLost {
			return Result{Phase: v1alpha1.HostProcessPhaseLost, LostReason: obj.Status.LostReason, Message: obj.Status.Message}
		}
		if o.Restarted {
			message := fmt.Sprintf("restarted in place by its manager (restartPolicy %s, invocation %s, recorded %s)",
				obj.Spec.RestartPolicy, o.Unit.InvocationID, rt.InvocationID)
			if o.StartedKey != "" && o.StartedKey == obj.Spec.RestartKey {
				message = fmt.Sprintf("restarted in place for spec.restartKey %q (invocation %s, recorded %s)",
					obj.Spec.RestartKey, o.Unit.InvocationID, rt.InvocationID)
			}
			return Result{Phase: phaseOf(o.Unit), Identify: o.Unit.State == hostsvc.Active, Message: message}
		}
		if o.Check == nil {
			return Result{Phase: obj.Status.Phase, Unknown: true, Message: "the recorded identity was not checked"}
		}
		switch o.Check.Verdict {
		case hostsvc.VerdictUnknown:
			return Result{Phase: obj.Status.Phase, Unknown: true, Message: "Unknown: " + o.Check.Detail}
		case hostsvc.VerdictRunning:
			if obj.Spec.RestartKey != obj.Status.RestartKey {
				spec, err := UnitSpec(obj, o.Exec)
				if err != nil {
					return Result{Phase: v1alpha1.HostProcessPhaseFailed, Message: err.Error()}
				}
				return Result{Phase: v1alpha1.HostProcessPhaseStarting, Identify: true,
					Ops: []Op{{Kind: OpStop, Unit: obj.Spec.Unit}, {Kind: OpStart, Spec: spec, Unit: spec.Name}},
					Message: fmt.Sprintf("restarted in place: spec.restartKey is %q and the running process started under %q",
						obj.Spec.RestartKey, obj.Status.RestartKey)}
			}
			message := "adopted by unit name and InvocationID"
			if o.StartedKey != "" && o.StartedKey == obj.Spec.RestartKey {
				message = fmt.Sprintf("restarted in place for spec.restartKey %q (invocation %s)", obj.Spec.RestartKey, o.Check.Status.InvocationID)
			}
			return Result{Phase: v1alpha1.HostProcessPhaseRunning, Identify: o.Check.Status.InvocationID != rt.InvocationID,
				Message: message}
		}
		if o.Check.Reason == hostsvc.LostUnitExited && o.Check.Status.ExecMainStatus == 0 && o.Check.Status.ExecMainCode <= 1 {
			return Result{Phase: v1alpha1.HostProcessPhaseStopped, Message: "the process exited 0"}
		}
		r := Result{Phase: v1alpha1.HostProcessPhaseLost, LostReason: lost(o.Check.Reason), Message: o.Check.Detail}
		if o.Check.Reason == hostsvc.LostIdentityMismatch && !o.Check.Stopped && o.Unit.State == hostsvc.Active {
			r.Ops = []Op{{Kind: OpStop, Unit: obj.Spec.Unit}}
		}
		return r
	}
	switch o.Unit.State {
	case hostsvc.Active:
		return Result{Phase: v1alpha1.HostProcessPhaseRunning, Identify: true, Message: "an active unit of this name is adopted"}
	case hostsvc.Activating:
		return Result{Phase: v1alpha1.HostProcessPhaseStarting, Identify: true}
	}
	if obj.Spec.RestartPolicy == v1alpha1.HostProcessRestartNever && obj.Status.Phase != "" && obj.Status.Phase != v1alpha1.HostProcessPhasePending {
		if o.Unit.State == hostsvc.Failed {
			return Result{Phase: v1alpha1.HostProcessPhaseLost, LostReason: v1alpha1.HostProcessLostUnitFailed,
				Message: fmt.Sprintf("the unit failed (result %s) before it was identified; restartPolicy Never places it once", o.Unit.Result)}
		}
		return Result{Phase: v1alpha1.HostProcessPhaseStopped, Message: "the unit ended before it was identified; restartPolicy Never places it once"}
	}
	spec, err := UnitSpec(obj, o.Exec)
	if err != nil {
		return Result{Phase: v1alpha1.HostProcessPhaseFailed, Message: err.Error()}
	}
	return Result{Phase: v1alpha1.HostProcessPhaseStarting, Ops: []Op{{Kind: OpStart, Spec: spec, Unit: spec.Name}}, Identify: true}
}