mesh-host/internal/liveness/liveness.go

// Package liveness is the node-engine judging whether what a module runs stays up (novox/hq ADR 0240
// rule 1, to-be 48 §1 and §4, Phase A).
//
// **A module's container that crash-looped behind every passing check** is what this exists for: the
// agent server restarted about a hundred times while the mesh read it applied and its tools served, and a
// person found it reading its log for something else (research 032). The release gate judged a module by
// what the mesh saw from outside; nothing looked at what the module runs.
//
// Every long-running resource — a container that stays up, a process the mesh runs that stays up, a
// service stated `running` — is judged on every look, with no declaration:
//
//   - **alive** when it is running and has not restarted twice within the settle window after its grace;
//   - **unhealthy** when it is not running after its grace (`down`), or restarted twice in the window
//     (`restarting`);
//   - **starting** inside its grace after a start — a new build, a recreate, a restart the engine or a
//     person made — in which a restart is not counted: churn that stops is not a crash loop (issue 058);
//   - **held** while an open maintenance window holds it still (ADR 0189), and judged from a fresh grace
//     when the window ends;
//   - **unknown** when nothing could be read.
//
// **The engine counts restarts itself and keeps the count**, in a file beside the node's state, across a
// recreate of the container and across its own restarts: the runtime's count is lost on every recreate
// and its event history is a minute long on a busy machine (research 032 §2). Neither is read for a
// verdict.
//
// **It reads; it never acts** (ADR 0240 rule 6). One read of every container's state per look, one of
// every unit's per service manager — tens of milliseconds on the busiest machine — and never an
// execution inside a container. Nothing here restarts, recreates or stops anything.
package liveness

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/novox/mesh-host/internal/declaration"
)

// The bounds of ADR 0240 rule 1 and to-be 48 §1. Grace is the default a declaration will override in
// Phase B; the settle window is the gate's bound.
const (
	DefaultGrace = 60 * time.Second
	SettleWindow = 10 * time.Minute
	// LookEvery is how often the engine looks. A crash loop is said within the grace, two restarts and a
	// look — well inside the gate's bound — and a look is one read of the runtime.
	LookEvery = 15 * time.Second
)

// The kinds of long-running resource.
const (
	KindContainer = "container"
	KindService   = "service"
	KindProcess   = "process"
)

// The states, as the report says them (mesh-host internal/link and the controller agree on the words).
const (
	Healthy   = "healthy"
	Unhealthy = "unhealthy"
	Starting  = "starting"
	Held      = "held"
	Unknown   = "unknown"

	ReasonDown       = "down"
	ReasonRestarting = "restarting"
)

// Resource is one long-running resource of a module, as the judge needs it.
type Resource struct {
	Module string `json:"module"`
	ID     string `json:"id"`
	Kind   string `json:"kind"`
	// Target is the container's name, or the unit.
	Target string `json:"target"`
	// Scope and User are a service's manager: "user" and the account for a unit in an account's own.
	Scope string `json:"scope,omitempty"`
	User  string `json:"user,omitempty"`
}

// LongRunning is every long-running resource a declaration asks this machine to run for a module: its
// containers that stay up, its services stated running and its processes that stay up. A step, anything
// on a schedule, a service whose lifecycle is the machine's, what the mesh declares in its own right (no
// module) and what an adopted machine holds as it was found are not judged here.
func LongRunning(d *declaration.Declaration, held map[string]bool) []Resource {
	var out []Resource
	if d == nil {
		return nil
	}
	for _, r := range d.Resources {
		module, ok := ModuleOf(r.Identity())
		if !ok || held[r.Identity()] {
			continue
		}
		switch v := r.(type) {
		case *declaration.Container:
			if v.RunOnce || v.Schedule != "" {
				continue
			}
			out = append(out, Resource{Module: module, ID: v.ID, Kind: KindContainer, Target: v.Name})
		case *declaration.Service:
			if v.State != "running" {
				continue
			}
			res := Resource{Module: module, ID: v.ID, Kind: KindService, Target: v.Unit}
			if v.UserScoped() {
				res.Scope, res.User = declaration.ScopeUser, v.User
			}
			out = append(out, res)
		case *declaration.Process:
			if v.RunOnce || v.Schedule != "" {
				continue
			}
			out = append(out, Resource{Module: module, ID: v.ID, Kind: KindProcess, Target: v.Name + ".service"})
		}
	}
	return out
}

