llmgw/internal/cli/backup.go

package cli

import (
	"archive/tar"
	"compress/gzip"
	"context"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"slices"
	"strings"
	"time"

	"github.com/spf13/cobra"

	"gitea.mynodes.xyz/ilker/llmgw/internal/config"
	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
	"gitea.mynodes.xyz/ilker/llmgw/internal/version"
)

// DefaultBackupDir is where backups are written and where the control plane
// looks for the newest one.
//
// Its own directory rather than /var/backups/llmgw, which a PostgreSQL move
// had already made for its own dumps, owned by postgres and closed to
// everyone else — so the control plane could not see a backup landing in it.
const DefaultBackupDir = "/var/backups/llmgw-server"

func newBackupCmd() *cobra.Command {
	var (
		configPath string
		etcDir     string
		dir        string
		keep       int
	)
	c := &cobra.Command{
		Use:   "backup",
		Short: "Write the store and /etc/llmgw to one archive, and keep the last few",
		Long: "Everything a control plane cannot be rebuilt without: the store — keys,\n" +
			"people, quotas, usage, the models added at runtime, the secret engine\n" +
			"keys derive from — and /etc/llmgw, which holds the configuration, the\n" +
			"store's address and the certificate authority's key.\n\n" +
			"The archive holds that key, so it is written readable by root only. It\n" +
			"is on the same disk as what it backs up: copy it elsewhere, or it does\n" +
			"not survive the loss of this machine.\n\n" +
			"Run daily by llmgw-backup.timer (`llmgw server backup`). The control plane publishes the time of\n" +
			"the newest archive as llmgw_backup_last_success_timestamp_seconds, and\n" +
			"BackupTooOld fires when it is more than a day old.",
		RunE: func(cmd *cobra.Command, _ []string) error {
			cfg, err := config.Load(configPath)
			if err != nil {
				return err
			}
			path, err := writeBackup(cmd.Context(), cfg.Auth.Store, etcDir, dir)
			if err != nil {
				return err
			}
			if why := unlistable(dir); why != "" {
				fmt.Fprintf(cmd.ErrOrStderr(), "warning: %s; the control plane will report no backup "+
					"and BackupTooOld will fire\n", why)
			}
			removed, err := pruneBackups(dir, keep)
			if err != nil {
				return err
			}
			fmt.Fprintf(cmd.OutOrStdout(), "wrote %s", path)
			if removed > 0 {
				fmt.Fprintf(cmd.OutOrStdout(), "; removed %d older", removed)
			}
			fmt.Fprintln(cmd.OutOrStdout())
			return nil
		},
	}
	c.Flags().StringVar(&configPath, "config", "/etc/llmgw/server.yaml", "the control plane's configuration")
	c.Flags().StringVar(&etcDir, "etc", "/etc/llmgw", "configuration directory to include")
	c.Flags().StringVar(&dir, "dir", DefaultBackupDir, "where archives are written")
	c.Flags().IntVar(&keep, "keep", 14, "how many archives to keep")
	return c
}

const backupPrefix = "llmgw-backup-"

// writeBackup writes one archive and returns its path. Written beside its
// final name and renamed, so a backup interrupted halfway is never the newest
// one the control plane reports.
func writeBackup(ctx context.Context, storeAddr, etcDir, dir string) (string, error) {
	// Listable by anyone, so the control plane — not root — can see when the
	// newest archive was written. The archives themselves are root's alone.
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return "", err
	}
	host, _ := os.Hostname()
	name := backupPrefix + host + "-" + time.Now().UTC().Format("20060102T150405Z") + ".tar.gz"
	final := filepath.Join(dir, name)
	part := final + ".part"

	f, err := os.OpenFile(part, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
	if err != nil {
		return "", err
	}
	ok := false
	defer func() {
		if !ok {
			_ = os.Remove(part)
		}
	}()
	gz := gzip.NewWriter(f)
	tw := tar.NewWriter(gz)

	manifest := fmt.Sprintf("llmgw %s\nhost %s\ntime %s\n", version.Version, host,
		time.Now().UTC().Format(time.RFC3339))
	if err := addBytes(tw, "MANIFEST", []byte(manifest)); err != nil {
		return "", err
	}
	if storeAddr != "" {
		if err := addStore(ctx, tw, storeAddr, dir); err != nil {
			return "", fmt.Errorf("the store: %w", err)
		}
	}
	if err := addTree(tw, etcDir, "etc/llmgw"); err != nil {
		return "", fmt.Errorf("%s: %w", etcDir, err)
	}
	if err := errors.Join(tw.Close(), gz.Close(), f.Sync(), f.Close()); err != nil {
		return "", err
	}
	if err := os.Rename(part, final); err != nil {
		return "", err
	}
	ok = true
	return final, nil
}

