mesh-catalog/modules/zfs/cmd/zfs-tools/machine.go

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"git.novox.be/novox/mesh-sdk/go/views"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"time"
)

// The machine as the module sees it.
//
// **It runs fixed commands, never a shell.** Every command is a program and an argument list built here;
// nothing an argument says is ever read by a shell. The tools run as the operator account (novox/hq ADR
// 0175 §4). What ZFS lets any account read — `zfs list`, `zfs get`, `zpool list`, `zpool status`,
// `zpool iostat` — runs as that account. What ZFS keeps for root — `zpool events`, `zpool history` and
// `zfs snapshot` — goes through `sudo -n`, as the packet filter's, the service manager's and nfs-server's
// tools do: without a prompt, so a refusal is said rather than waited on.

// CommandTimeout bounds every command but a sample.
const CommandTimeout = 30 * time.Second

// KillGrace is how long a command that ran out of time is given to end on SIGTERM before its whole
// process group is killed.
var KillGrace = 5 * time.Second

// Ran is what one command did.
type Ran struct {
	Out  string
	Err  string
	Code int
	// TimedOut is a command that did not answer within its time and was ended: what it did is not known
	// from its answer, only from reading the machine again.
	TimedOut bool
}

// Runner runs one command; injected, so every tool is tested over recorded answers.
type Runner func(ctx context.Context, name string, args ...string) Ran

// ExecRunner runs a command on the machine, bounded by the context (CommandTimeout unless the caller set
// one).
//
// **A command that runs out of time is ended with everything it started.** It runs in a process group of
// its own; at the deadline the group is sent SIGTERM, and after KillGrace SIGKILL. Under sudo the command
// itself runs as root, which this account cannot signal: SIGTERM reaches sudo, which passes it on to the
// command, and the group's SIGKILL ends sudo. The output pipes are closed after KillGrace whatever still
// holds them, so the call always returns.
func ExecRunner(ctx context.Context, name string, args ...string) Ran {
	if _, has := ctx.Deadline(); !has {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, CommandTimeout)
		defer cancel()
	}
	cmd := exec.Command(name, args...)
	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	cmd.WaitDelay = KillGrace
	var stdout, stderr bytes.Buffer
	cmd.Stdout, cmd.Stderr = &stdout, &stderr
	if err := cmd.Start(); err != nil {
		return Ran{Code: -1, Err: err.Error()}
	}
	done := make(chan error, 1)
	go func() { done <- cmd.Wait() }()
	var err error
	timedOut := false
	select {
	case err = <-done:
	case <-ctx.Done():
		timedOut = true
		pgid := cmd.Process.Pid
		_ = syscall.Kill(-pgid, syscall.SIGTERM)
		select {
		case err = <-done:
		case <-time.After(KillGrace):
			_ = syscall.Kill(-pgid, syscall.SIGKILL)
			err = <-done
		}
		// Anything of the group still alive and signalable goes with it.
		_ = syscall.Kill(-pgid, syscall.SIGKILL)
	}
	r := Ran{Out: stdout.String(), Err: strings.TrimSpace(stderr.String()), TimedOut: timedOut}
	if err != nil {
		var exit *exec.ExitError
		if errors.As(err, &exit) && !timedOut {
			r.Code = exit.ExitCode()
		} else {
			r.Code = -1
			if r.Err == "" {
				r.Err = err.Error()
			}
		}
	}
	if timedOut {
		r.Code = -1
		r.Err = strings.TrimSpace(fmt.Sprintf("did not answer within its time and was ended. %s", r.Err))
	}
	return r
}

