hamlab/internal/api/zfs.go

package api

// The ZFS tab: pools, datasets, snapshots, and a schedule for taking and expiring them.
//
// grid already runs zfSnap from root cron and has done for longer than this repo has
// existed, so this does not replace it. It reads the same snapshot names, expires by the
// same rule, and surfaces the cron job that drives it — so what is on screen is the truth
// about the machine, whether Ham Lab or zfSnap is doing the work. A Ham Lab schedule is for
// a machine that has no such tool, and the page says plainly when both are running.

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"io"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"time"

	"hamlab/internal/storage"
	"hamlab/internal/store"
)

type zfsState struct {
	Policies []storage.Policy       `json:"policies"`
	Verify   []storage.VerifyPolicy `json:"verify"` // landing-tier verify schedules (landing.go)
}

func (s *Server) zfsLoad() zfsState {
	var raw json.RawMessage
	s.store.View(func(st *store.State) { raw = st.ZFS })
	var z zfsState
	if len(raw) > 0 {
		_ = json.Unmarshal(raw, &z)
	}
	if z.Policies == nil {
		z.Policies = []storage.Policy{}
	}
	if z.Verify == nil {
		z.Verify = []storage.VerifyPolicy{}
	}
	return z
}

func (s *Server) zfsSave(z zfsState) error {
	b, err := json.Marshal(z)
	if err != nil {
		return err
	}
	return s.store.Update(func(st *store.State) { st.ZFS = b })
}

// externalSnapshotJob finds a cron job that is already taking snapshots on this machine, so
// the page can say so rather than let two schedules quietly double up.
func (s *Server) externalSnapshotJob(ctx context.Context) map[string]any {
	if s.cron == nil {
		return nil
	}
	ov := s.cron.Overview(ctx)
	for _, j := range ov.Jobs {
		if !j.Enabled {
			continue
		}
		c := strings.ToLower(j.Command)
		if !strings.Contains(c, "zfsnap") && !strings.Contains(c, "zfs-auto-snapshot") && !strings.Contains(c, "sanoid") &&
			!(strings.Contains(c, "zfs ") && strings.Contains(c, "snapshot")) {
			continue
		}
		return map[string]any{
			"tool": toolName(c), "schedule": j.Schedule, "command": j.Command,
			"user": j.User, "source": j.Source, "id": j.ID, "next": j.Next, "last": j.Last,
		}
	}
	return nil
}

func toolName(cmd string) string {
	switch {
	case strings.Contains(cmd, "zfsnap"):
		return "zfSnap"
	case strings.Contains(cmd, "zfs-auto-snapshot"):
		return "zfs-auto-snapshot"
	case strings.Contains(cmd, "sanoid"):
		return "sanoid"
	}
	return "a cron job"
}

// randomID is a short opaque id for a schedule.
func randomID() (string, error) {
	b := make([]byte, 6)
	if _, err := rand.Read(b); err != nil {
		return "", err
	}
	return hex.EncodeToString(b), nil
}

// ---- the cron file ----
//
// A schedule is a cron line, written to /etc/cron.d/hamlab-zfs, which Ham Lab owns and
// rewrites whole whenever the schedules change. Nothing in the daemon runs them: snapshots
// are backup-shaped and should not stop because a web console is down, being upgraded, or
// uninstalled. The line calls `hamlab zfs-snapshot <id>` as the account Ham Lab runs as,
// which reaches zfs through sudo exactly as the daemon does.

// syncCron puts the current schedules on the machine.
func (s *Server) syncCron(ctx context.Context, z zfsState) error {
	exe, err := os.Executable()
	if err != nil {
		return err
	}
	exe, _ = filepath.EvalSymlinks(exe)
	return s.storage.WriteCronFile(ctx, z.Policies, s.user, exe, s.cfg.Path())
}

// ---- endpoints ----

func (s *Server) zfsOverview(w http.ResponseWriter, r *http.Request) {
	z := s.zfsLoad()
	now := time.Now()
	dir := s.cfg.DataDir()
	out := make([]map[string]any, 0, len(z.Policies))
	for i := range z.Policies {
		p := z.Policies[i]
		expr, _ := p.CronExpr()
		m := map[string]any{"policy": p, "cron": expr, "last": storage.ReadRun(dir, p.ID)}
		if p.Enabled {
			if n := p.NextRun(now); !n.IsZero() {
				m["next"] = n
			}
		}
		out = append(out, m)
	}
	jsonOK(w, 200, map[string]any{
		"zfs": s.storage.ZFS(r.Context()), "policies": out, "external": s.externalSnapshotJob(r.Context()),
	})
}