// addStore dumps the store into the archive.
func addStore(ctx context.Context, tw *tar.Writer, addr, scratch string) error {
	addr, err := store.ResolveAddress(addr)
	if err != nil {
		return err
	}
	// Inside a directory only its owner can enter. The archive directory is
	// listable on purpose, and VACUUM INTO creates its file with the default
	// mode: made there directly, a full copy of the store — password hashes,
	// the engine-key secret — sat readable by any local user for as long as
	// the backup took.
	tmpDir, err := os.MkdirTemp(scratch, ".store-")
	if err != nil {
		return err
	}
	defer os.RemoveAll(tmpDir)
	tmpPath := filepath.Join(tmpDir, "store")

	name := "store.sqlite"
	if store.IsPostgres(addr) {
		name = "store.pgdump"
		if err := pgDump(ctx, addr, tmpPath); err != nil {
			return err
		}
	} else {
		db, err := store.Open(ctx, addr)
		if err != nil {
			return err
		}
		err = db.SnapshotTo(ctx, tmpPath)
		_ = db.Close()
		if err != nil {
			return err
		}
	}
	return addFile(tw, tmpPath, name)
}

// pgDump runs pg_dump in its custom format, which pg_restore can restore
// selectively. The connection is given through the environment, not the
// command line: a DSN carries the password, and every user on the machine
// can read another process's arguments.
func pgDump(ctx context.Context, dsn, out string) error {
	vars, err := pgEnv(dsn)
	if err != nil {
		return err
	}
	cmd := exec.CommandContext(ctx, "pg_dump", "--format=custom", "--no-owner", "--file="+out) //nolint:gosec // fixed binary, our own file
	cmd.Env = append(os.Environ(), vars...)
	if b, err := cmd.CombinedOutput(); err != nil {
		return fmt.Errorf("pg_dump: %v: %s", err, strings.TrimSpace(string(b)))
	}
	return nil
}

// pgEnv turns a store address into libpq's environment.
//
// Several hosts are allowed, as pgx allows them — "h1:5432,h2:5432", the
// members of a replicated cluster — and become libpq's own lists in PGHOST and
// PGPORT. url.Parse does not read such a host: it took everything before the
// last colon as one name, which libpq could not resolve, and a backup worked
// only because it fell through to the second host.
func pgEnv(dsn string) ([]string, error) {
	scheme, rest, ok := strings.Cut(dsn, "://")
	if !ok {
		return nil, fmt.Errorf("the store address is not a URL")
	}
	authority, tail := rest, ""
	if i := strings.IndexAny(rest, "/?"); i >= 0 {
		authority, tail = rest[:i], rest[i:]
	}
	userinfo, hostlist := "", authority
	if i := strings.LastIndexByte(authority, '@'); i >= 0 {
		userinfo, hostlist = authority[:i+1], authority[i+1:]
	}
	var hosts, ports []string
	for _, hp := range strings.Split(hostlist, ",") {
		u, err := url.Parse("x://" + hp)
		if err != nil || u.Hostname() == "" {
			return nil, fmt.Errorf("the store address has a host that is not one: %q", hp)
		}
		port := u.Port()
		if port == "" {
			port = "5432"
		}
		hosts, ports = append(hosts, u.Hostname()), append(ports, port)
	}
	// The rest of the address with one placeholder host, which url.Parse reads.
	u, err := url.Parse(scheme + "://" + userinfo + "host" + tail)
	if err != nil {
		return nil, fmt.Errorf("the store address is not a URL")
	}
	var env []string
	set := func(k, v string) {
		if v != "" {
			env = append(env, k+"="+v)
		}
	}
	set("PGHOST", strings.Join(hosts, ","))
	set("PGPORT", strings.Join(ports, ","))
	set("PGUSER", u.User.Username())
	if pw, ok := u.User.Password(); ok {
		set("PGPASSWORD", pw)
	}
	set("PGDATABASE", strings.TrimPrefix(u.Path, "/"))
	set("PGSSLMODE", u.Query().Get("sslmode"))
	// A standby if there is one, else whichever answers — not the address's
	// read-write: a dump only reads, it need not load the primary, and one
	// that insists on the primary fails exactly while the cluster is failing
	// over.
	if len(hosts) > 1 {
		set("PGTARGETSESSIONATTRS", "prefer-standby")
	}
	return env, nil
}