// Machine is what the module reads and the one thing it does.
type Machine struct {
	Run Runner
	// UID is this process's account: root runs a command as it is, any other account through sudo -n.
	UID int
	Now func() time.Time
	// ReadFile reads a kernel file (the ARC's counters).
	ReadFile func(path string) ([]byte, error)
	// WriteFile writes a file as root (the ARC's live parameters).
	WriteFile func(path string, data []byte) error
	// Lock takes an exclusive lock on a file: waiting for it, or answering at once whether it was got. The
	// release gives it up.
	Lock func(path string, wait bool) (release func(), got bool, err error)
	// Ledger is the ledger `snapshots` as the tools bundle reaches it; nil in a process, which cannot (A-20).
	Ledger LedgerStore
	// Call invokes a tool or a seat verb as this module (`seat:<seat>.<verb>@<node>`, `<module>.<tool>`); nil where
	// this process reaches none.
	Call func(key string, body any) (json.RawMessage, error)
	// Node is this machine's name, for a seat verb of its own.
	Node string
	// Await waits at most the bound for an event of the module whose kind is one of kinds, and answers it; false
	// when none came.
	Await func(kinds []string, bound time.Duration) (map[string]any, bool)
	// Findings writes the module's own state `findings` as the seat mesh-conditions' layout says (A-32).
	Findings views.FindingWriter
	// Warnings are what the apply saw that is no failure (a change the bus did not take, a parameter awaiting
	// a load).
	Warnings []string
	// Resolve follows a link to its kernel path (a declared device's name under /dev/disk/by-id).
	Resolve func(path string) (string, error)
	// Emit publishes an event as this module (the record of an act); nil where nothing can be published.
	Emit func(eventType string, body any) error
	// States is the module's states on the bus (§26); nil where this process reaches none.
	States StateStore
	// Exists says whether a path is there (a link counts); nil says no path is.
	Exists func(path string) bool
	// Log is the module's own events stream as the tools bundle reaches it; nil in a process.
	Log EventLog
	// Errors is what the health check last saw of each device's error counts; nil judges every count
	// as past.
	Errors *ErrorMemory
}

// Here is the machine this process runs on.
func Here() *Machine {
	return &Machine{Run: ExecRunner, UID: os.Getuid(), Now: time.Now, ReadFile: os.ReadFile, Resolve: filepath.EvalSymlinks, Node: os.Getenv("MESH_NODE"), WriteFile: writeLive, Lock: FileLock, Exists: func(p string) bool { _, err := os.Lstat(p); return err == nil }, Errors: LoadErrorMemory(ErrorsFile)}
}

// NotApplicable is the answer of a machine without ZFS: the module does not apply here.
type NotApplicable struct {
	Applicable bool   `json:"applicable"`
	Said       string `json:"said"`
}

// absent says why ZFS cannot be asked on this machine, from a command's failure; "" when the failure is
// something else.
func absent(r Ran) string {
	switch {
	case r.Code == -1 && (strings.Contains(r.Err, "executable file not found") || strings.Contains(r.Err, "no such file or directory")):
		return "ZFS is not installed on this machine (there is no zfs or zpool command), so this module does not apply here"
	case strings.Contains(r.Err, "/dev/zfs") || strings.Contains(r.Err, "ZFS modules cannot be auto-loaded") ||
		strings.Contains(r.Err, "modprobe zfs"):
		return "ZFS's kernel module is not loaded on this machine (" + firstLine(r.Err) + "), so there is no pool to read"
	}
	return ""
}

// errAbsent is the error a tool answers when ZFS is not here.
type errAbsent struct{ why string }

func (e errAbsent) Error() string { return e.why }

// errTimedOut is an act that did not answer within its time and was ended: whether it took effect is
// known only by reading the machine again.
type errTimedOut struct{ what string }

func (e errTimedOut) Error() string {
	return e.what + " did not answer within its time and was ended; whether it took effect is not known"
}

// ErrorMemory is what the health check last saw of each device's error counts, so a count that rises is
// told from one that was there before (a fault now, against a fault past).
type ErrorMemory struct {
	mu     sync.Mutex
	counts map[string][3]uint64
	rose   map[string]time.Time
	path   string
}

// NewErrorMemory remembers nothing yet: every count on the first look is past.
func NewErrorMemory() *ErrorMemory {
	return &ErrorMemory{counts: map[string][3]uint64{}, rose: map[string]time.Time{}}
}

// ErrorsFile is where the error counts last seen are kept, so a restart of the bundle does not take every
// count as past again.
const ErrorsFile = "/var/lib/zfs-tools/errors.json"

type keptErrors struct {
	Counts map[string][3]uint64 `json:"counts"`
	Rose   map[string]time.Time `json:"rose"`
}