// zfsARC is the cache: the sampler's live ARC (size, the hit rate over the last interval)
// beside the storage package's fuller picture (since-boot counters, tunables, and each
// pool's cache/log/special devices or the lack of them), and the write path — the dirty
// buffer in RAM, each pool's write rate, and any landing tier in front of a pool.
func (s *Server) zfsARC(w http.ResponseWriter, r *http.Request) {
	out := map[string]any{"detail": s.storage.ARC(r.Context()), "live": nil, "writes": nil}
	if smp, ok := s.metrics.Latest(); ok && smp.ARC != nil {
		out["live"] = smp.ARC
		out["writes"] = smp.ZFSWrite
	}
	jsonOK(w, 200, out)
}

func (s *Server) zfsPolicyPut(w http.ResponseWriter, r *http.Request) {
	var p storage.Policy
	if err := json.NewDecoder(io.LimitReader(r.Body, 16<<10)).Decode(&p); err != nil {
		jsonErr(w, 400, "bad request")
		return
	}
	if err := p.Validate(); err != nil {
		jsonErr(w, 400, err.Error())
		return
	}
	z := s.zfsLoad()
	found := false
	for i := range z.Policies {
		if z.Policies[i].ID == p.ID && p.ID != "" {
			z.Policies[i] = p
			found = true
		}
	}
	if !found {
		if p.ID == "" {
			id, err := randomID()
			if err != nil {
				jsonErr(w, 500, err.Error())
				return
			}
			p.ID = id
		}
		z.Policies = append(z.Policies, p)
	}
	if err := s.zfsSave(z); err != nil {
		jsonErr(w, 500, err.Error())
		return
	}
	if err := s.syncCron(r.Context(), z); err != nil {
		jsonErr(w, 500, "saved, but writing "+storage.CronFile+" failed: "+err.Error())
		return
	}
	expr, _ := p.CronExpr()
	log.Printf("zfs schedule %q saved by %s (%s, %q, keep %s)", p.Name, clientIP(r), p.Dataset, expr, p.KeepFor)
	jsonOK(w, 200, p)
}

func (s *Server) zfsPolicyDelete(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	z := s.zfsLoad()
	kept := z.Policies[:0]
	found := false
	for _, p := range z.Policies {
		if p.ID == id {
			found = true
			continue
		}
		kept = append(kept, p)
	}
	if !found {
		jsonErr(w, 404, "no such schedule")
		return
	}
	z.Policies = kept
	if err := s.zfsSave(z); err != nil {
		jsonErr(w, 500, err.Error())
		return
	}
	if err := s.syncCron(r.Context(), z); err != nil {
		jsonErr(w, 500, "removed, but writing "+storage.CronFile+" failed: "+err.Error())
		return
	}
	storage.ForgetRun(s.cfg.DataDir(), id)
	log.Printf("zfs schedule %s removed by %s (its snapshots are left alone)", id, clientIP(r))
	w.WriteHeader(204)
}

func (s *Server) zfsPolicy(id string) (storage.Policy, bool) {
	for _, p := range s.zfsLoad().Policies {
		if p.ID == id {
			return p, true
		}
	}
	return storage.Policy{}, false
}

func (s *Server) zfsRunNow(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	p, ok := s.zfsPolicy(id)
	if !ok {
		jsonErr(w, 404, "no such schedule")
		return
	}
	res, err := s.storage.Run(r.Context(), &p)
	rec := storage.RunRecord{At: time.Now(), Taken: res.Taken, Pruned: res.Pruned}
	if err != nil {
		rec.Error = err.Error()
	}
	_ = storage.WriteRun(s.cfg.DataDir(), id, rec)
	if err != nil {
		jsonErr(w, 500, err.Error())
		return
	}
	log.Printf("zfs schedule %q run by %s: took %s, expired %d", p.Name, clientIP(r), res.Taken, len(res.Pruned))
	jsonOK(w, 200, res)
}

// zfsSnapshotDelete removes one snapshot, named in full and typed back by the caller.
func (s *Server) zfsSnapshotDelete(w http.ResponseWriter, r *http.Request) {
	var body struct {
		Name      string `json:"name"`
		Confirm   string `json:"confirm"`
		Recursive bool   `json:"recursive"`
	}
	if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil {
		jsonErr(w, 400, "bad request")
		return
	}
	if body.Confirm != body.Name {
		jsonErr(w, 400, "type the snapshot's full name to confirm")
		return
	}
	if err := s.storage.DestroySnapshot(r.Context(), body.Name, body.Recursive); err != nil {
		msg := err.Error()
		switch {
		case msg == "bad snapshot name":
			jsonErr(w, 400, msg)
		case strings.Contains(msg, "does not exist"):
			jsonErr(w, 404, msg)
		default:
			jsonErr(w, 500, msg)
		}
		return
	}
	log.Printf("zfs snapshot %s destroyed by %s%s", body.Name, clientIP(r), map[bool]string{true: " (recursive)", false: ""}[body.Recursive])
	w.WriteHeader(204)
}