amadan

carlos / platform Public

instancectl: the root control service, peer-authenticated, with the five verbs and the probe loop (plan 2, task 5)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YZ1ZvWj7naf73GjTX8MBv5
Paul Campbell pushed by paul@keymail.dev 82caa6922f30901f41835fa38c5d636e62dbc278 parent acaa97c
4 files changed, +947 −0
  • internal/instancectl/peer_linux.go +157 −0
  • internal/instancectl/peer_other.go +15 −0
  • internal/instancectl/server.go +477 −0
  • internal/instancectl/server_test.go +298 −0
diff --git a/internal/instancectl/peer_linux.go b/internal/instancectl/peer_linux.go
new file mode 100644
index 000000000..c36f15059
--- /dev/null
+++ b/internal/instancectl/peer_linux.go
@@ -0,0 +1,157 @@
+//go:build linux
+
+package instancectl
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "os"
+ "os/exec"
+ "os/user"
+ "strconv"
+ "strings"
+ "sync"
+ "syscall"
+ "time"
+
+ "golang.org/x/sys/unix"
+)
+
+// peerCred reads SO_PEERCRED from a connected unix socket.
+func peerCred(conn net.Conn) (*unix.Ucred, error) {
+ uc, ok := conn.(*net.UnixConn)
+ if !ok {
+ return nil, errors.New("not a unix connection")
+ }
+ raw, err := uc.SyscallConn()
+ if err != nil {
+ return nil, err
+ }
+ var cred *unix.Ucred
+ var cerr error
+ if err := raw.Control(func(fd uintptr) {
+ cred, cerr = unix.GetsockoptUcred(int(fd), unix.SOL_SOCKET, unix.SO_PEERCRED)
+ }); err != nil {
+ return nil, err
+ }
+ return cred, cerr
+}
+
+// productionPeerOK requires the peer to be uid carlos AND the main process
+// of carlos-edge.service as systemd reports it (design §5.4): a flat tenant
+// is a child of that pid, never the pid itself, and a nested tenant cannot
+// reach the socket at all. MainPID is asked of systemctl and cached for one
+// second, because the agent restarts and the pid changes.
+func productionPeerOK() func(net.Conn) error {
+ u, err := user.Lookup("carlos")
+ if err != nil {
+ return func(net.Conn) error { return fmt.Errorf("no carlos user: %v", err) }
+ }
+ carlosUID, _ := strconv.Atoi(u.Uid)
+ var mu sync.Mutex
+ var mainPID int
+ var at time.Time
+ edgePID := func() (int, error) {
+ mu.Lock()
+ defer mu.Unlock()
+ if time.Since(at) < time.Second && mainPID > 0 {
+ return mainPID, nil
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer cancel()
+ out, err := exec.CommandContext(ctx, "systemctl", "show", "-p", "MainPID", "--value", edgeUnit).Output()
+ if err != nil {
+ return 0, err
+ }
+ pid, err := strconv.Atoi(strings.TrimSpace(string(out)))
+ if err != nil || pid <= 0 {
+ return 0, fmt.Errorf("carlos-edge.service has no MainPID")
+ }
+ mainPID, at = pid, time.Now()
+ return pid, nil
+ }
+ return func(conn net.Conn) error {
+ cred, err := peerCred(conn)
+ if err != nil {
+ return err
+ }
+ if int(cred.Uid) != carlosUID {
+ return fmt.Errorf("peer uid %d is not carlos", cred.Uid)
+ }
+ pid, err := edgePID()
+ if err != nil {
+ return err
+ }
+ if int(cred.Pid) != pid {
+ return fmt.Errorf("peer pid %d is not the edge's main pid %d", cred.Pid, pid)
+ }
+ return nil
+ }
+}
+
+// activatedListener returns the socket-activated listener when systemd
+// passed one (LISTEN_PID is us, LISTEN_FDS is 1, the fd is 3).
+func activatedListener() (net.Listener, bool, error) {
+ if os.Getenv("LISTEN_PID") != strconv.Itoa(os.Getpid()) {
+ return nil, false, nil
+ }
+ if os.Getenv("LISTEN_FDS") != "1" {
+ return nil, false, fmt.Errorf("LISTEN_FDS=%q, want 1", os.Getenv("LISTEN_FDS"))
+ }
+ f := os.NewFile(3, "control.sock")
+ defer f.Close()
+ ln, err := net.FileListener(f)
+ if err != nil {
+ return nil, false, err
+ }
+ return ln, true, nil
+}
+
+// bindControlSocket is the hand-run fallback: bind the path with the
+// ownership the socket unit would give it (carlos, 0600).
+func bindControlSocket(path string) (net.Listener, error) {
+ _ = os.Remove(path)
+ ln, err := net.Listen("unix", path)
+ if err != nil {
+ return nil, err
+ }
+ u, err := user.Lookup("carlos")
+ if err != nil {
+ ln.Close()
+ return nil, err
+ }
+ uid, _ := strconv.Atoi(u.Uid)
+ gid, _ := strconv.Atoi(u.Gid)
+ if err := os.Chown(path, uid, gid); err != nil {
+ ln.Close()
+ return nil, err
+ }
+ if err := os.Chmod(path, 0o600); err != nil {
+ ln.Close()
+ return nil, err
+ }
+ return ln, nil
+}
+
+// fixtureIsRootCarlos0640 checks the probe fixture tmpfiles stages.
+func fixtureIsRootCarlos0640() error {
+ fi, err := os.Stat("/etc/carlos/probe-secret")
+ if err != nil {
+ return fmt.Errorf("probe fixture absent: %v", err)
+ }
+ st, ok := fi.Sys().(*syscall.Stat_t)
+ if !ok {
+ return errors.New("probe fixture: no stat")
+ }
+ g, err := user.LookupGroup("carlos")
+ if err != nil {
+ return err
+ }
+ gid, _ := strconv.Atoi(g.Gid)
+ if st.Uid != 0 || int(st.Gid) != gid || fi.Mode().Perm() != 0o640 {
+ return fmt.Errorf("probe fixture is uid %d gid %d mode %o, want root:carlos 0640", st.Uid, st.Gid, fi.Mode().Perm())
+ }
+ return nil
+}
diff --git a/internal/instancectl/peer_other.go b/internal/instancectl/peer_other.go
new file mode 100644
index 000000000..30cc02d32
--- /dev/null
+++ b/internal/instancectl/peer_other.go
@@ -0,0 +1,15 @@
+//go:build !linux
+
+package instancectl
+
+import (
+ "errors"
+ "net"
+)
+
+func productionPeerOK() func(net.Conn) error {
+ return func(net.Conn) error { return errors.ErrUnsupported }
+}
+func activatedListener() (net.Listener, bool, error) { return nil, false, errors.ErrUnsupported }
+func bindControlSocket(string) (net.Listener, error) { return nil, errors.ErrUnsupported }
+func fixtureIsRootCarlos0640() error { return errors.ErrUnsupported }
diff --git a/internal/instancectl/server.go b/internal/instancectl/server.go
new file mode 100644
index 000000000..5995fc057
--- /dev/null
+++ b/internal/instancectl/server.go
@@ -0,0 +1,477 @@
+package instancectl
+
+import (
+ "bufio"
+ "context"
+ "encoding/json"
+ "errors"
+ "flag"
+ "fmt"
+ "log"
+ "net"
+ "net/http"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/cliflag"
+ "github.com/carlosframework/platform/internal/controlsock"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// Server is the root instance-control service (design §5). It does four
+// things and nothing else: starts, stops and sweeps the three templates'
+// instances, sets a host slice's memory tier before a start, runs the
+// readiness probe, and answers /ready. Every systemctl argument it passes
+// is built from a validated host by UnitName/SliceName; nothing from a
+// request reaches systemctl as a string.
+type Server struct {
+ // Systemctl runs one systemctl invocation and returns its combined
+ // output; production is exec.Command, tests record.
+ Systemctl func(ctx context.Context, args ...string) (string, error)
+ // PeerOK is asked once per accepted connection, before a byte is read.
+ // Production requires SO_PEERCRED uid carlos and pid == MainPID of
+ // carlos-edge.service (design §5.4). A refused connection is closed
+ // with no response.
+ PeerOK func(conn net.Conn) error
+ // CgroupRoot is /sys/fs/cgroup; tests point it at a fixture tree.
+ CgroupRoot string
+ // ProbeOut is where the probe unit's StandardOutput=truncate: lands.
+ ProbeOut string
+ // FixtureOK verifies /etc/carlos/probe-secret is root:carlos 0640
+ // before a probe verdict is accepted (design §8.5, codex R11-1).
+ FixtureOK func() error
+ // ProbeEvery is the re-run cadence (five minutes in production).
+ ProbeEvery time.Duration
+ Logf func(string, ...any)
+
+ // cgroupWaitBudget bounds a wait for an empty cgroup; tests shorten it.
+ cgroupWaitBudget time.Duration
+
+ mu sync.Mutex
+ report ProbeReport
+ probed bool
+}
+
+const (
+ probeUnit = "carlos-instance-probe.service"
+ edgeUnit = "carlos-edge.service"
+ parentSlices = "carlos.slice/carlos-instances.slice"
+)
+
+// Run is `carlos instance-control`.
+func Run(args []string) {
+ fs := flag.NewFlagSet("instance-control", flag.ExitOnError)
+ fs.Usage = cliflag.Usage(fs, "usage: carlos instance-control (socket-activated by carlos-instance-control.socket)")
+ _ = fs.Parse(args)
+ if os.Geteuid() != 0 {
+ log.Fatal("instance-control: this is the root service behind carlos-instance-control.socket; it is not for operators")
+ }
+ s := &Server{
+ Systemctl: func(ctx context.Context, args ...string) (string, error) {
+ out, err := exec.CommandContext(ctx, "systemctl", args...).CombinedOutput()
+ return string(out), err
+ },
+ PeerOK: productionPeerOK(),
+ CgroupRoot: "/sys/fs/cgroup",
+ ProbeOut: filepath.Join(controlsock.Dir, "probe.out"),
+ FixtureOK: fixtureIsRootCarlos0640,
+ ProbeEvery: 5 * time.Minute,
+ Logf: log.Printf,
+ }
+ ln, activated, err := activatedListener()
+ if err != nil {
+ log.Fatalf("instance-control: %v", err)
+ }
+ if !activated {
+ // Not socket-activated (a hand run): bind the path ourselves with
+ // the same ownership the socket unit gives it.
+ ln, err = bindControlSocket(controlsock.Path)
+ if err != nil {
+ log.Fatalf("instance-control: %v", err)
+ }
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ go s.probeLoop(ctx)
+ if err := s.Serve(ctx, ln); err != nil {
+ log.Fatalf("instance-control: %v", err)
+ }
+}
+
+// Serve accepts connections, asks PeerOK about each, and hands the accepted
+// ones to the HTTP server one at a time through a listener that only ever
+// yields vetted connections.
+func (s *Server) Serve(ctx context.Context, ln net.Listener) error {
+ vetted := &vettedListener{Listener: ln, ok: s.PeerOK, logf: s.logf}
+ srv := &http.Server{Handler: s.Handler(), ReadHeaderTimeout: 5 * time.Second}
+ go func() {
+ <-ctx.Done()
+ srv.Close()
+ }()
+ err := srv.Serve(vetted)
+ if errors.Is(err, http.ErrServerClosed) || ctx.Err() != nil {
+ return nil
+ }
+ return err
+}
+
+// vettedListener runs the peer check inside Accept so a refused peer is
+// closed before net/http ever reads from it.
+type vettedListener struct {
+ net.Listener
+ ok func(net.Conn) error
+ logf func(string, ...any)
+}
+
+func (v *vettedListener) Accept() (net.Conn, error) {
+ for {
+ c, err := v.Listener.Accept()
+ if err != nil {
+ return nil, err
+ }
+ if err := v.ok(c); err != nil {
+ v.logf("instance-control: refused connection: %v", err)
+ c.Close()
+ continue
+ }
+ return c, nil
+ }
+}
+
+func (s *Server) logf(format string, args ...any) {
+ if s.Logf != nil {
+ s.Logf(format, args...)
+ }
+}
+
+func (s *Server) Handler() http.Handler {
+ mux := http.NewServeMux()
+ mux.HandleFunc("POST /start", s.handleStart)
+ mux.HandleFunc("POST /seed", s.handleSeed)
+ mux.HandleFunc("POST /stop", s.handleStop)
+ mux.HandleFunc("POST /reset", s.handleReset)
+ mux.HandleFunc("POST /sweep", s.handleSweep)
+ mux.HandleFunc("GET /ready", s.handleReady)
+ return mux
+}
+
+// request is the validated form of a verb's query: nothing else reaches a
+// unit name.
+type request struct {
+ host string
+ role Role
+ tier cgroup.Limit
+}
+
+func parseRequest(r *http.Request, roles []Role, wantTier bool) (request, error) {
+ q := r.URL.Query()
+ host := q.Get("host")
+ if !provision.ValidHostShape(host) {
+ return request{}, fmt.Errorf("host %q is not a host", host)
+ }
+ req := request{host: host}
+ if len(roles) > 0 {
+ role, ok := ParseRole(q.Get("role"))
+ if !ok {
+ return request{}, fmt.Errorf("role %q is not run, seed or sidecar", q.Get("role"))
+ }
+ allowed := false
+ for _, r := range roles {
+ allowed = allowed || r == role
+ }
+ if !allowed {
+ return request{}, fmt.Errorf("role %s is not valid for this verb", role)
+ }
+ req.role = role
+ }
+ if wantTier {
+ n, err := strconv.ParseInt(q.Get("memory_max"), 10, 64)
+ if err != nil {
+ return request{}, fmt.Errorf("memory_max %q is not a number", q.Get("memory_max"))
+ }
+ switch n {
+ case cgroup.TierDefault.MemoryMax, cgroup.TierChildren.MemoryMax:
+ req.tier = cgroup.Limit{MemoryMax: n}
+ default:
+ return request{}, fmt.Errorf("memory_max %d is not a platform tier", n)
+ }
+ }
+ return req, nil
+}
+
+func (s *Server) setTier(ctx context.Context, host string, tier cgroup.Limit) error {
+ max := strconv.FormatInt(tier.MemoryMax, 10)
+ // Multiply before dividing: MemoryMax/10*9 truncates a step early and
+ // mis-rounds by up to a few bytes against the tier constants' expected
+ // MemoryHigh (536870912 -> 483183820, not 483183819).
+ high := strconv.FormatInt(tier.MemoryMax*9/10, 10)
+ _, err := s.Systemctl(ctx, "set-property", "--runtime", SliceName(host),
+ "MemoryMax="+max, "MemoryHigh="+high, "MemorySwapMax="+max)
+ return err
+}
+
+func writeJSON(w http.ResponseWriter, status int, v any) {
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(status)
+ _ = json.NewEncoder(w).Encode(v)
+}
+
+func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) {
+ req, err := parseRequest(r, []Role{RoleRun, RoleSidecar}, true)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ ctx := r.Context()
+ unit := UnitName(req.role, req.host)
+ if err := s.setTier(ctx, req.host, req.tier); err != nil {
+ http.Error(w, "set-property: "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ if out, err := s.Systemctl(ctx, "start", unit); err != nil {
+ http.Error(w, "start: "+strings.TrimSpace(out+" "+err.Error()), http.StatusBadGateway)
+ return
+ }
+ inv, err := s.Systemctl(ctx, "show", "-p", "InvocationID", "--value", unit)
+ if err != nil {
+ http.Error(w, "show: "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ s.logf("instance-control: %s: started %s memory.max=%d", req.host, unit, req.tier.MemoryMax)
+ writeJSON(w, http.StatusOK, StartResponse{InvocationID: strings.TrimSpace(inv)})
+}
+
+// parseShow reads systemctl show's KEY=value lines.
+func parseShow(out string) map[string]string {
+ m := map[string]string{}
+ sc := bufio.NewScanner(strings.NewReader(out))
+ for sc.Scan() {
+ if k, v, ok := strings.Cut(sc.Text(), "="); ok {
+ m[k] = strings.TrimSpace(v)
+ }
+ }
+ return m
+}
+
+func (s *Server) handleSeed(w http.ResponseWriter, r *http.Request) {
+ req, err := parseRequest(r, nil, true)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ ctx := r.Context()
+ unit := UnitName(RoleSeed, req.host)
+ if err := s.setTier(ctx, req.host, req.tier); err != nil {
+ http.Error(w, "set-property: "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ // A oneshot's start job completes when ExecStart exits; RemainAfterExit
+ // keeps the unit loaded with its result until the stop below.
+ startOut, startErr := s.Systemctl(ctx, "start", unit)
+ out, err := s.Systemctl(ctx, "show", "-p", "InvocationID,Result,ExecMainStatus,ExecMainCode", unit)
+ if err != nil {
+ http.Error(w, "show: "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ m := parseShow(out)
+ wctx, cancel := context.WithTimeout(ctx, s.waitBudget())
+ defer cancel()
+ if err := WaitCgroupEmpty(wctx, s.CgroupRoot, SliceName(req.host), unit, 200*time.Millisecond); err != nil {
+ http.Error(w, "seed: "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ if _, err := s.Systemctl(ctx, "stop", unit); err != nil {
+ s.logf("instance-control: %s: stop after seed: %v", req.host, err)
+ }
+ if startErr != nil && m["Result"] == "" {
+ http.Error(w, "start: "+strings.TrimSpace(startOut+" "+startErr.Error()), http.StatusBadGateway)
+ return
+ }
+ writeJSON(w, http.StatusOK, SeedResponse{InvocationID: m["InvocationID"], Result: m["Result"], Status: m["ExecMainStatus"], Code: m["ExecMainCode"]})
+}
+
+func (s *Server) handleStop(w http.ResponseWriter, r *http.Request) {
+ req, err := parseRequest(r, []Role{RoleRun, RoleSeed, RoleSidecar}, false)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ if out, err := s.Systemctl(r.Context(), "stop", UnitName(req.role, req.host)); err != nil {
+ http.Error(w, "stop: "+strings.TrimSpace(out+" "+err.Error()), http.StatusBadGateway)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+}
+
+func (s *Server) handleReset(w http.ResponseWriter, r *http.Request) {
+ req, err := parseRequest(r, []Role{RoleRun, RoleSeed, RoleSidecar}, false)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ if out, err := s.Systemctl(r.Context(), "reset-failed", UnitName(req.role, req.host)); err != nil && !strings.Contains(out+err.Error(), "not loaded") {
+ http.Error(w, "reset-failed: "+strings.TrimSpace(out+" "+err.Error()), http.StatusBadGateway)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+}
+
+// handleSweep stops every instance of the three templates that is not
+// inactive or has a job, keyed on the names systemd reports, so a unit
+// whose identifier is hashed and whose row is gone is still stopped
+// (design §6.7, codex R7-1).
+func (s *Server) handleSweep(w http.ResponseWriter, r *http.Request) {
+ ctx := r.Context()
+ targets := map[string]bool{}
+ out, err := s.Systemctl(ctx, "list-units", "--all", "--plain", "--no-legend", "--no-pager",
+ UnitPrefixRun+"*.service", UnitPrefixSeed+"*.service", UnitPrefixSidecar+"*.service")
+ if err != nil {
+ http.Error(w, "list-units: "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ for _, line := range strings.Split(out, "\n") {
+ f := strings.Fields(line)
+ if len(f) < 3 {
+ continue
+ }
+ if _, _, ok := ParseUnitName(f[0]); !ok {
+ continue
+ }
+ if f[2] != "inactive" {
+ targets[f[0]] = true
+ }
+ }
+ jobs, err := s.Systemctl(ctx, "list-jobs", "--plain", "--no-legend", "--no-pager")
+ if err != nil {
+ http.Error(w, "list-jobs: "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ for _, line := range strings.Split(jobs, "\n") {
+ f := strings.Fields(line)
+ if len(f) < 2 {
+ continue
+ }
+ if _, _, ok := ParseUnitName(f[1]); ok {
+ targets[f[1]] = true
+ }
+ }
+ var stopped []string
+ for unit := range targets {
+ if _, err := s.Systemctl(ctx, "stop", unit); err != nil {
+ s.logf("instance-control: sweep: stop %s: %v", unit, err)
+ }
+ _, esc, _ := ParseUnitName(unit)
+ wctx, cancel := context.WithTimeout(ctx, s.waitBudget())
+ err := WaitCgroupEmpty(wctx, s.CgroupRoot, "carlos-instances-"+esc+".slice", unit, 200*time.Millisecond)
+ cancel()
+ if err != nil {
+ http.Error(w, "sweep: "+unit+": "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ stopped = append(stopped, unit)
+ }
+ writeJSON(w, http.StatusOK, SweepResponse{Units: stopped})
+}
+
+func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) {
+ rep, ready := s.Ready()
+ if ready {
+ w.WriteHeader(http.StatusNoContent)
+ return
+ }
+ writeJSON(w, http.StatusServiceUnavailable, rep)
+}
+
+// Ready returns the last probe verdict. Before the first probe, or when
+// the last one could not be read, the box is not ready.
+func (s *Server) Ready() (ProbeReport, bool) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.report, s.probed && s.report.Ready
+}
+
+func (s *Server) waitBudget() time.Duration {
+ if s.cgroupWaitBudget > 0 {
+ return s.cgroupWaitBudget
+ }
+ return 30 * time.Second
+}
+
+// WaitCgroupEmpty waits until the unit's cgroup under the host slice is
+// absent or reports "populated 0" in cgroup.events, polling every `every`,
+// bounded by ctx and the caller's budget (design §6.5, codex R2-5).
+func WaitCgroupEmpty(ctx context.Context, root, slice, unit string, every time.Duration) error {
+ events := filepath.Join(root, parentSlices, slice, unit, "cgroup.events")
+ for {
+ b, err := os.ReadFile(events)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ if err == nil && strings.Contains(string(b), "populated 0") {
+ return nil
+ }
+ select {
+ case <-ctx.Done():
+ return fmt.Errorf("processes remain in %s", unit)
+ case <-time.After(every):
+ }
+ }
+}
+
+// probeLoop runs the probe at start and every ProbeEvery.
+func (s *Server) probeLoop(ctx context.Context) {
+ s.RunProbe(ctx)
+ t := time.NewTicker(s.ProbeEvery)
+ defer t.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-t.C:
+ s.RunProbe(ctx)
+ }
+ }
+}
+
+// RunProbe starts the probe unit (a oneshot: start blocks until it exits),
+// reads the report it left, and stores the verdict. A fixture that is not
+// root:carlos 0640, a probe that could not be started, or a report that
+// could not be read all leave the box not ready, with the reason on a
+// verdict line; readiness is never the default.
+func (s *Server) RunProbe(ctx context.Context) {
+ set := func(r ProbeReport) {
+ s.mu.Lock()
+ s.report, s.probed = r, true
+ s.mu.Unlock()
+ }
+ if err := s.FixtureOK(); err != nil {
+ set(ProbeReport{Lines: []VerdictLine{{Check: "fixture", Verdict: "fail", Detail: err.Error()}}})
+ return
+ }
+ if out, err := s.Systemctl(ctx, "start", probeUnit); err != nil {
+ set(ProbeReport{Lines: []VerdictLine{{Check: "probe-unit", Verdict: "fail", Detail: strings.TrimSpace(out + " " + err.Error())}}})
+ return
+ }
+ b, err := os.ReadFile(s.ProbeOut)
+ if err != nil {
+ set(ProbeReport{Lines: []VerdictLine{{Check: "probe-output", Verdict: "fail", Detail: err.Error()}}})
+ return
+ }
+ var r ProbeReport
+ if err := json.Unmarshal(b, &r); err != nil {
+ set(ProbeReport{Lines: []VerdictLine{{Check: "probe-output", Verdict: "fail", Detail: err.Error()}}})
+ return
+ }
+ for _, l := range r.Lines {
+ if l.Verdict != "ok" {
+ s.logf("instance-control: probe: %s %s %s", l.Check, l.Verdict, l.Detail)
+ }
+ }
+ set(r)
+}
diff --git a/internal/instancectl/server_test.go b/internal/instancectl/server_test.go
new file mode 100644
index 000000000..d686645ff
--- /dev/null
+++ b/internal/instancectl/server_test.go
@@ -0,0 +1,298 @@
+package instancectl
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "net"
+ "net/http"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+)
+
+type fakeCtl struct {
+ mu sync.Mutex
+ calls []string
+ // answers maps a leading argument pattern ("show -p InvocationID") to output.
+ answers map[string]string
+ fail map[string]error
+}
+
+func (f *fakeCtl) run(_ context.Context, args ...string) (string, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ line := strings.Join(args, " ")
+ f.calls = append(f.calls, line)
+ for k, err := range f.fail {
+ if strings.HasPrefix(line, k) {
+ return "", err
+ }
+ }
+ for k, out := range f.answers {
+ if strings.HasPrefix(line, k) {
+ return out, nil
+ }
+ }
+ return "", nil
+}
+
+func (f *fakeCtl) joined() string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return strings.Join(f.calls, "\n")
+}
+
+func newServer(t *testing.T, ctl *fakeCtl) (*Server, *http.Client) {
+ t.Helper()
+ root := t.TempDir()
+ s := &Server{
+ Systemctl: ctl.run,
+ PeerOK: func(net.Conn) error { return nil },
+ CgroupRoot: filepath.Join(root, "cgroup"),
+ ProbeOut: filepath.Join(root, "probe.out"),
+ FixtureOK: func() error { return nil },
+ ProbeEvery: time.Hour,
+ Logf: t.Logf,
+ }
+ dir, err := os.MkdirTemp("", "ic")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ sock := filepath.Join(dir, "c.sock")
+ ln, err := net.Listen("unix", sock)
+ if err != nil {
+ t.Skipf("cannot bind a unix socket here (%v); run with TMPDIR=/tmp/g outside the sandbox", err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ t.Cleanup(cancel)
+ go s.Serve(ctx, ln) //nolint:errcheck
+ t.Cleanup(func() { ln.Close() })
+ client := &http.Client{Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "unix", sock)
+ }}}
+ return s, client
+}
+
+func post(t *testing.T, c *http.Client, path string) *http.Response {
+ t.Helper()
+ resp, err := c.Post("http://unix"+path, "", nil)
+ if err != nil {
+ t.Fatalf("POST %s: %v", path, err)
+ }
+ return resp
+}
+
+func TestStartSetsTheTierThenStartsThenReadsTheInvocation(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "abc123\n"}}
+ _, c := newServer(t, ctl)
+ resp := post(t, c, "/start?host=bill-2.example.com&role=run&memory_max=536870912")
+ if resp.StatusCode != 200 {
+ t.Fatalf("status %d", resp.StatusCode)
+ }
+ var out StartResponse
+ if err := json.NewDecoder(resp.Body).Decode(&out); err != nil || out.InvocationID != "abc123" {
+ t.Errorf("response = %+v, %v", out, err)
+ }
+ want := "set-property --runtime carlos-instances-bill\\x2d2.example.com.slice MemoryMax=536870912 MemoryHigh=483183820 MemorySwapMax=536870912\n" +
+ "start carlos-instance@bill\\x2d2.example.com.service\n" +
+ "show -p InvocationID --value carlos-instance@bill\\x2d2.example.com.service"
+ if got := ctl.joined(); got != want {
+ t.Errorf("systemctl calls:\n%s\nwant:\n%s", got, want)
+ }
+}
+
+func TestStartRefusesBadInputsWithoutCallingSystemctl(t *testing.T) {
+ ctl := &fakeCtl{}
+ _, c := newServer(t, ctl)
+ for _, q := range []string{
+ "host=Bad_Host&role=run&memory_max=536870912",
+ "host=h.example.com&role=exec&memory_max=536870912",
+ "host=h.example.com&role=run&memory_max=1",
+ "host=h.example.com&role=seed&memory_max=536870912", // seed is not a /start role
+ "role=run&memory_max=536870912",
+ } {
+ resp := post(t, c, "/start?"+q)
+ if resp.StatusCode != 400 {
+ t.Errorf("%s: status %d, want 400", q, resp.StatusCode)
+ }
+ }
+ if ctl.joined() != "" {
+ t.Errorf("systemctl was called: %s", ctl.joined())
+ }
+}
+
+func TestStartReportsSystemctlFailureAs502(t *testing.T) {
+ ctl := &fakeCtl{fail: map[string]error{"start ": errors.New("Job for carlos-instance@h.example.com.service failed")}}
+ _, c := newServer(t, ctl)
+ resp := post(t, c, "/start?host=h.example.com&role=sidecar&memory_max=1073741824")
+ if resp.StatusCode != 502 {
+ t.Errorf("status %d, want 502", resp.StatusCode)
+ }
+ if !strings.Contains(ctl.joined(), "MemoryMax=1073741824 MemoryHigh=966367641 MemorySwapMax=1073741824") {
+ t.Errorf("children tier not applied: %s", ctl.joined())
+ }
+}
+
+func TestSeedBlocksReadsTheRetainedResultWaitsAndStops(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ "show -p InvocationID,Result,ExecMainStatus,ExecMainCode": "InvocationID=feed\nResult=success\nExecMainStatus=0\nExecMainCode=exited\n",
+ }}
+ s, c := newServer(t, ctl)
+ // An empty cgroup: cgroup.events says populated 0.
+ ev := filepath.Join(s.CgroupRoot, "carlos.slice", "carlos-instances.slice", "carlos-instances-h.example.com.slice", "carlos-instance-seed@h.example.com.service", "cgroup.events")
+ if err := os.MkdirAll(filepath.Dir(ev), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(ev, []byte("populated 0\nfrozen 0\n"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ resp := post(t, c, "/seed?host=h.example.com&memory_max=536870912")
+ if resp.StatusCode != 200 {
+ t.Fatalf("status %d", resp.StatusCode)
+ }
+ var out SeedResponse
+ if err := json.NewDecoder(resp.Body).Decode(&out); err != nil || out.InvocationID != "feed" || out.Result != "success" || out.Status != "0" || out.Code != "exited" {
+ t.Errorf("response = %+v, %v", out, err)
+ }
+ calls := ctl.joined()
+ for _, want := range []string{"set-property --runtime carlos-instances-h.example.com.slice", "start carlos-instance-seed@h.example.com.service", "show -p InvocationID,Result,ExecMainStatus,ExecMainCode carlos-instance-seed@h.example.com.service", "stop carlos-instance-seed@h.example.com.service"} {
+ if !strings.Contains(calls, want) {
+ t.Errorf("missing %q in\n%s", want, calls)
+ }
+ }
+ if strings.Index(calls, "show -p") > strings.Index(calls, "stop ") {
+ t.Errorf("stop before show:\n%s", calls)
+ }
+}
+
+func TestSeedFailsWhenTheCgroupStaysPopulated(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID,Result": "InvocationID=feed\nResult=success\nExecMainStatus=0\nExecMainCode=exited\n"}}
+ s, c := newServer(t, ctl)
+ s.cgroupWaitBudget = 200 * time.Millisecond
+ ev := filepath.Join(s.CgroupRoot, "carlos.slice", "carlos-instances.slice", "carlos-instances-h.example.com.slice", "carlos-instance-seed@h.example.com.service", "cgroup.events")
+ if err := os.MkdirAll(filepath.Dir(ev), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(ev, []byte("populated 1\n"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ resp := post(t, c, "/seed?host=h.example.com&memory_max=536870912")
+ if resp.StatusCode != 502 {
+ t.Errorf("status %d, want 502 while processes remain", resp.StatusCode)
+ }
+}
+
+func TestStopResetAndTheirTolerances(t *testing.T) {
+ ctl := &fakeCtl{fail: map[string]error{"reset-failed carlos-instance-sidecar@": errors.New("Unit carlos-instance-sidecar@h.example.com.service not loaded.")}}
+ _, c := newServer(t, ctl)
+ if resp := post(t, c, "/stop?host=h.example.com&role=seed"); resp.StatusCode != 204 {
+ t.Errorf("stop status %d", resp.StatusCode)
+ }
+ if resp := post(t, c, "/reset?host=h.example.com&role=sidecar"); resp.StatusCode != 204 {
+ t.Errorf("reset of an unloaded unit: status %d, want 204 (tolerated)", resp.StatusCode)
+ }
+ if resp := post(t, c, "/reset?host=h.example.com&role=nope"); resp.StatusCode != 400 {
+ t.Errorf("bad role: status %d", resp.StatusCode)
+ }
+ if !strings.Contains(ctl.joined(), "stop carlos-instance-seed@h.example.com.service") {
+ t.Errorf("stop not issued: %s", ctl.joined())
+ }
+}
+
+func TestSweepStopsEveryInstanceUnitAndJob(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ "list-units": "carlos-instance@a.example.com.service loaded active running carlos exec-backed instance a.example.com\n" +
+ "carlos-instance-sidecar@a.example.com.service loaded inactive dead ...\n" +
+ "carlos-instance-seed@_aaaa\\x2d0123456789abcdef.service loaded activating start ...\n",
+ "list-jobs": "12 carlos-instance@b.example.com.service start waiting\n",
+ "show -p ActiveState --value": "inactive\n",
+ }}
+ s, c := newServer(t, ctl)
+ s.cgroupWaitBudget = time.Second
+ resp := post(t, c, "/sweep")
+ if resp.StatusCode != 200 {
+ t.Fatalf("status %d", resp.StatusCode)
+ }
+ var out SweepResponse
+ if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+ t.Fatal(err)
+ }
+ got := strings.Join(out.Units, " ")
+ for _, want := range []string{"carlos-instance@a.example.com.service", "carlos-instance-seed@_aaaa\\x2d0123456789abcdef.service", "carlos-instance@b.example.com.service"} {
+ if !strings.Contains(got, want) {
+ t.Errorf("sweep did not stop %s: %v", want, out.Units)
+ }
+ }
+ if strings.Contains(got, "sidecar@a") {
+ t.Errorf("sweep stopped an inactive unit with no job: %v", out.Units)
+ }
+ calls := ctl.joined()
+ if !strings.Contains(calls, "stop carlos-instance@a.example.com.service") || !strings.Contains(calls, "stop carlos-instance@b.example.com.service") {
+ t.Errorf("stops not issued:\n%s", calls)
+ }
+}
+
+func TestReadyReflectsTheProbeVerdict(t *testing.T) {
+ ctl := &fakeCtl{}
+ s, c := newServer(t, ctl)
+ // The fake "start carlos-instance-probe.service" writes the report the
+ // unit's StandardOutput=truncate: would.
+ report := ProbeReport{Ready: false, Lines: []VerdictLine{{Check: "pid1", Verdict: "fail", Detail: "pid=4242"}}}
+ ctl.answers = map[string]string{}
+ ctl.fail = map[string]error{}
+ writeReport := func(r ProbeReport) {
+ b, _ := json.Marshal(r)
+ if err := os.WriteFile(s.ProbeOut, b, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }
+ writeReport(report)
+ s.RunProbe(context.Background())
+ resp, err := c.Get("http://unix/ready")
+ if err != nil || resp.StatusCode != 503 {
+ t.Fatalf("not-ready: %v %v", resp, err)
+ }
+ var got ProbeReport
+ if err := json.NewDecoder(resp.Body).Decode(&got); err != nil || got.Ready || len(got.Lines) != 1 {
+ t.Errorf("body = %+v, %v", got, err)
+ }
+ writeReport(ProbeReport{Ready: true, Lines: []VerdictLine{{Check: "pid1", Verdict: "ok"}}})
+ s.RunProbe(context.Background())
+ if resp, err := c.Get("http://unix/ready"); err != nil || resp.StatusCode != 204 {
+ t.Fatalf("ready: %v %v", resp, err)
+ }
+ if !strings.Contains(ctl.joined(), "start carlos-instance-probe.service") {
+ t.Errorf("probe unit not started: %s", ctl.joined())
+ }
+ // A missing fixture is never a pass.
+ s.FixtureOK = func() error { return errors.New("probe fixture absent") }
+ s.RunProbe(context.Background())
+ if resp, _ := c.Get("http://unix/ready"); resp.StatusCode != 503 {
+ t.Errorf("fixture absent yet ready")
+ }
+}
+
+func TestPeerCheckClosesTheConnectionWithoutAResponse(t *testing.T) {
+ ctl := &fakeCtl{}
+ s, c := newServer(t, ctl)
+ s.PeerOK = func(net.Conn) error { return errors.New("not the agent") }
+ _, err := c.Get("http://unix/ready")
+ if err == nil {
+ t.Fatal("a refused peer got an HTTP response")
+ }
+ if ctl.joined() != "" {
+ t.Errorf("systemctl ran for a refused peer: %s", ctl.joined())
+ }
+}
+
+func TestWaitCgroupEmptyTreatsAnAbsentCgroupAsEmpty(t *testing.T) {
+ root := t.TempDir()
+ if err := WaitCgroupEmpty(context.Background(), root, "carlos-instances-h.slice", "carlos-instance@h.service", 10*time.Millisecond); err != nil {
+ t.Errorf("absent cgroup: %v", err)
+ }
+}