mesh-catalog/modules/restic/cmd/restic-backups/data.go

// The data the modules on this machine declare, measured (novox/hq ADR 0233).
//
// The mesh composes a second file beside the backup lines: every data item every module on the
// machine declares, one line each,
//
//	item <id> <class> <path> <covered-by> <protection> <measure>
//
// where covered-by is the path whose snapshot keeps it (the item itself, or the directory its dump
// writes into), or `-` when it is not backed up; protection is backup, redundancy, both, or none; and
// measure is how the item is measured. This holder measures each item that is not a cache — its size,
// its newest write, and the redundant storage it is on and whether that is healthy (ZFS, md, btrfs) —
// and says it, with each module's last good backup and the precision of the measurement, in
// `backed-up`.
//
// **Never an unbounded walk of something large.** Three methods, the item's own choice:
//
//   - `walk` (the default, for small items): every file summed and the newest change found, in one walk
//     as root — at most once a day, and stopped after walkFor or walkFiles, said as "measured partially";
//   - `dataset`: the ZFS dataset holding the path — its `used` from the filesystem's own counters — and
//     the newest change among the path's top-level entries; hourly, and nothing is walked;
//   - `shallow`: the newest change among the top-level entries only, and no size; hourly. The controller keeps the readings and says when an item shrinks, stops being written,
//
// goes without a backup, or is replaced by an empty copy of itself; this holder only measures.
//
// And it deletes, when a person has decided: an item the mesh retired — its module gone from this
// machine — is removed by `backup_delete_retired`, and only after a last restore point of it has been
// taken, so the deletion can be undone until a person forgets that restore point.
package main

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

// Item is one data item a module declares.
type Item struct {
	// Measure is walk, dataset or shallow.
	Measure    string `json:"measure,omitempty"`
	Module     string `json:"-"`
	ID         string `json:"item"`
	Class      string `json:"class"`
	Path       string `json:"path"`
	CoveredBy  string `json:"covered_by,omitempty"`
	Protection string `json:"protection,omitempty"`
}

// Redundancy is the redundant storage an item is on, as read here.
type Redundancy struct {
	// Kind is zfs, md or btrfs.
	Kind string `json:"kind"`
	// Where is the pool, the array device or the filesystem.
	Where string `json:"where"`
	// Healthy is nil when its state could not be read.
	Healthy *bool  `json:"healthy"`
	Said    string `json:"said"`
}

// Measured is what one measurement found.
type Measured struct {
	SizeBytes  *int64     `json:"size_bytes,omitempty"`
	LastWrite  *time.Time `json:"last_write,omitempty"`
	MeasuredAt *time.Time `json:"measured_at,omitempty"`
	// Precision says what the size is: "exact", "dataset <name>: its whole size", "partial: …" (a lower
	// bound, never compared), or "no size: …".
	Precision  string      `json:"precision,omitempty"`
	Error      string      `json:"error,omitempty"`
	Redundancy *Redundancy `json:"redundancy,omitempty"`
}

// ItemReport is one item as `backed-up` says it.
type ItemReport struct {
	Item
	Measured
	LastBackup *time.Time `json:"last_backup,omitempty"`
}

// The classes the mesh declares; a cache is listed and never measured.
const classCache = "cache"

var safePath = regexp.MustCompile(`^/[^\s'"\\$` + "`" + `]*$`)

// parseItems reads the composed data file into each module's items. A line this holder does not read
// is refused, naming the module, as a backup line is.
func parseItems(text string) (map[string][]Item, error) {
	out := map[string][]Item{}
	module := ""
	for _, raw := range strings.Split(text, "\n") {
		line := strings.TrimSpace(raw)
		if line == "" {
			continue
		}
		if m := moduleHeader.FindStringSubmatch(line); m != nil {
			module = m[1]
			continue
		}
		if strings.HasPrefix(line, "#") || module == "" {
			continue
		}
		f := strings.Fields(line)
		if len(f) < 5 || len(f) > 7 || f[0] != "item" || !safePath.MatchString(f[3]) || (f[4] != "-" && !safePath.MatchString(f[4])) {
			return nil, fmt.Errorf("%s declares a data line this holder does not read: %s", module, line)
		}
		it := Item{Module: module, ID: f[1], Class: f[2], Path: f[3]}
		if len(f) >= 6 {
			it.Protection = f[5]
		}
		it.Measure = measureWalk
		if len(f) == 7 {
			it.Measure = f[6]
		}
		switch it.Measure {
		case measureWalk, measureDataset, measureShallow:
		default:
			return nil, fmt.Errorf("%s declares a data line this holder does not read: %s", module, line)
		}
		if f[4] != "-" {
			it.CoveredBy = f[4]
		}
		out[module] = append(out[module], it)
	}
	return out, nil
}

