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

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
)

// The scheduled run (design 42 zfs-owns-zfs §20, §24).
//
// `zfs-tools run` is the module's one scheduled run, root's, fired by the node-engine's timer every quarter of
// an hour. Each run takes the apply's lock; held, it does nothing and says `run.skipped` {reason: lock-held}.
// It does nothing while a scrub or resilver runs or before the pool is imported, saying that reason.
// Otherwise it acts on the hand acts waiting in the state `schedule`, starts the scrub when due and returns
// at once (the scrub's end is the event follower's to hear, §27), and starts a trim when trim-schedule is due.

// The reasons of `run.skipped`.
const (
	SkipLockHeld        = "lock-held"
	SkipPoolNotImported = "pool-not-imported"
	SkipScanRunning     = "scrub-or-resilver-running"
)

// KeyPaused is the key of the state `schedule` that holds the schedule's snapshots (§24).
const KeyPaused = "paused"

// TrimArgv starts a trim of the pool.
func TrimArgv(pool string) []string { return []string{"trim", "--", pool} }

// handActs are the acts waiting in the state `schedule`, each under its own key (A-17), but the pause.
func (m *Machine) handActs() []string {
	if m.States == nil {
		return nil
	}
	keys, err := m.States.Keys(StateSchedule)
	if err != nil {
		return nil
	}
	var acts []string
	for _, k := range keys {
		if k != KeyPaused {
			acts = append(acts, k)
		}
	}
	sort.Strings(acts)
	return acts
}

func (m *Machine) skip(reason string) string {
	var failed []error
	if err := m.say(EventRunSkipped, map[string]any{"reason": reason}); err != nil {
		m.failure(&failed, "run.skipped", err)
	}
	return "skipped: " + reason
}

// RunScheduled is the scheduled run. It answers what it did in a line; its error is a command that failed.
func (m *Machine) RunScheduled(ctx context.Context, servedPath string) (string, error) {
	release, got, err := m.Lock(ApplyLock, false)
	if err != nil {
		return "", fmt.Errorf("the lock %s: %v", ApplyLock, err)
	}
	if !got {
		return m.skip(SkipLockHeld), nil
	}
	defer release()
	text, err := m.ReadFile(ConfigFile)
	if err != nil {
		return "", fmt.Errorf("%s could not be read: %v", ConfigFile, err)
	}
	c, refused := ParseConfig(text, m.memMiB())
	if len(refused) > 0 {
		return "refused, nothing run: " + refused[0].Key + " " + refused[0].Reason, nil
	}
	var said []string
	var failed []error
	acts := m.handActs()
	guard := ""
	var p *PoolStatus
	if c.Pool != "" {
		wait, perr := m.poolWait(ctx, c.Pool)
		switch {
		case perr != nil:
			return "", perr
		case strings.Contains(wait, "to be imported"):
			guard = SkipPoolNotImported
		case wait != "":
			guard = SkipScanRunning
		}
		if guard == "" {
			st, err := m.Status(ctx)
			if err != nil {
				return "", err
			}
			p = statusOf(st, c.Pool)
		}
	}
	for _, act := range acts {
		by := m.actBy(act)
		switch {
		case act == ActSnapshots && guard == "":
			ts, err := m.TakeScheduled(ctx, c, true, by)
			said = append(said, ts...)
			if err != nil {
				failed = append(failed, err)
			}
		case act == "apply":
			s, err := m.applyLocked(ctx)
			said = append(said, "apply: "+s)
			if err != nil {
				failed = append(failed, err)
			}
		case act == ActPrune && guard == "":
			ts, err := m.PruneScheduled(ctx, c, true, by)
			said = append(said, ts...)
			if err != nil {
				failed = append(failed, err)
			}
		case guard != "":
			said = append(said, m.skip(guard))
		default:
			m.Warnings = append(m.Warnings, "the hand act "+act+" is not one this run knows")
		}
		_ = m.States.Delete(StateSchedule, act)
	}
	if guard != "" || c.Pool == "" {
		if guard != "" && len(acts) == 0 {
			said = append(said, m.skip(guard))
		}
		return strings.Join(said, "; "), joinErrors(failed)
	}
	served := m.readServed(servedPath)
	now := m.Now()
	for _, work := range []struct {
		name, key, verb string
		s               *Schedule
		start           func() error
	}{{"scrub", c.Pool, "start", ScheduleOf(c), func() error {
		_, err := m.AsRoot(ctx, "zpool", ScrubArgv("start", c.Pool)...)
		return err
	}}, {"trim", c.Pool + "/trim", "start", TrimScheduleOf(c), func() error {
		_, err := m.AsRoot(ctx, "zpool", TrimArgv(c.Pool)...)
		return err
	}}} {
		if work.s == nil {
			continue
		}
		slot := work.s.Slot(now)
		if slot.IsZero() || served[work.key] >= slot.Unix() {
			continue
		}
		if work.name == "scrub" && startedSince(p, slot) {
			said = append(said, fmt.Sprintf("the scrub of %s scheduled %s ran already", c.Pool, slot.Format("2006-01-02 15:04")))
		} else if err := work.start(); err != nil && !strings.Contains(err.Error(), "currently") {
			failed = append(failed, fmt.Errorf("the scheduled %s of %s could not start: %v", work.name, c.Pool, err))
			continue
		} else {
			said = append(said, fmt.Sprintf("started the %s of %s scheduled %s", work.name, c.Pool, slot.Format("2006-01-02 15:04")))
		}
		served[work.key] = slot.Unix()
		if err := writeServed(servedPath, served); err != nil {
			failed = append(failed, fmt.Errorf("the %s of %s was started, and the served slot could not be kept: %v", work.name, c.Pool, err))
		}
	}
	ts, err := m.TakeScheduled(ctx, c, false, "")
	said = append(said, ts...)
	if err != nil {
		failed = append(failed, err)
	}
	ps, err := m.PruneScheduled(ctx, c, false, "")
	said = append(said, ps...)
	if err != nil {
		failed = append(failed, err)
	}
	if len(said) == 0 {
		said = append(said, "nothing due")
	}
	return strings.Join(said, "; "), joinErrors(failed)
}

func joinErrors(errs []error) error {
	if len(errs) == 0 {
		return nil
	}
	var parts []string
	for _, e := range errs {
		parts = append(parts, e.Error())
	}
	return fmt.Errorf("%s", strings.Join(parts, "; "))
}

// actBy is who a hand act was by, from its key.
func (m *Machine) actBy(act string) string {
	raw, ok, err := m.States.Get(StateSchedule, act)
	if err != nil || !ok {
		return ""
	}
	var v struct {
		By string `json:"by"`
	}
	_ = json.Unmarshal(raw, &v)
	return v.By
}