// LoadErrorMemory is the memory kept in path; an empty one when there is none, and either way it is kept
// there from now on.
func LoadErrorMemory(path string) *ErrorMemory {
	mem := NewErrorMemory()
	mem.path = path
	if b, err := os.ReadFile(path); err == nil {
		var k keptErrors
		if json.Unmarshal(b, &k) == nil {
			if k.Counts != nil {
				mem.counts = k.Counts
			}
			if k.Rose != nil {
				mem.rose = k.Rose
			}
		}
	}
	return mem
}

// save keeps the memory, atomically; a failure only costs the next restart its memory.
func (mem *ErrorMemory) save() {
	if mem == nil || mem.path == "" {
		return
	}
	mem.mu.Lock()
	b, err := json.Marshal(keptErrors{Counts: mem.counts, Rose: mem.rose})
	mem.mu.Unlock()
	if err != nil {
		return
	}
	tmp := mem.path + ".new"
	if os.WriteFile(tmp, b, 0o644) == nil {
		_ = os.Rename(tmp, mem.path)
	}
}

// ask runs a read-only command as this account and answers its output, or why it could not.
func (m *Machine) ask(ctx context.Context, name string, args ...string) (string, error) {
	r := m.Run(ctx, name, args...)
	if r.Code != 0 {
		if why := absent(r); why != "" {
			return "", errAbsent{why}
		}
		return "", fmt.Errorf("%s %s: exit %d: %s", name, strings.Join(args, " "), r.Code, firstLine(orOut(r)))
	}
	return r.Out, nil
}

// Escalated is a command as it is run: as given when this process is root, else through sudo without a
// prompt.
func (m *Machine) Escalated(name string, args ...string) (string, []string) {
	if m.UID == 0 {
		return name, args
	}
	return "sudo", append([]string{"-n", name}, args...)
}

// AsRoot runs what only root may do, and names a refusal by how it failed.
func (m *Machine) AsRoot(ctx context.Context, name string, args ...string) (Ran, error) {
	program, argv := m.Escalated(name, args...)
	r := m.Run(ctx, program, argv...)
	if r.TimedOut {
		return r, errTimedOut{name + " " + strings.Join(args, " ")}
	}
	if r.Code != 0 {
		why := orOut(r)
		if why2 := absent(r); why2 != "" && program != "sudo" {
			return r, errAbsent{why2}
		}
		if program == "sudo" && (strings.Contains(why, "a password is required") || strings.Contains(why, "not allowed") ||
			strings.Contains(why, "may not run sudo")) {
			return r, fmt.Errorf("%s needs root, and sudo refused the operator account without a prompt: %s", name, firstLine(why))
		}
		if program == "sudo" && strings.Contains(why, "command not found") {
			return r, errAbsent{"ZFS is not installed on this machine (sudo found no " + name + " command), so this module does not apply here"}
		}
		return r, fmt.Errorf("%s %s: exit %d: %s", name, strings.Join(args, " "), r.Code, firstLine(why))
	}
	return r, nil
}

func orOut(r Ran) string {
	if strings.TrimSpace(r.Err) != "" {
		return r.Err
	}
	return strings.TrimSpace(r.Out)
}

func firstLine(s string) string {
	s = strings.TrimSpace(s)
	if i := strings.IndexByte(s, '\n'); i >= 0 {
		return s[:i]
	}
	return s
}

// bytesSaid says a size for a person: 1.2 TiB, 512 MiB, 900 B.
func bytesSaid(n uint64) string {
	units := []string{"B", "KiB", "MiB", "GiB", "TiB", "PiB"}
	f := float64(n)
	i := 0
	for f >= 1024 && i < len(units)-1 {
		f /= 1024
		i++
	}
	if i == 0 {
		return fmt.Sprintf("%d B", n)
	}
	return fmt.Sprintf("%.1f %s", f, units[i])
}

// round1 rounds to one decimal, for what a person reads.
func round1(f float64) float64 {
	if f < 0 {
		return -float64(int64(-f*10+0.5)) / 10
	}
	return float64(int64(f*10+0.5)) / 10
}