// Items is what the modules on this machine declare; none when the mesh composed no data file — an
// older controller, which composes none.
func (b *Backups) Items() (map[string][]Item, error) {
	if b.Where.Data == "" {
		return map[string][]Item{}, nil
	}
	raw, err := os.ReadFile(b.Where.Data)
	if errors.Is(err, os.ErrNotExist) {
		return map[string][]Item{}, nil
	}
	if err != nil {
		return nil, err
	}
	return parseItems(string(raw))
}

func (b *Backups) measuredFile() string { return filepath.Join(b.Where.State, "measured.json") }

// Measurements is the newest measurement of each item, by module and item.
func (b *Backups) Measurements() map[string]map[string]Measured {
	out := map[string]map[string]Measured{}
	if raw, err := os.ReadFile(b.measuredFile()); err == nil {
		_ = json.Unmarshal(raw, &out)
	}
	return out
}

var measuring sync.Mutex

// The ways an item is measured.
const (
	measureWalk    = "walk"
	measureDataset = "dataset"
	measureShallow = "shallow"
)

// The bounds on a walk, and how often one runs: a walk is for small items, and one that meets a large
// one stops and says so rather than loading the disks for hours.
var (
	walkFor   = 10 * time.Minute
	walkFiles = 2_000_000
	walkEvery = 23 * time.Hour
)

// walkCommand sums the files under a path and finds its newest change, in one walk, as root — bounded
// by time and by files. One line out: "<bytes> <epoch seconds> <files> <complete 1|0>".
func walkCommand(path string) string {
	return fmt.Sprintf("timeout %d find '%s' -xdev -printf '%%y %%s %%T@\\n' 2>/dev/null | head -n %d | "+
		"awk 'BEGIN{s=0;m=0;n=0} {n++; if ($1==\"f\") s+=$2; if ($3>m) m=$3} END {printf \"%%d %%.0f %%d\\n\", s, m, n}'; "+
		"echo \"${PIPESTATUS[0]:-0}\"", int(walkFor.Seconds()), path, walkFiles)
}

// topCommand is the newest change among a path and its top-level entries, and how many there are:
// one directory read, bounded. "<epoch seconds> <entries>".
func topCommand(path string) string {
	return "timeout 60 find '" + path + "' -maxdepth 1 -printf '%T@\\n' 2>/dev/null | " +
		"awk 'BEGIN{m=0;n=0} {n++; if ($1>m) m=$1} END {printf \"%.0f %d\\n\", m, n-1}'"
}

func epochTime(s string) *time.Time {
	epoch, err := strconv.ParseInt(s, 10, 64)
	if err != nil || epoch <= 0 {
		return nil
	}
	t := time.Unix(epoch, 0).UTC()
	return &t
}

// measure is one item, measured now, by its own method.
func (b *Backups) measure(ctx context.Context, it Item) Measured {
	at := b.Now().UTC()
	m := Measured{MeasuredAt: &at}
	if !b.exists(ctx, it.Path) {
		m.Error = it.Path + " does not exist"
		zero := int64(0)
		m.SizeBytes, m.Precision = &zero, "exact"
		return m
	}
	m.Redundancy = b.redundancyOf(ctx, it.Path)
	switch it.Measure {
	case measureDataset, measureShallow:
		out, err := b.Run(ctx, "bash", "-c", topCommand(it.Path))
		if f := strings.Fields(out); err == nil && len(f) == 2 {
			m.LastWrite = epochTime(f[0])
		} else if err != nil {
			m.Error = err.Error()
		}
		m.Precision = "no size: the newest change among its top-level entries only"
		if it.Measure == measureShallow {
			return m
		}
		out, err = b.Run(ctx, "zfs", "list", "-Hp", "-o", "name,used", it.Path)
		f := strings.Fields(out)
		if err != nil || len(f) != 2 {
			m.Precision = "no size: it is on no ZFS dataset to read one from; the newest change among its top-level entries only"
			return m
		}
		used, err := strconv.ParseInt(f[1], 10, 64)
		if err != nil {
			return m
		}
		m.SizeBytes = &used
		m.Precision = "dataset " + f[0] + ": its whole size, from the filesystem's counters; the newest change among the top-level entries"
		return m
	}
	out, err := b.Run(ctx, "bash", "-c", walkCommand(it.Path))
	if err != nil {
		m.Error = err.Error()
		return m
	}
	lines := strings.Split(strings.TrimSpace(out), "\n")
	f := strings.Fields(lines[0])
	if len(f) != 3 || len(lines) < 2 {
		m.Error = "the measurement said " + strings.TrimSpace(out)
		return m
	}
	size, err1 := strconv.ParseInt(f[0], 10, 64)
	files, err2 := strconv.Atoi(f[2])
	if err1 != nil || err2 != nil {
		m.Error = "the measurement said " + strings.TrimSpace(out)
		return m
	}
	m.SizeBytes, m.LastWrite = &size, epochTime(f[1])
	m.Precision = "exact"
	if status := strings.TrimSpace(lines[len(lines)-1]); status == "124" || files >= walkFiles {
		m.Precision = fmt.Sprintf("partial: measured partially — stopped after %s or %d files; the size is a lower bound, "+
			"and this item wants measure: dataset or shallow", walkFor, walkFiles)
	}
	return m
}