func addTree(tw *tar.Writer, root, as string) error {
	return filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}
		rel, _ := filepath.Rel(root, p)
		name := filepath.ToSlash(filepath.Join(as, rel))
		info, err := d.Info()
		if err != nil {
			return err
		}
		switch {
		case d.IsDir():
			return tw.WriteHeader(&tar.Header{Name: name + "/", Mode: int64(info.Mode().Perm()),
				Typeflag: tar.TypeDir, ModTime: info.ModTime()})
		case info.Mode().IsRegular():
			return addFile(tw, p, name)
		}
		return nil // sockets, links: not configuration
	})
}

func addFile(tw *tar.Writer, path, name string) error {
	f, err := os.Open(path) //nolint:gosec // files we chose to back up
	if err != nil {
		return err
	}
	defer f.Close()
	info, err := f.Stat()
	if err != nil {
		return err
	}
	if err := tw.WriteHeader(&tar.Header{Name: name, Mode: int64(info.Mode().Perm()),
		Size: info.Size(), ModTime: info.ModTime(), Typeflag: tar.TypeReg}); err != nil {
		return err
	}
	_, err = io.Copy(tw, f)
	return err
}

func addBytes(tw *tar.Writer, name string, b []byte) error {
	if err := tw.WriteHeader(&tar.Header{Name: name, Mode: 0o600, Size: int64(len(b)),
		ModTime: time.Now(), Typeflag: tar.TypeReg}); err != nil {
		return err
	}
	_, err := tw.Write(b)
	return err
}

// pruneBackups keeps the newest keep archives. Only files this command
// wrote are considered, so nothing else in the directory is ever removed.
func pruneBackups(dir string, keep int) (int, error) {
	entries, err := os.ReadDir(dir)
	if err != nil {
		return 0, err
	}
	var names []string
	for _, e := range entries {
		if n := e.Name(); strings.HasPrefix(n, backupPrefix) && strings.HasSuffix(n, ".tar.gz") {
			names = append(names, n)
		}
	}
	// The name carries a UTC timestamp, so name order is age order.
	slices.Sort(names)
	removed := 0
	for len(names) > max(keep, 1) {
		if err := os.Remove(filepath.Join(dir, names[0])); err != nil {
			return removed, err
		}
		names = names[1:]
		removed++
	}
	return removed, nil
}

// NewestBackup is the modification time of the newest complete archive in
// dir, or zero when there is none.
func NewestBackup(dir string) time.Time {
	entries, err := os.ReadDir(dir)
	if err != nil {
		return time.Time{}
	}
	var newest time.Time
	for _, e := range entries {
		if n := e.Name(); !strings.HasPrefix(n, backupPrefix) || !strings.HasSuffix(n, ".tar.gz") {
			continue
		}
		if info, err := e.Info(); err == nil && info.ModTime().After(newest) {
			newest = info.ModTime()
		}
	}
	return newest
}

// unlistable says why the control plane, which does not run as root, could
// not see the archives in dir, or "" when it can.
func unlistable(dir string) string {
	info, err := os.Stat(dir)
	if err != nil {
		return err.Error()
	}
	if info.Mode().Perm()&0o005 != 0o005 {
		return fmt.Sprintf("%s is %v, so only its owner can list it", dir, info.Mode().Perm())
	}
	return ""
}