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

package main

import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"os"
	"regexp"
	"strconv"
	"strings"
	"syscall"
)

// The apply: who, when, and what it converges (design 42 zfs-owns-zfs §15, §18).
//
// `zfs-tools apply` runs as root, holds an exclusive lock on ApplyLock (a second start waits for it), reads
// the config file the node-engine wrote, and converges the node's ZFS id, the ARC's live parameters, the
// pool and dataset properties and the cache and log devices. It converges nothing while a scrub or resilver
// runs on the pool or before the pool is imported: it says `config.waiting` and ends, and the event follower
// starts it again on the event that ends the wait (§27). It never runs on a timer. It exits 0 on a refusal or
// a wait, and non-zero only when a command it ran failed (A-8), so the node-engine gates the module on a real
// failure.

// ApplyLock is the lock the apply, the scheduled run and the hand acts take.
const ApplyLock = "/run/zfs-tools/apply.lock"

// The files the apply reads and writes.
const (
	HostIDFile    = "/etc/hostid"
	ParametersDir = "/sys/module/zfs/parameters/"
	MemInfoFile   = "/proc/meminfo"
)

// ZdbConfigArgv reads a pool's configuration, where its label's hostid is.
func ZdbConfigArgv(pool string) []string { return []string{"-C", pool} }

// ZgenhostidArgv writes /etc/hostid with ZFS's own writer; the value is one the parse validated (§11).
func ZgenhostidArgv(value string) []string { return []string{"-f", value} }

// FileLock takes an exclusive lock on a file, waiting for it or answering at once.
func FileLock(path string, wait bool) (func(), bool, error) {
	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
	if err != nil {
		return nil, false, err
	}
	how := syscall.LOCK_EX
	if !wait {
		how |= syscall.LOCK_NB
	}
	for {
		err = syscall.Flock(int(f.Fd()), how)
		if err != syscall.EINTR {
			break
		}
	}
	if err != nil {
		f.Close()
		if err == syscall.EWOULDBLOCK {
			return nil, false, nil
		}
		return nil, false, err
	}
	return func() { f.Close() }, true, nil
}

func writeLive(path string, data []byte) error { return os.WriteFile(path, data, 0o644) }

// memMiB is the node's memory in MiB from /proc/meminfo, 0 when it cannot be read (A-29).
func (m *Machine) memMiB() uint64 {
	b, err := m.ReadFile(MemInfoFile)
	if err != nil {
		return 0
	}
	for _, l := range strings.Split(string(b), "\n") {
		if rest, ok := strings.CutPrefix(l, "MemTotal:"); ok {
			f := strings.Fields(rest)
			if len(f) > 0 {
				if kb, err := strconv.ParseUint(f[0], 10, 64); err == nil {
					return kb / 1024
				}
			}
		}
	}
	return 0
}

// Managed says whether the config value a state key belongs to (its words before the first dot) is set.
func (c *Config) Managed(key string) bool {
	k, _, _ := strings.Cut(key, ".")
	switch k {
	case "pool":
		return c.Pool != ""
	case "scrub-cadence":
		return c.ScrubCadence != ""
	case "scrub-day":
		return c.ScrubDay != ""
	case "scrub-time":
		return c.ScrubTime != ""
	case "arc-max":
		return c.ArcMax != nil
	case "arc-min":
		return c.ArcMin != nil
	case "module-options":
		return len(c.ModuleOptions) > 0
	case "hostid":
		return c.HostID != ""
	case "zed":
		return c.Zed != ""
	case "zed-handlers":
		return len(c.ZedHandlers) > 0
	case "trim-schedule":
		return c.TrimSchedule != ""
	case "trim-day":
		return c.TrimDay != ""
	case "trim-time":
		return c.TrimTime != ""
	case "cache-devices":
		return len(c.CacheDevices) > 0
	case "log-devices":
		return len(c.LogDevices) > 0
	case "log-mirror":
		return c.LogMirror != nil
	case "pool-properties":
		return len(c.PoolProperties) > 0
	case "dataset-properties":
		return len(c.DatasetProperties) > 0
	case "snapshots":
		return c.Snapshots != nil
	}
	return false
}

// scanBusy says what runs on the pool that holds the apply back, "" when nothing does.
func scanBusy(scans []Scan) string {
	for _, s := range scans {
		if (s.Kind == "scrub" || s.Kind == "resilver") && (s.State == "running" || s.State == "paused") {
			return "a " + s.Kind + " is running on the pool"
		}
	}
	return ""
}