// redundancyOf is the redundant storage a path is on, read as root: a ZFS pool's health and its own
// verdict, an md array's members, a btrfs filesystem's error counters. Nil for storage with no
// redundancy this holder knows how to read — which, for an item said to be protected by redundancy, the
// controller says is no protection at all.
func (b *Backups) redundancyOf(ctx context.Context, path string) *Redundancy {
	out, err := b.Run(ctx, "findmnt", "-no", "SOURCE,FSTYPE", "--target", path)
	if err != nil {
		return nil
	}
	f := strings.Fields(out)
	if len(f) < 2 {
		return nil
	}
	source, fstype := f[0], f[1]
	yes, no := true, false
	switch {
	case fstype == "zfs":
		pool, _, _ := strings.Cut(source, "/")
		r := &Redundancy{Kind: "zfs", Where: pool}
		health, err := b.Run(ctx, "zpool", "list", "-H", "-o", "health", pool)
		if err != nil {
			r.Said = "zpool list: " + err.Error()
			return r
		}
		status, err := b.Run(ctx, "zpool", "status", "-x", pool)
		if err != nil {
			r.Said = "zpool status: " + err.Error()
			return r
		}
		health, status = strings.TrimSpace(health), strings.TrimSpace(status)
		r.Said = "health " + health + "; " + firstLineOf(status)
		if health == "ONLINE" && strings.Contains(status, "is healthy") {
			r.Healthy = &yes
		} else {
			r.Healthy = &no
			r.Said = "health " + health + "; " + oneLineOf(status)
		}
		return r
	case strings.HasPrefix(source, "/dev/md"):
		device := strings.TrimPrefix(source, "/dev/")
		r := &Redundancy{Kind: "md", Where: device}
		mdstat, err := b.Run(ctx, "cat", "/proc/mdstat")
		if err != nil {
			r.Said = err.Error()
			return r
		}
		healthy, said, found := mdHealth(mdstat, device)
		if !found {
			r.Said = device + " is not in /proc/mdstat"
			return r
		}
		r.Said = said
		if healthy {
			r.Healthy = &yes
		} else {
			r.Healthy = &no
		}
		return r
	case fstype == "btrfs":
		r := &Redundancy{Kind: "btrfs", Where: source}
		stats, err := b.Run(ctx, "btrfs", "device", "stats", "--check", path)
		if err != nil {
			r.Healthy, r.Said = &no, "device errors: "+oneLineOf(err.Error())
			return r
		}
		r.Healthy, r.Said = &yes, firstLineOf(stats)
		return r
	}
	return nil
}

var mdMembers = regexp.MustCompile(`\[([U_]+)\]`)

// mdHealth reads one array's block of /proc/mdstat: healthy when every member is up and it is not
// recovering.
func mdHealth(mdstat, device string) (bool, string, bool) {
	lines := strings.Split(mdstat, "\n")
	for i, l := range lines {
		if !strings.HasPrefix(l, device+" ") {
			continue
		}
		block := l
		for j := i + 1; j < len(lines) && strings.HasPrefix(lines[j], " "); j++ {
			block += " " + strings.TrimSpace(lines[j])
		}
		members := mdMembers.FindStringSubmatch(block)
		healthy := members != nil && !strings.Contains(members[1], "_") && !strings.Contains(block, "recovery")
		return healthy, oneLineOf(block), true
	}
	return false, "", false
}

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

func oneLineOf(s string) string { return strings.Join(strings.Fields(s), " ") }

