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

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"time"
)

// The scrub schedule the module owns.
//
// The mesh writes ScrubConfig from the assignment's settings. The node-engine runs the module's
// scheduled process, `zfs-tools scrub-run`, from its own timer (ScrubTimer) every quarter of an hour, and
// the run decides from the configuration whether a scrub is due: the last scheduled time not after now
// (the slot) not yet served. Due, it starts the pool's scrub, or waits for one already running, as
// upstream's zfs-scrub@.service does; it never pauses or stops one, and nothing stops the unit to pause
// it (the node-engine writes no ExecStop). So no setting value ever reaches a unit file: the day and the
// time are read and checked here, and an invalid one is refused before anything runs.
//
// The mesh holds the distribution's own scrub timers for the pool stopped and disabled, so one schedule
// scrubs it; the node-engine gives them back running and enabled when the module goes (novox/hq ADR 0269).

// ScrubTimer and ScrubService are the node-engine's units for the scheduled run (the process's name).
const (
	ScrubTimer   = "zfs-mesh-scrub.timer"
	ScrubService = "zfs-mesh-scrub.service"
)

// ServedFile is where the scheduled run keeps the last slot it served, per pool.
const ServedFile = "/var/lib/zfs-tools/scrub-served.json"

// MissedAfter is how long after its slot a scrub that has not started is said to be missed: the run
// fires every quarter of an hour.
const MissedAfter = time.Hour

// ScrubConfig is the schedule as the mesh wrote it from the settings.
const ScrubConfig = "/etc/zfs-tools/scrub.conf"

// Margins after a cadence before a last scrub is said to be late: a weekly run that waited for a busy
// pool, a monthly run on the first given weekday of a month (28 or 35 days apart).
var cadenceAllows = map[string]time.Duration{
	"weekly":  9 * 24 * time.Hour,
	"monthly": 38 * 24 * time.Hour,
}

// Schedule is the pool's scrub schedule: what the settings say and what the service manager says of it.
type Schedule struct {
	Pool    string `json:"pool"`
	Cadence string `json:"cadence"`
	Day     string `json:"day"`
	Time    string `json:"time"`
	Timer   string `json:"timer"`
	Service string `json:"service"`
	// TimerState and Enabled are the timer's ActiveState and UnitFileState.
	TimerState string `json:"timer_state,omitempty"`
	Enabled    string `json:"enabled,omitempty"`
	// Next is the next scheduled scrub, from the settings; TimerNext when the timer next fires the run that
	// decides (every quarter of an hour).
	Next        *time.Time `json:"next_run,omitempty"`
	TimerNext   *time.Time `json:"timer_next,omitempty"`
	LastTrigger *time.Time `json:"last_triggered,omitempty"`
	// LastResult is the service's Result of its last run (success, exit-code, …), LastExit when it ended.
	LastResult   string     `json:"last_result,omitempty"`
	ServiceState string     `json:"service_state,omitempty"`
	LastExit     *time.Time `json:"last_run_ended,omitempty"`
	Note         string     `json:"note,omitempty"`
}