// ModuleOf is the module a resource's id belongs to: everything before its last dot — a module's name may
// carry a dot, its resources' own ids never do (the apply's moduleOf, for the same reason). False for
// what the mesh declares in its own right.
func ModuleOf(id string) (string, bool) {
	if strings.HasPrefix(id, declaration.AdoptionPrefix) {
		return "", false
	}
	at := strings.LastIndex(id, ".")
	if at <= 0 {
		return "", false
	}
	return id[:at], true
}

// Observed is one resource as one read found it.
type Observed struct {
	// Found is false for a container that is not there and a unit the manager does not know.
	Found bool
	// Identity is what changes when the thing is made again: the container's id, the unit's invocation.
	Identity string
	Running  bool
	// Restarting is the runtime or the manager restarting it after it exited.
	Restarting bool
	// Restarts is the runtime's own count of the restarts it made — read only to see it move.
	Restarts int64
	// Started is when the runtime says the current run started, as it says it: a change with no restart
	// counted is a restart somebody made, which is a new start.
	Started string
}

// Runtime is what one look reads: every container's state in one read, and every unit's per manager.
// An error is "could not be read" — said as unknown, never as down.
type Runtime interface {
	Containers(ctx context.Context, names []string) (map[string]Observed, error)
	Units(ctx context.Context, scope, user string, units []string) (map[string]Observed, error)
}

// State is one resource's state as the judge says it.
type State struct {
	Resource
	State    string    `json:"state"`
	Reason   string    `json:"reason,omitempty"`
	Since    time.Time `json:"since"`
	Streak   int       `json:"streak,omitempty"`
	Restarts int       `json:"restarts,omitempty"`
}

// Statement is one look at every long-running resource: when, and each resource's state.
type Statement struct {
	At        time.Time
	Resources []State
}

// Healthy says every resource in it is healthy.
func (s Statement) Healthy() bool {
	for _, r := range s.Resources {
		if r.State != Healthy {
			return false
		}
	}
	return true
}

// kept is what the judge keeps about one resource, on disk, across its own restarts.
type kept struct {
	Resource
	// Seen says a read has found it at least once; Identity, RuntimeRestarts and RuntimeStarted are what
	// the last read found, to see a restart or a recreate by.
	Seen            bool   `json:"seen,omitempty"`
	Identity        string `json:"identity,omitempty"`
	RuntimeRestarts int64  `json:"runtime-restarts,omitempty"`
	RuntimeStarted  string `json:"runtime-started,omitempty"`
	// Started is when the current start began: its grace is counted from here.
	Started time.Time `json:"started"`
	// Counted is every restart counted after a grace, kept across recreates; Recent those inside the
	// settle window since the current start.
	Counted int         `json:"counted,omitempty"`
	Recent  []time.Time `json:"recent,omitempty"`
	WasHeld bool        `json:"held,omitempty"`

	State  string    `json:"state,omitempty"`
	Reason string    `json:"reason,omitempty"`
	Since  time.Time `json:"since"`
	Streak int       `json:"streak,omitempty"`
}

// file is the judge's file beside the node's state.
type file struct {
	Resources []Resource       `json:"resources"`
	Kept      map[string]*kept `json:"kept"`
}

// FileName is the judge's file, beside the node's state.
const FileName = "liveness.json"

// Judge is the one judge of liveness on a machine. Safe for the apply and the looking loop at once.
type Judge struct {
	path    string
	runtime Runtime
	// Now, Grace and Settle are the clock and the bounds; replaced in tests.
	Now    func() time.Time
	Grace  time.Duration
	Settle time.Duration
	// HeldNow is the containers an open maintenance window holds still, by runtime name. Nil holds none.
	HeldNow func(now time.Time) map[string]bool

	mu    sync.Mutex
	f     file
	dirty bool
	// said is the statement said last, to know a change by.
	said map[string]string
}