// MeasureAll measures every item that is not a cache, one at a time, and keeps what it found. An item
// measured by a walk is walked only when walks is true or its last walk is older than walkEvery: the
// hourly round reads counters, and a walk runs at most once a day.
func (b *Backups) MeasureAll(ctx context.Context, walks bool) error {
	measuring.Lock()
	defer measuring.Unlock()
	items, err := b.Items()
	if err != nil {
		return err
	}
	before := b.Measurements()
	found := map[string]map[string]Measured{}
	for module, its := range items {
		for _, it := range its {
			if it.Class == classCache {
				continue
			}
			if found[module] == nil {
				found[module] = map[string]Measured{}
			}
			if was, ok := before[module][it.ID]; ok && it.Measure == measureWalk && !walks && was.MeasuredAt != nil &&
				b.Now().Sub(*was.MeasuredAt) < walkEvery {
				found[module][it.ID] = was
				continue
			}
			found[module][it.ID] = b.measure(ctx, it)
		}
	}
	raw, _ := json.MarshalIndent(found, "", "  ")
	return os.WriteFile(b.measuredFile(), append(raw, '\n'), 0o600)
}

func (b *Backups) goodFile() string { return filepath.Join(b.Where.State, "good.json") }

// Good is each module's newest good night: what a night that failed since does not erase.
func (b *Backups) Good() map[string]time.Time {
	out := map[string]time.Time{}
	if raw, err := os.ReadFile(b.goodFile()); err == nil {
		_ = json.Unmarshal(raw, &out)
	}
	// A night recorded good before this file existed counts.
	for module, n := range b.Nights() {
		if n.OK && n.At.After(out[module]) {
			out[module] = n.At
		}
	}
	return out
}

func (b *Backups) recordGood(module string, at time.Time) {
	good := b.Good()
	good[module] = at
	raw, _ := json.MarshalIndent(good, "", "  ")
	if err := os.WriteFile(b.goodFile(), append(raw, '\n'), 0o600); err != nil {
		b.Say("recording %s's good night failed: %v", module, err)
	}
}

// itemReports is every item of one module, with its newest measurement and its newest good backup: the
// module's newest good night, where that night keeps the path that covers the item.
func itemReports(its []Item, measured map[string]Measured, paths []string, good time.Time) []ItemReport {
	out := []ItemReport{}
	for _, it := range its {
		r := ItemReport{Item: it, Measured: measured[it.ID]}
		if it.CoveredBy != "" && !good.IsZero() {
			for _, p := range paths {
				if p == it.CoveredBy {
					g := good
					r.LastBackup = &g
				}
			}
		}
		out = append(out, r)
	}
	return out
}

// ---- deleting a retired item -------------------------------------------------------------------

// Deletion is how one deletion went, by path.
type Deletion struct {
	Module   string    `json:"module"`
	Item     string    `json:"item"`
	Started  time.Time `json:"started"`
	Running  bool      `json:"running"`
	Done     bool      `json:"done"`
	OK       bool      `json:"ok"`
	Snapshot string    `json:"snapshot,omitempty"`
	Error    string    `json:"error,omitempty"`
	By       string    `json:"by,omitempty"`
	Why      string    `json:"why,omitempty"`
}

func (b *Backups) deletionsFile() string { return filepath.Join(b.Where.State, "deleted.json") }

var deletionsMu sync.Mutex

func (b *Backups) deletions() map[string]Deletion {
	out := map[string]Deletion{}
	if raw, err := os.ReadFile(b.deletionsFile()); err == nil {
		_ = json.Unmarshal(raw, &out)
	}
	return out
}

func (b *Backups) keepDeletion(path string, d Deletion) {
	deletionsMu.Lock()
	defer deletionsMu.Unlock()
	all := b.deletions()
	all[path] = d
	raw, _ := json.MarshalIndent(all, "", "  ")
	if err := os.WriteFile(b.deletionsFile(), append(raw, '\n'), 0o600); err != nil {
		b.Say("recording the deletion of %s failed: %v", path, err)
	}
}

