init --patroni user@host ... installs PostgreSQL, Patroni and etcd on the given machines over ssh, with this one as the third etcd vote, makes the role and database, and moves the store in use into the cluster. Each step checks first, so it changes nothing on a cluster that is up. For a store of several hosts the control plane publishes each member's role, state and lag from Patroni, a stopped member as missing, prints it in ctl status, and alerts on it. Making an administrator on PostgreSQL sent a bool to an integer column; it sends an integer now. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
331 lines
10 KiB
Go
331 lines
10 KiB
Go
package cli
|
|
|
|
import (
|
|
"archive/tar"
|
|
"compress/gzip"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"io/fs"
|
|
"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 — the
|
|
// members of a replicated cluster — become libpq's own lists in PGHOST and
|
|
// PGPORT; read as one URL host, the list was one name libpq could not
|
|
// resolve, and a backup worked only by falling through to the second.
|
|
func pgEnv(dsn string) ([]string, error) {
|
|
hosts, ports, u, err := store.SplitPostgres(dsn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
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 ""
|
|
}
|