// Apply converges the node to its config file and answers a line of what it did. Its error is a failed
// command, never a refusal or a wait.
func (m *Machine) Apply(ctx context.Context) (string, error) {
	release, _, err := m.Lock(ApplyLock, true)
	if err != nil {
		return "", fmt.Errorf("the apply's lock %s: %v", ApplyLock, err)
	}
	defer release()
	return m.applyLocked(ctx)
}

// applyLocked is the apply under a lock already held (the scheduled run's hand act takes the same lock).
func (m *Machine) applyLocked(ctx context.Context) (string, error) {
	text, err := m.ReadFile(ConfigFile)
	if err != nil {
		return "", fmt.Errorf("the config file %s cannot be read: %v", ConfigFile, err)
	}
	c, refusals := ParseConfig(text, m.memMiB())
	if len(refusals) > 0 {
		var failed []error
		for _, r := range refusals {
			if err := m.Refused(r); err != nil {
				m.failure(&failed, r.Key, err)
			}
		}
		return fmt.Sprintf("%d values refused; nothing converged", len(refusals)), errors.Join(failed...)
	}
	m.forget(c)
	var failed []error
	// The pool must be imported and idle; until then nothing converges.
	if c.Pool != "" {
		wait, err := m.poolWait(ctx, c.Pool)
		if err != nil {
			m.failure(&failed, "pool", err)
			return "the pool could not be read", errors.Join(failed...)
		}
		if wait != "" {
			if err := m.Waiting("pool", wait); err != nil {
				m.failure(&failed, "pool", err)
			}
			return "waiting for " + wait, errors.Join(failed...)
		}
		_ = m.NotManaged("pool")
	}
	if c.HostID != "" && c.Pool == "" {
		_ = m.Refused(Refusal{Key: "hostid", Value: c.HostID, Reason: "needs the pool set, to read its label"})
	} else if c.HostID != "" {
		m.ApplyHostID(ctx, c, &failed)
	}
	m.ApplyARC(c, &failed)
	if c.Pool != "" {
		if err := m.ApplyProperties(ctx, c); err != nil {
			failed = append(failed, err)
		}
		if err := m.ApplyDevices(ctx, c); err != nil {
			failed = append(failed, err)
		}
	}
	return "applied", errors.Join(failed...)
}

// forget deletes the keys of the state `applied` that are a refusal or a wait of a value, or a value now not
// managed: what is still wrong is said again by this apply, and healthy is a key that is not there.
func (m *Machine) forget(c *Config) {
	if m.States == nil {
		return
	}
	keys, err := m.States.Keys(StateApplied)
	if err != nil {
		return
	}
	for _, k := range keys {
		if !c.Managed(k) {
			_ = m.States.Delete(StateApplied, k)
			continue
		}
		raw, ok, err := m.States.Get(StateApplied, k)
		if err != nil || !ok {
			continue
		}
		if strings.Contains(string(raw), `"outcome":"`+OutcomeRefused+`"`) || strings.Contains(string(raw), `"outcome":"`+OutcomeFailed+`"`) {
			_ = m.States.Delete(StateApplied, k)
		}
	}
}

// poolWait is what the apply waits on, "" when the pool is imported and idle.
func (m *Machine) poolWait(ctx context.Context, pool string) (string, error) {
	pools, err := m.Pools(ctx)
	if err != nil {
		return "", err
	}
	found := false
	for _, p := range pools {
		found = found || p.Name == pool
	}
	if !found {
		return "the pool " + pool + " to be imported", nil
	}
	out, err := m.ask(ctx, "zpool", StatusArgv()...)
	if err != nil {
		return "", err
	}
	ps, err := ParseStatus(out)
	if err != nil {
		return "", err
	}
	for _, p := range ps {
		if p.Name == pool {
			if busy := scanBusy(p.Scans); busy != "" {
				return "the scrub or resilver of the pool " + pool + " to end (" + busy + ")", nil
			}
		}
	}
	return "", nil
}

var labelHostID = regexp.MustCompile(`(?m)^\s*hostid:\s*(\d+)`)

func nativeHostID(b []byte) (uint32, bool) {
	if len(b) != 4 {
		return 0, false
	}
	return binary.NativeEndian.Uint32(b), true
}

