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"
	"strings"
	"sync"
	"time"

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

type zfsState struct {
	Policies []storage.Policy `json:"policies"`
}

type zfsEngine struct {
	s  *Server
	mu sync.Mutex
}

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{}
	}
	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 schedule runs itself ----

// runSchedules takes the snapshots that are due. Every 30 s, which is fine for a schedule
// whose finest grain is hourly, and cheap when nothing is due.
func (s *Server) runSchedules(ctx context.Context) {
	t := time.NewTicker(30 * time.Second)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
		z := s.zfsLoad()
		now := time.Now()
		changed := false
		for i := range z.Policies {
			p := &z.Policies[i]
			if !p.Due(now) {
				continue
			}
			res, err := s.storage.Run(ctx, p)
			p.LastRun = now
			p.LastTaken, p.Pruned = res.Taken, len(res.Pruned)
			if err != nil {
				p.LastError = err.Error()
				log.Printf("zfs schedule %q: %v", p.Name, err)
			} else {
				p.LastError = ""
				log.Printf("zfs schedule %q: took %s, expired %d", p.Name, res.Taken, len(res.Pruned))
			}
			changed = true
		}
		if changed {
			if err := s.zfsSave(z); err != nil {
				log.Printf("zfs schedule: saving: %v", err)
			}
		}
	}
}

// ---- endpoints ----

func (s *Server) zfsOverview(w http.ResponseWriter, r *http.Request) {
	z := s.zfsLoad()
	now := time.Now()
	out := make([]map[string]any, 0, len(z.Policies))
	for i := range z.Policies {
		p := z.Policies[i]
		m := map[string]any{"policy": p}
		if p.Enabled {
			m["next"] = p.NextRun(now)
		}
		out = append(out, m)
	}
	jsonOK(w, 200, map[string]any{
		"zfs": s.storage.ZFS(r.Context()), "policies": out, "external": s.externalSnapshotJob(r.Context()),
	})
}

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 != "" {
			p.LastRun, p.LastError, p.LastTaken, p.Pruned = z.Policies[i].LastRun, z.Policies[i].LastError, z.Policies[i].LastTaken, z.Policies[i].Pruned
			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
	}
	log.Printf("zfs schedule %q saved by %s (%s, %s, keep %s)", p.Name, clientIP(r), p.Dataset, p.Every, 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
	}
	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
}

// zfsPreview is what this schedule would destroy if it ran now. Nothing is deleted.
func (s *Server) zfsPreview(w http.ResponseWriter, r *http.Request) {
	p, ok := s.zfsPolicy(r.PathValue("id"))
	if !ok {
		jsonErr(w, 404, "no such schedule")
		return
	}
	doomed, err := s.storage.Doomed(r.Context(), &p)
	if err != nil {
		jsonErr(w, 500, err.Error())
		return
	}
	if doomed == nil {
		doomed = []storage.Snapshot{}
	}
	jsonOK(w, 200, map[string]any{"expired": doomed})
}

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)
	now := time.Now()
	z := s.zfsLoad()
	for i := range z.Policies {
		if z.Policies[i].ID == id {
			z.Policies[i].LastRun, z.Policies[i].LastTaken, z.Policies[i].Pruned = now, res.Taken, len(res.Pruned)
			z.Policies[i].LastError = ""
			if err != nil {
				z.Policies[i].LastError = err.Error()
			}
		}
	}
	_ = s.zfsSave(z)
	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 {
		jsonErr(w, 500, err.Error())
		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)
}