// refuseDeleting says why a path may not be deleted: not absolute, too near the root, or declared now
// — by any module's item or backup line on this machine, itself, inside one, or holding one.
func (b *Backups) refuseDeleting(path string) error {
	clean := filepath.Clean(path)
	if !safePath.MatchString(path) || clean != strings.TrimRight(path, "/") {
		return fmt.Errorf("%q is not a path this holder deletes", path)
	}
	if strings.Count(clean, "/") < 2 {
		return fmt.Errorf("%s is too near the root to be a module's data", clean)
	}
	near := func(declared string) bool {
		d := filepath.Clean(declared)
		return d == clean || strings.HasPrefix(d, clean+"/") || strings.HasPrefix(clean, d+"/")
	}
	items, err := b.Items()
	if err != nil {
		return fmt.Errorf("what is declared here cannot be read, so nothing is deleted: %v", err)
	}
	for module, its := range items {
		for _, it := range its {
			if near(it.Path) {
				return fmt.Errorf("%s is declared now — %s's %s at %s — so it is not retired; nothing is deleted",
					clean, module, it.ID, it.Path)
			}
		}
	}
	declared, err := b.Declared()
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		return fmt.Errorf("what is backed up here cannot be read, so nothing is deleted: %v", err)
	}
	for _, d := range declared {
		for _, p := range d.Paths {
			if near(p) {
				return fmt.Errorf("%s is backed up now as %s's %s, so it is not retired; nothing is deleted", clean, d.Module, p)
			}
		}
	}
	if clean == filepath.Clean(b.Where.Repository) || strings.HasPrefix(b.Where.Repository, clean+"/") ||
		clean == filepath.Clean(b.Where.State) {
		return fmt.Errorf("%s holds this machine's backups; nothing is deleted", clean)
	}
	return nil
}

// DeleteRetired starts deleting one retired item: a last restore point of it, tagged as retired, then
// its removal — never the removal without the restore point. Answers at once; DeletedOutcome follows
// it. A path whose deletion already finished answers how it went.
func (b *Backups) DeleteRetired(ctx context.Context, module, item, path, confirm, by, why string) (Deletion, error) {
	if module == "" || item == "" || path == "" {
		return Deletion{}, errors.New("module, item and path are required")
	}
	if confirm != item {
		return Deletion{}, fmt.Errorf("confirm is the item's name again (%s), to say this is meant", item)
	}
	if strings.TrimSpace(why) == "" {
		return Deletion{}, errors.New("why is required: a deletion is a person's decision, and says why")
	}
	if d, ok := b.deletions()[path]; ok && (d.Running || (d.Done && d.OK)) {
		return d, nil
	}
	if err := b.refuseDeleting(path); err != nil {
		return Deletion{}, err
	}
	if !b.exists(ctx, path) {
		d := Deletion{Module: module, Item: item, Started: b.Now().UTC(), Done: true, OK: true,
			Error: path + " was already gone", By: by, Why: why}
		b.keepDeletion(path, d)
		return d, nil
	}
	d := Deletion{Module: module, Item: item, Started: b.Now().UTC(), Running: true, By: by, Why: why}
	b.keepDeletion(path, d)
	go b.deleteNow(context.WithoutCancel(ctx), path, d)
	return d, nil
}

func (b *Backups) deleteNow(ctx context.Context, path string, d Deletion) {
	b.mu.Lock()
	defer b.mu.Unlock()
	finish := func(err error) {
		d.Running, d.Done = false, true
		if err != nil {
			d.Error = err.Error()
			b.Say("%s's retired %s at %s was NOT deleted: %v", d.Module, d.Item, path, err)
		} else {
			d.OK = true
			b.Say("%s's retired %s at %s DELETED by %s: %s; its last restore point is %s", d.Module, d.Item, path,
				orSomebody(d.By), d.Why, d.Snapshot)
		}
		b.keepDeletion(path, d)
	}
	if err := b.ensureRepository(ctx); err != nil {
		finish(fmt.Errorf("no restore point could be taken first: %w", err))
		return
	}
	out, err := b.restic(ctx, "backup", "--json", "--tag", tagOf(d.Module), "--tag", "retired="+d.Item, path)
	if err != nil {
		finish(fmt.Errorf("the last restore point could not be taken, so nothing was deleted: %w", err))
		return
	}
	d.Snapshot = snapshotOf(out)
	if d.Snapshot == "" {
		finish(errors.New("restic took the last restore point and named none, so nothing was deleted"))
		return
	}
	if _, err := b.Run(ctx, "rm", "-rf", "--one-file-system", "--", path); err != nil {
		finish(err)
		return
	}
	finish(nil)
}

// DeletedOutcome is how the deletion of a path went.
func (b *Backups) DeletedOutcome(path string) (Deletion, error) {
	d, ok := b.deletions()[path]
	if !ok {
		return Deletion{}, fmt.Errorf("no deletion of %s was asked of this holder", path)
	}
	return d, nil
}

func orSomebody(by string) string {
	if by == "" {
		return "somebody"
	}
	return by
}