// ApplyHostID converges /etc/hostid to the set value, only when the pool's label is 0 or already that value
// (§15), and reads the file back.
func (m *Machine) ApplyHostID(ctx context.Context, c *Config, failed *[]error) {
	const key = "hostid"
	refuse := func(why string) {
		if err := m.Refused(Refusal{Key: key, Value: c.HostID, Reason: why}); err != nil {
			m.failure(failed, key, err)
		}
	}
	n, _ := strconv.ParseUint(c.HostID, 16, 32)
	want := uint32(n)
	current := func() string {
		b, err := m.ReadFile(HostIDFile)
		if err != nil {
			return "not there"
		}
		if v, ok := nativeHostID(b); ok {
			return fmt.Sprintf("%08x", v)
		}
		return "not four bytes"
	}
	r, err := m.AsRoot(ctx, "zdb", ZdbConfigArgv(c.Pool)...)
	if err != nil {
		refuse("the pool's label cannot be read: " + err.Error())
		return
	}
	lm := labelHostID.FindStringSubmatch(r.Out)
	if lm == nil {
		refuse("the pool's label cannot be read: zdb -C shows no hostid")
		return
	}
	label, _ := strconv.ParseUint(lm[1], 10, 32)
	if label != 0 && uint32(label) != want {
		refuse(fmt.Sprintf("the pool's import would need -f: its label holds the hostid %08x, not %s", label, c.HostID))
		return
	}
	before := current()
	if before != c.HostID {
		if _, err := m.AsRoot(ctx, "zgenhostid", ZgenhostidArgv(c.HostID)...); err != nil {
			m.failure(failed, key, err)
			return
		}
	}
	back := current()
	if back != c.HostID {
		refuse(fmt.Sprintf("%s holds %s after the write, not the set %s: another writer changed it", HostIDFile, back, c.HostID))
		return
	}
	if before == back {
		if err := m.keepAsSet(key, back); err != nil {
			m.failure(failed, key, err)
		}
		return
	}
	if err := m.Changed("file", key, before, c.HostID, back); err != nil {
		m.failure(failed, key, err)
	}
}

// ApplyARC writes arc-max and arc-min live and reads both back; a module option whose live value differs is
// a warning that it takes effect at the next load of the kernel module, never a failure.
func (m *Machine) ApplyARC(c *Config, failed *[]error) {
	params := []struct {
		key, file string
		mib       *uint64
	}{{"arc-max", "zfs_arc_max", c.ArcMax}, {"arc-min", "zfs_arc_min", c.ArcMin}}
	live := func(file string) (string, error) {
		b, err := m.ReadFile(ParametersDir + file)
		return strings.TrimSpace(string(b)), err
	}
	before := map[string]string{}
	for _, p := range params {
		if p.mib == nil {
			continue
		}
		v, err := live(p.file)
		if err != nil {
			m.failure(failed, p.key, fmt.Errorf("%s cannot be read: %v", ParametersDir+p.file, err))
			continue
		}
		before[p.key] = v
	}
	// arc-min may not pass arc-max while ZFS holds them, so a second pass settles an order that failed.
	for pass := 0; pass < 2; pass++ {
		for _, p := range params {
			if p.mib == nil || before[p.key] == "" {
				continue
			}
			want := strconv.FormatUint(*p.mib<<20, 10)
			if v, _ := live(p.file); v != want {
				if err := m.WriteFile(ParametersDir+p.file, []byte(want)); err != nil && pass == 1 {
					m.failure(failed, p.key, err)
				}
			}
		}
	}
	for _, p := range params {
		if p.mib == nil || before[p.key] == "" {
			continue
		}
		want := strconv.FormatUint(*p.mib<<20, 10)
		back, _ := live(p.file)
		var err error
		switch {
		case back != want:
			err = fmt.Errorf("%s reads %q after writing %q", ParametersDir+p.file, back, want)
			m.failure(failed, p.key, err)
		case before[p.key] == want:
			err = m.keepAsSet(p.key, back)
		default:
			err = m.Changed("file", p.key, before[p.key], want, back)
		}
		if err != nil && back == want {
			m.failure(failed, p.key, err)
		}
	}
	for _, o := range c.ModuleOptions {
		key := "module-options." + o.Name
		v, err := live(o.Name)
		switch {
		case err == nil && v == o.Value:
			if err := m.keepAsSet(key, v); err != nil {
				m.failure(failed, key, err)
			}
		default:
			if err == nil {
				m.Warnings = append(m.Warnings, fmt.Sprintf("%s: live %s, set %s: takes effect at the next load of the kernel module", o.Name, v, o.Value))
			} else {
				m.Warnings = append(m.Warnings, fmt.Sprintf("%s: not a live parameter here: takes effect at the next load of the kernel module", o.Name))
			}
			if err := m.keep(key, AppliedEntry{Before: v, After: o.Value, Outcome: OutcomeAwaitingLoad,
				Why: "takes effect at the next load of the kernel module"}); err != nil {
				m.failure(failed, key, err)
			}
		}
	}
}