var (
	poolName = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_.:-]*$`)
	weekday  = regexp.MustCompile(`^(Mon|Tue|Wed|Thu|Fri|Sat|Sun)$`)
	clock    = regexp.MustCompile(`^([01][0-9]|2[0-3]):[0-5][0-9]$`)
)

// ParseScrubConfig reads the schedule: pool, cadence (weekly or monthly), day (Mon … Sun) and time (HH:MM).
func ParseScrubConfig(text string) (*Schedule, error) {
	v := map[string]string{}
	for _, l := range strings.Split(text, "\n") {
		l = strings.TrimSpace(l)
		if l == "" || strings.HasPrefix(l, "#") {
			continue
		}
		k, val, ok := strings.Cut(l, "=")
		if !ok {
			return nil, fmt.Errorf("%s: %q is not key=value", ScrubConfig, l)
		}
		v[strings.TrimSpace(k)] = strings.TrimSpace(val)
	}
	s := &Schedule{Pool: v["pool"], Cadence: v["cadence"], Day: v["day"], Time: v["time"]}
	switch {
	case !poolName.MatchString(s.Pool):
		return nil, fmt.Errorf("the scrub-pool setting %q is not a pool's name", s.Pool)
	case cadenceAllows[s.Cadence] == 0:
		return nil, fmt.Errorf("the scrub-cadence setting is %q; it is weekly or monthly", s.Cadence)
	case !weekday.MatchString(s.Day):
		return nil, fmt.Errorf("the scrub-day setting is %q; it is Mon, Tue, Wed, Thu, Fri, Sat or Sun", s.Day)
	case !clock.MatchString(s.Time):
		return nil, fmt.Errorf("the scrub-time setting is %q; it is HH:MM, 00:00 to 23:59", s.Time)
	}
	s.Timer, s.Service = ScrubTimer, ScrubService
	return s, nil
}

var weekdays = map[string]time.Weekday{"Sun": time.Sunday, "Mon": time.Monday, "Tue": time.Tuesday,
	"Wed": time.Wednesday, "Thu": time.Thursday, "Fri": time.Friday, "Sat": time.Saturday}

// at is the scheduled time on one day.
func (s *Schedule) at(day time.Time) time.Time {
	h, _ := strconv.Atoi(s.Time[:2])
	mi, _ := strconv.Atoi(s.Time[3:])
	return time.Date(day.Year(), day.Month(), day.Day(), h, mi, 0, 0, day.Location())
}

// scheduledOn says whether a scrub is scheduled on a day: its weekday, and for a monthly one the first
// such day of the month.
func (s *Schedule) scheduledOn(day time.Time) bool {
	return day.Weekday() == weekdays[s.Day] && (s.Cadence == "weekly" || day.Day() <= 7)
}

// Slot is the last scheduled time not after now.
func (s *Schedule) Slot(now time.Time) time.Time {
	day := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
	for i := 0; i < 45; i++ {
		if s.scheduledOn(day) && !s.at(day).After(now) {
			return s.at(day)
		}
		day = day.AddDate(0, 0, -1)
	}
	return time.Time{}
}

// NextSlot is the first scheduled time after now.
func (s *Schedule) NextSlot(now time.Time) time.Time {
	day := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
	for i := 0; i < 45; i++ {
		if s.scheduledOn(day) && s.at(day).After(now) {
			return s.at(day)
		}
		day = day.AddDate(0, 0, 1)
	}
	return time.Time{}
}

// WaitArgv waits for a pool's running scrub to end.
func WaitArgv(pool string) []string { return []string{"wait", "-t", "scrub", "--", pool} }

// WaitBound is how long the scheduled run waits for a scrub: past it the run ends and leaves the scrub
// running; a scrub never pauses for it.
const WaitBound = 7 * 24 * time.Hour

// readServed is the last slot served per pool; none when the file is not there or not readable.
func (m *Machine) readServed(path string) map[string]int64 {
	out := map[string]int64{}
	b, err := m.ReadFile(path)
	if err != nil {
		return out
	}
	_ = json.Unmarshal(b, &out)
	return out
}

// startedSince says whether status shows the pool's scrub started at or after a time.
func startedSince(p *PoolStatus, slot time.Time) bool {
	if p == nil {
		return false
	}
	sc := p.Scrub()
	return sc != nil && sc.Started != nil && !sc.Started.Before(slot)
}

// ScrubRun is the scheduled run: a due scrub started, or one running waited for. It answers what it did in
// a line, and fails on an invalid schedule, which it never acts on.
func (m *Machine) ScrubRun(ctx context.Context, servedPath string) (string, error) {
	raw, err := m.ReadFile(ScrubConfig)
	if err != nil {
		return "", fmt.Errorf("%s could not be read: %v", ScrubConfig, err)
	}
	s, err := ParseScrubConfig(string(raw))
	if err != nil {
		return "", fmt.Errorf("refused, nothing run: %v", err)
	}
	now := m.Now()
	slot := s.Slot(now)
	served := m.readServed(servedPath)
	if served[s.Pool] >= slot.Unix() {
		return fmt.Sprintf("the scrub of %s scheduled %s was served", s.Pool, slot.Format("2006-01-02 15:04")), nil
	}
	st, err := m.Status(ctx)
	if err != nil {
		return "", err
	}
	p := statusOf(st, s.Pool)
	if p == nil {
		return "", fmt.Errorf("the scheduled pool %s is not imported here", s.Pool)
	}
	mark := func() error {
		served[s.Pool] = slot.Unix()
		b, _ := json.Marshal(served)
		if err := os.MkdirAll(filepath.Dir(servedPath), 0o755); err != nil {
			return err
		}
		tmp := servedPath + ".new"
		if err := os.WriteFile(tmp, b, 0o644); err != nil {
			return err
		}
		return os.Rename(tmp, servedPath)
	}
	if startedSince(p, slot) && p.Scrub().State != "running" {
		return fmt.Sprintf("the scrub of %s scheduled %s ran already", s.Pool, slot.Format("2006-01-02 15:04")), mark()
	}
	for _, r := range p.Resilvers() {
		if r.State == "running" {
			return fmt.Sprintf("a resilver of %s runs; the scrub scheduled %s waits for the next run after it", s.Pool,
				slot.Format("2006-01-02 15:04")), nil
		}
	}
	started := false
	if sc := p.Scrub(); sc == nil || sc.State != "running" {
		_, err := m.AsRoot(ctx, "zpool", ScrubArgv("start", s.Pool)...)
		switch {
		case err == nil:
			started = true
		case strings.Contains(err.Error(), "currently scrubbing"):
			// One began since status was read: it serves the slot, and is waited for like any other.
		default:
			return "", fmt.Errorf("the scheduled scrub of %s could not start: %v", s.Pool, err)
		}
	}
	if err := mark(); err != nil {
		return "", fmt.Errorf("the scrub of %s runs, and the served slot could not be kept: %v", s.Pool, err)
	}
	wctx, cancel := context.WithTimeout(ctx, WaitBound)
	defer cancel()
	_, werr := m.AsRoot(wctx, "zpool", WaitArgv(s.Pool)...)
	what := "waited for the scrub of " + s.Pool + " already running"
	if started {
		what = "started the scrub of " + s.Pool + " scheduled " + slot.Format("2006-01-02 15:04") + ", and waited for it"
	}
	if werr != nil {
		return what + "; the wait ended without its end (" + werr.Error() + "), and the scrub goes on", nil
	}
	return what + " to finish", nil
}

// TimerShowArgv asks the service manager what a timer says: whether it runs and is enabled, when it
// fires next and when it last did, as Unix times.
func TimerShowArgv(unit string) []string {
	return []string{"show", "--timestamp=unix", "--property=ActiveState,UnitFileState,NextElapseUSecRealtime,LastTriggerUSec", "--", unit}
}

// ServiceShowArgv asks what the scrub service's last run was.
func ServiceShowArgv(unit string) []string {
	return []string{"show", "--timestamp=unix", "--property=ActiveState,Result,ExecMainStartTimestamp,ExecMainExitTimestamp", "--", unit}
}

// parseShow reads `systemctl show`'s key=value lines.
func parseShow(out string) map[string]string {
	v := map[string]string{}
	for _, l := range lines(out) {
		if k, val, ok := strings.Cut(l, "="); ok {
			v[k] = strings.TrimSpace(val)
		}
	}
	return v
}

// unixStamp reads `--timestamp=unix`: "@1791842400"; nil for "", "n/a" or 0.
func unixStamp(s string) *time.Time {
	s = strings.TrimPrefix(strings.TrimSpace(s), "@")
	n, err := strconv.ParseInt(s, 10, 64)
	if err != nil || n == 0 {
		return nil
	}
	t := time.Unix(n, 0)
	return &t
}

// ScheduleNow is the schedule with what the service manager says of it; nil when the module has no
// schedule on this machine (no configuration written).
func (m *Machine) ScheduleNow(ctx context.Context) (*Schedule, error) {
	raw, err := m.ReadFile(ScrubConfig)
	if err != nil {
		return nil, nil
	}
	s, err := ParseScrubConfig(string(raw))
	if err != nil {
		return nil, err
	}
	if out, err := m.ask(ctx, "systemctl", TimerShowArgv(s.Timer)...); err != nil {
		s.Note = "the service manager did not say what the timer is: " + err.Error()
	} else {
		v := parseShow(out)
		s.TimerState, s.Enabled = v["ActiveState"], v["UnitFileState"]
		s.TimerNext, s.LastTrigger = unixStamp(v["NextElapseUSecRealtime"]), unixStamp(v["LastTriggerUSec"])
	}
	next := s.NextSlot(m.Now())
	s.Next = &next
	if out, err := m.ask(ctx, "systemctl", ServiceShowArgv(s.Service)...); err == nil {
		v := parseShow(out)
		s.ServiceState, s.LastResult, s.LastExit = v["ActiveState"], v["Result"], unixStamp(v["ExecMainExitTimestamp"])
	}
	return s, nil
}

// scheduleFindings are the schedule's warnings: a timer not waiting, a last run that failed with no scrub
// finished since, a slot past by MissedAfter that no scrub served.
func scheduleFindings(s *Schedule, p *PoolStatus, now time.Time, served map[string]int64) []string {
	var warn []string
	var scrubEnded *time.Time
	if p != nil {
		scrubEnded = LastScrubEnded(*p)
	}
	if slot := s.Slot(now); now.Sub(slot) > MissedAfter && served[s.Pool] < slot.Unix() && !startedSince(p, slot) {
		warn = append(warn, fmt.Sprintf("The scheduled scrub of %s due %s has not started.", s.Pool, slot.Format("2006-01-02 15:04")))
	}
	if s.TimerState != "" && (s.TimerState != "active" || s.Enabled != "enabled") {
		warn = append(warn, fmt.Sprintf("The scrub schedule's timer %s is %s and %s, so no scheduled scrub of %s will start.",
			s.Timer, s.TimerState, s.Enabled, s.Pool))
	}
	if s.LastResult != "" && s.LastResult != "success" && s.ServiceState != "activating" &&
		(scrubEnded == nil || s.LastExit == nil || scrubEnded.Before(*s.LastExit)) {
		when := "at a time the service manager did not say"
		if s.LastExit != nil {
			when = s.LastExit.Local().Format("2006-01-02 15:04")
		}
		warn = append(warn, fmt.Sprintf("The last scheduled scrub of %s failed (%s), %s; its journal says why.", s.Pool, s.LastResult, when))
	}
	return warn
}

// ScrubEvent is the event each scrub act is published as (module.json's emits).
const ScrubEvent = "scrub.requested"

// ScrubArgv is each act's command: fixed words, the pool after `--`.
func ScrubArgv(act, pool string) []string {
	switch act {
	case "pause":
		return []string{"scrub", "-p", "--", pool}
	case "stop":
		return []string{"scrub", "-s", "--", pool}
	}
	return []string{"scrub", "--", pool}
}

// ScrubActAnswer is what zfs_scrub_start, zfs_scrub_pause and zfs_scrub_stop answer.
type ScrubActAnswer struct {
	Done     bool     `json:"done"`
	Act      string   `json:"act"`
	Pool     string   `json:"pool"`
	Why      string   `json:"why"`
	Ran      []string `json:"ran"`
	After    string   `json:"after"`
	Recorded bool     `json:"recorded"`
	Record   string   `json:"record"`
	Said     string   `json:"said"`
}

// ScrubAct starts, pauses or stops a pool's scrub, and answers what status says afterwards. A scrub reads
// every block and rewrites from redundancy only a block it finds damaged; it adds I/O load, and never
// changes what a file holds.
func (m *Machine) ScrubAct(ctx context.Context, act, pool, why string) (ScrubActAnswer, error) {
	a := ScrubActAnswer{Act: act, Pool: pool, Why: why}
	switch act {
	case "start", "pause", "stop":
	default:
		return a, fmt.Errorf("%q is not an act of the scrub: start, pause or stop", act)
	}
	switch {
	case why == "":
		return a, fmt.Errorf("say why: a scrub act is a person's, and the why is kept with it")
	case strings.ContainsAny(why, "\n\r") || len(why) > MaxWhy:
		return a, fmt.Errorf("say why in one line of at most %d characters", MaxWhy)
	}
	ps, err := m.Pools(ctx)
	if err != nil {
		return a, err
	}
	known := false
	for _, p := range ps {
		known = known || p.Name == pool
	}
	if !known {
		return a, fmt.Errorf("there is no pool %q on this machine; zfs_pools lists them", pool)
	}
	program, argv := m.Escalated("zpool", ScrubArgv(act, pool)...)
	a.Ran = append([]string{program}, argv...)
	_, runErr := m.AsRoot(ctx, "zpool", ScrubArgv(act, pool)...)
	_, timedOut := runErr.(errTimedOut)
	if st, err := m.Status(ctx); err == nil {
		if p := statusOf(st, pool); p != nil {
			a.After = p.ScansSaid()
		}
	}
	a.Done = runErr == nil
	switch {
	case timedOut:
		a.Said = fmt.Sprintf("zpool scrub did not answer within its time and was ended: outcome unknown; check zfs_scrub. Status now: %s.", a.After)
	case runErr != nil:
		a.Said = fmt.Sprintf("NOT DONE: %v. Status now: %s.", runErr, a.After)
	default:
		a.Said = fmt.Sprintf("Asked %s to %s its scrub. Status now: %s. Why: %s.", pool, act, a.After, strings.TrimSuffix(why, "."))
	}
	if m.Emit == nil {
		a.Record = "this process publishes no events"
	} else if err := m.Emit(ScrubEvent, map[string]any{"pool": pool, "act": act, "why": why, "done": a.Done,
		"after": a.After, "at": m.Now().UTC().Format(time.RFC3339)}); err != nil {
		a.Record = err.Error()
	} else {
		a.Recorded, a.Record = true, "published as the event "+ScrubEvent
	}
	if !a.Recorded {
		a.Said += " NOT RECORDED on the bus: " + a.Record + "; the pool's own history (zpool history) has the command."
	}
	if runErr != nil {
		return a, fmt.Errorf("%s", a.Said)
	}
	return a, nil
}