// Open is the judge whose file is at path, reading what it kept. A file that cannot be read is said and
// started afresh: a count lost is a crash loop judged from now, never a machine left unjudged.
func Open(path string, rt Runtime) (*Judge, error) {
	j := &Judge{path: path, runtime: rt, Now: time.Now, Grace: DefaultGrace, Settle: SettleWindow,
		f: file{Kept: map[string]*kept{}}, said: map[string]string{}}
	raw, err := os.ReadFile(path)
	switch {
	case os.IsNotExist(err):
		return j, nil
	case err != nil:
		return j, fmt.Errorf("the liveness kept at %s cannot be read, so restarts are counted from now: %w", path, err)
	}
	var f file
	if err := json.Unmarshal(raw, &f); err != nil {
		return j, fmt.Errorf("the liveness kept at %s cannot be read, so restarts are counted from now: %w", path, err)
	}
	if f.Kept == nil {
		f.Kept = map[string]*kept{}
	}
	j.f = f
	return j, nil
}

// Set is what this machine runs now, from the declaration the apply just applied. A resource no longer
// declared is forgotten; one newly declared is judged from its first look.
func (j *Judge) Set(resources []Resource) {
	j.mu.Lock()
	defer j.mu.Unlock()
	sorted := append([]Resource(nil), resources...)
	sort.Slice(sorted, func(a, b int) bool {
		if sorted[a].Module != sorted[b].Module {
			return sorted[a].Module < sorted[b].Module
		}
		return sorted[a].ID < sorted[b].ID
	})
	declared := map[string]bool{}
	for _, r := range sorted {
		declared[r.ID] = true
		if k, ok := j.f.Kept[r.ID]; ok && (k.Kind != r.Kind || k.Target != r.Target || k.Scope != r.Scope || k.User != r.User) {
			// The same id now names another thing: judged as a new one, its count kept.
			counted := k.Counted
			j.f.Kept[r.ID] = &kept{Resource: r, Counted: counted}
		}
	}
	for id := range j.f.Kept {
		if !declared[id] {
			delete(j.f.Kept, id)
		}
	}
	j.f.Resources = sorted
	j.dirty = true
}

// Look reads every long-running resource once, judges each, keeps what it counted, and answers the
// statement and whether any resource's state or reason changed since the last look.
func (j *Judge) Look(ctx context.Context) (Statement, bool) {
	j.mu.Lock()
	defer j.mu.Unlock()
	now := j.Now()
	var held map[string]bool
	if j.HeldNow != nil {
		held = j.HeldNow(now)
	}
	observed, unread := j.read(ctx)

	st := Statement{At: now}
	changed := false
	seen := map[string]bool{}
	for _, r := range j.f.Resources {
		k := j.f.Kept[r.ID]
		if k == nil {
			k = &kept{Resource: r}
			j.f.Kept[r.ID] = k
		}
		k.Resource = r
		why, blind := unread[groupOf(r)]
		o := observed[keyOf(r)]
		j.judge(k, o, now, r.Kind == KindContainer && held[r.Target], blind, why)
		seen[r.ID] = true
		st.Resources = append(st.Resources, State{Resource: r, State: k.State, Reason: k.Reason, Since: k.Since,
			Streak: k.Streak, Restarts: k.Counted})
		word := k.State + "/" + k.Reason
		if j.said[r.ID] != word {
			changed = true
			j.said[r.ID] = word
		}
	}
	for id := range j.said {
		if !seen[id] {
			delete(j.said, id)
			changed = true
		}
	}
	j.save()
	return st, changed
}

// judge is one resource's verdict on one look.
func (j *Judge) judge(k *kept, o Observed, now time.Time, held, blind bool, why string) {
	set := func(state, reason string) {
		if k.State != state || k.Reason != reason || k.Since.IsZero() {
			k.State, k.Reason, k.Since = state, reason, now
		}
		if state == Unhealthy {
			k.Streak++
		} else {
			k.Streak = 0
		}
		j.dirty = true
	}
	fresh := func(o Observed, started time.Time) {
		k.Seen, k.Identity, k.RuntimeRestarts, k.RuntimeStarted = o.Found, o.Identity, o.Restarts, o.Started
		k.Started, k.Recent = started, nil
	}

	// **Held is neither alive nor dead**, and the window ending is a start: judged from a fresh grace.
	if held {
		k.WasHeld = true
		set(Held, "")
		return
	}
	if k.WasHeld {
		k.WasHeld = false
		fresh(o, now)
	}
	if blind {
		set(Unknown, why)
		return
	}

	switch {
	case !k.Seen:
		// First sight: of a resource just applied, or of one running before this engine judged. Grace is
		// counted from when the runtime says it started, where it says, so an engine restarted beside
		// a long-running container does not call it starting for a minute.
		started := now
		if t, err := time.Parse(time.RFC3339Nano, o.Started); err == nil && t.Before(now) && t.Year() > 1 {
			started = t
		}
		switch {
		case o.Found:
			fresh(o, started)
		case k.Started.IsZero():
			// Not there at its first look: its grace runs from now, and it is down after it.
			k.Started = now
		}
	case o.Found && o.Identity != "" && o.Identity != k.Identity && o.Restarts <= k.RuntimeRestarts,
		o.Found && o.Identity == k.Identity && o.Restarts == k.RuntimeRestarts && o.Started != k.RuntimeStarted:
		// Made again — recreated by an apply, or restarted by somebody — and not by the runtime's policy:
		// a new start, judged from a fresh grace. The count is kept.
		fresh(o, now)
	case o.Found && o.Restarts > k.RuntimeRestarts:
		// The runtime restarted it after it exited. Counted only after its grace.
		delta := int(o.Restarts - k.RuntimeRestarts)
		k.Identity, k.RuntimeRestarts, k.RuntimeStarted = o.Identity, o.Restarts, o.Started
		if !now.Before(k.Started.Add(j.Grace)) {
			k.Counted += delta
			for i := 0; i < delta; i++ {
				k.Recent = append(k.Recent, now)
			}
		}
		j.dirty = true
	}
	// Restarts older than the settle window no longer say anything.
	recent := k.Recent[:0]
	for _, t := range k.Recent {
		if now.Sub(t) <= j.Settle {
			recent = append(recent, t)
		}
	}
	k.Recent = recent

	inGrace := now.Before(k.Started.Add(j.Grace))
	switch {
	case len(k.Recent) >= 2:
		set(Unhealthy, ReasonRestarting)
	case inGrace:
		set(Starting, "")
	case !o.Found || !o.Running:
		if o.Restarting {
			set(Unhealthy, ReasonRestarting)
		} else {
			set(Unhealthy, ReasonDown)
		}
	default:
		set(Healthy, "")
	}
}

// read is one read of everything: every container at once, every unit per manager. unread names each
// group that could not be read, with why.
func (j *Judge) read(ctx context.Context) (map[string]Observed, map[string]string) {
	observed := map[string]Observed{}
	unread := map[string]string{}
	groups := map[string][]Resource{}
	var order []string
	for _, r := range j.f.Resources {
		g := groupOf(r)
		if _, ok := groups[g]; !ok {
			order = append(order, g)
		}
		groups[g] = append(groups[g], r)
	}
	for _, g := range order {
		rs := groups[g]
		targets := make([]string, 0, len(rs))
		for _, r := range rs {
			targets = append(targets, r.Target)
		}
		var found map[string]Observed
		var err error
		if rs[0].Kind == KindContainer {
			found, err = j.runtime.Containers(ctx, targets)
		} else {
			found, err = j.runtime.Units(ctx, rs[0].Scope, rs[0].User, targets)
		}
		if err != nil {
			unread[g] = firstLine(err.Error())
			continue
		}
		for _, r := range rs {
			observed[keyOf(r)] = found[r.Target]
		}
	}
	return observed, unread
}

// groupOf is the one read a resource is in: the containers, or one service manager's units.
func groupOf(r Resource) string {
	if r.Kind == KindContainer {
		return KindContainer
	}
	return "units:" + r.Scope + ":" + r.User
}

func keyOf(r Resource) string { return groupOf(r) + "/" + r.Target }

// save keeps what was judged, when anything moved. Written whole and renamed, so a reader never sees
// half of it; a failure is said by the next Open, which counts from then.
func (j *Judge) save() {
	if !j.dirty || j.path == "" {
		return
	}
	raw, err := json.Marshal(j.f)
	if err != nil {
		return
	}
	if err := os.MkdirAll(filepath.Dir(j.path), 0o700); err != nil {
		return
	}
	tmp, err := os.CreateTemp(filepath.Dir(j.path), ".liveness-*.json")
	if err != nil {
		return
	}
	defer os.Remove(tmp.Name())
	if _, err := tmp.Write(raw); err != nil {
		tmp.Close()
		return
	}
	if err := tmp.Close(); err != nil {
		return
	}
	if os.Rename(tmp.Name(), j.path) == nil {
		j.dirty = false
	}
}

func firstLine(s string) string {
	line, _, _ := strings.Cut(strings.TrimSpace(s), "\n")
	return line
}