diff --git a/internal/dbbackup/dbbackup.go b/internal/dbbackup/dbbackup.go index c328a21..e35db0d 100644 --- a/internal/dbbackup/dbbackup.go +++ b/internal/dbbackup/dbbackup.go @@ -86,6 +86,18 @@ var labelRe = regexp.MustCompile(`^[a-z][a-z0-9-]{0,31}$`) // PATH; tests point them at fakes. type Tools struct { PGDump, PGRestore, PSQL string + // Exec, when set, is the argv prefix every tool runs under. The installer's + // database is a k3s Deployment and the host carries no PostgreSQL client, so + // the tools run in the database's own container: + // `k3s kubectl exec -i -n felis deploy/felis-postgres -c postgres --`. + // kubectl exec carries neither the environment nor files across: the dump + // comes back on stdout and archives go in on stdin, and the tools connect as + // Conn instead of the database URL. + Exec []string + // Conn is the libpq connection string the tools use under Exec: the + // container's own socket, which trusts local connections, so no password + // has to cross into it. + Conn string } func (t Tools) pgDump() string { return orDefault(t.PGDump, "pg_dump") } @@ -222,23 +234,43 @@ func (c conn) env() []string { return append(env, "PGCONNECT_TIMEOUT=15") } -func (c conn) command(ctx context.Context, bin string, args ...string) *exec.Cmd { - cmd := exec.CommandContext(ctx, bin, args...) +// toolCmd is one run of a client tool, named for errors by the tool itself: +// under Tools.Exec the process is kubectl, which says nothing. +type toolCmd struct { + *exec.Cmd + name string +} + +// command runs tool with args, directly or under t.Exec. +func (t Tools) command(ctx context.Context, c conn, tool string, args ...string) toolCmd { + if len(t.Exec) > 0 { + argv := append(append(append([]string{}, t.Exec[1:]...), tool), args...) + return toolCmd{exec.CommandContext(ctx, t.Exec[0], argv...), filepath.Base(tool)} + } + cmd := exec.CommandContext(ctx, tool, args...) cmd.Env = c.env() - return cmd + return toolCmd{cmd, filepath.Base(tool)} +} + +// dsn is what the tools pass as --dbname / -d. +func (t Tools) dsn(c conn) string { + if len(t.Exec) > 0 { + return t.Conn + } + return c.uri } // run executes cmd and folds its stderr into the error. -func run(cmd *exec.Cmd) ([]byte, error) { +func run(cmd toolCmd) ([]byte, error) { var stderr bytes.Buffer cmd.Stderr = &stderr out, err := cmd.Output() if err != nil { msg := strings.TrimSpace(stderr.String()) if msg == "" { - return out, fmt.Errorf("%s: %w", filepath.Base(cmd.Path), err) + return out, fmt.Errorf("%s: %w", cmd.name, err) } - return out, fmt.Errorf("%s: %w: %s", filepath.Base(cmd.Path), err, msg) + return out, fmt.Errorf("%s: %w: %s", cmd.name, err, msg) } return out, nil } @@ -250,15 +282,21 @@ var foreignObjectRe = regexp.MustCompile(`permission denied for (table|sequence| // an object created in the felis database by another role (typically postgres, // from a manual psql session). The dump runs as the felis role and must read // everything; leaving the object out would make the bundle an incomplete restore. -func dumpHint(err error, db DatabaseInfo) string { +func dumpHint(err error, db DatabaseInfo, t Tools) string { m := foreignObjectRe.FindStringSubmatch(err.Error()) if m == nil { return "" } + // The superuser's psql: the host's postgres account, or the image's + // postgres role over the container's socket. + su := "sudo -u postgres psql" + if len(t.Exec) > 0 { + su = "sudo " + strings.Join(t.Exec, " ") + " psql -U postgres" + } kind, name := strings.ToUpper(m[1]), m[2] return fmt.Sprintf("\n %s %s belongs to another role, so %s cannot dump it. Hand it over with\n"+ - " sudo -u postgres psql -d %s -c 'ALTER %s %s OWNER TO %s'\n"+ - " or drop it if it is a leftover.", strings.ToLower(kind), name, db.User, db.Name, kind, name, db.User) + " %s -d %s -c 'ALTER %s %s OWNER TO %s'\n"+ + " or drop it if it is a leftover.", strings.ToLower(kind), name, db.User, su, db.Name, kind, name, db.User) } // BundleName is the file name of a bundle taken at t with label. @@ -424,20 +462,17 @@ func Backup(ctx context.Context, o BackupOptions) (string, error) { dump := filepath.Join(o.Dir, "."+name+".dump.partial") defer os.Remove(dump) - if _, err := run(c.command(ctx, o.Tools.pgDump(), "--format=custom", "--no-password", "--file="+dump, "--dbname="+c.uri)); err != nil { - return "", fmt.Errorf("dump the database: %w%s", err, dumpHint(err, c.info)) - } - if err := os.Chmod(dump, 0o600); err != nil { + if err := dumpTo(ctx, c, o.Tools, dump); err != nil { return "", err } // A dump pg_restore cannot read is not a backup; find out now, not on the // day it is needed. - if _, err := run(exec.CommandContext(ctx, o.Tools.pgRestore(), "--list", dump)); err != nil { + if err := listArchive(ctx, c, o.Tools, dump); err != nil { return "", fmt.Errorf("the dump does not read back: %w", err) } m := Manifest{Format: formatV1, CreatedAt: created, Label: o.Label, FelisVersion: o.Version, Database: c.info} - if out, err := run(exec.CommandContext(ctx, o.Tools.pgDump(), "--version")); err == nil { + if out, err := run(o.Tools.command(ctx, c, o.Tools.pgDump(), "--version")); err == nil { m.PGDumpVersion = strings.TrimSpace(string(out)) } m.SchemaVersion = schemaVersion(ctx, c, o.Tools) @@ -496,6 +531,44 @@ func Backup(ctx context.Context, o BackupOptions) (string, error) { return final, nil } +// dumpTo writes pg_dump's custom-format archive of the database to path, +// created 0600. The archive travels on stdout, the one channel that reaches +// the host from a tool running under Tools.Exec. +func dumpTo(ctx context.Context, c conn, t Tools, path string) error { + f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) + if err != nil { + return err + } + cmd := t.command(ctx, c, t.pgDump(), "--format=custom", "--no-password", "--dbname="+t.dsn(c)) + var stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = f, &stderr + err = cmd.Run() + if cerr := f.Close(); err == nil && cerr != nil { + return cerr + } + if err != nil { + err = fmt.Errorf("%s: %w", cmd.name, err) + if msg := strings.TrimSpace(stderr.String()); msg != "" { + err = fmt.Errorf("%w: %s", err, msg) + } + return fmt.Errorf("dump the database: %w%s", err, dumpHint(err, c.info, t)) + } + return nil +} + +// listArchive proves pg_restore can read the archive at path, fed on stdin. +func listArchive(ctx context.Context, c conn, t Tools, path string) error { + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + cmd := t.command(ctx, c, t.pgRestore(), "--list") + cmd.Stdin = f + _, err = run(cmd) + return err +} + // record upserts st into platform_settings. The JSON travels as a psql // variable, quoted by psql itself, over stdin (-c does not interpolate). func record(ctx context.Context, c conn, t Tools, st Status) error { @@ -503,7 +576,7 @@ func record(ctx context.Context, c conn, t Tools, st Status) error { if err != nil { return err } - cmd := c.command(ctx, t.psql(), "-X", "-q", "-w", "-v", "ON_ERROR_STOP=1", "-v", "v="+string(v), "-d", c.uri) + cmd := t.command(ctx, c, t.psql(), "-X", "-q", "-w", "-v", "ON_ERROR_STOP=1", "-v", "v="+string(v), "-d", t.dsn(c)) cmd.Stdin = strings.NewReader("INSERT INTO platform_settings (key, value) VALUES ('" + StatusKey + "', :'v'::jsonb)\n" + "ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now();\n") _, err = run(cmd) @@ -531,7 +604,7 @@ felis_db_backup_last_size_bytes{label=%q} %d // schemaVersion reads the newest applied migration, or 0 when it cannot. func schemaVersion(ctx context.Context, c conn, t Tools) int { - out, err := run(c.command(ctx, t.psql(), "-X", "-q", "-t", "-A", "-w", "-d", c.uri, + out, err := run(t.command(ctx, c, t.psql(), "-X", "-q", "-t", "-A", "-w", "-d", t.dsn(c), "-c", "SELECT coalesce(max(version), 0) FROM schema_migrations")) if err != nil { return 0 diff --git a/internal/dbbackup/dbbackup_test.go b/internal/dbbackup/dbbackup_test.go index e042147..249fe26 100644 --- a/internal/dbbackup/dbbackup_test.go +++ b/internal/dbbackup/dbbackup_test.go @@ -27,19 +27,22 @@ printf '%s\n' "$*" > "$D/pg_dump.args" printf '%s' "$PGPASSWORD" > "$D/pg_dump.password" [ -f "$D/dump_fail" ] && { echo "pg_dump: error: connection refused" >&2; exit 1; } [ -f "$D/dump_denied" ] && { printf 'pg_dump: error: query failed: ERROR: permission denied for table servers_preserve\npg_dump: error: query was: LOCK TABLE public.servers_preserve IN ACCESS SHARE MODE\n' >&2; exit 1; } -for a in "$@"; do case "$a" in --file=*) out="${a#--file=}";; esac; done -if [ -f "$D/dump_garbage" ]; then echo garbage > "$out"; exit 0; fi -{ printf 'PGDMP\n'; cat "$D/db"; } > "$out" +for a in "$@"; do case "$a" in --file=*|-f) echo "pg_dump: the archive must come back on stdout" >&2; exit 2;; esac; done +if [ -f "$D/dump_garbage" ]; then echo garbage; exit 0; fi +printf 'PGDMP\n'; cat "$D/db" ` const fakePGRestore = `#!/bin/sh D="$FAKE_DIR" list=0 -for a in "$@"; do case "$a" in --list) list=1;; esac; last="$a"; done -head -n 1 "$last" | grep -q '^PGDMP$' || { echo "pg_restore: error: input file does not appear to be a valid archive" >&2; exit 1; } -[ $list = 1 ] && { echo "; Archive created"; exit 0; } +for a in "$@"; do case "$a" in --list) list=1;; -*) ;; *) echo "pg_restore: the archive must arrive on stdin, got $a" >&2; exit 2;; esac; done +in="$D/restore.in.$$" +cat > "$in" +head -n 1 "$in" | grep -q '^PGDMP$' || { rm -f "$in"; echo "pg_restore: error: input file does not appear to be a valid archive" >&2; exit 1; } +[ $list = 1 ] && { rm -f "$in"; echo "; Archive created"; exit 0; } echo "-- restore script" -tail -n +2 "$last" | sed 's/^/DATA /' +tail -n +2 "$in" | sed 's/^/DATA /' +rm -f "$in" [ -f "$D/restore_fail" ] && { echo "pg_restore: error: could not read input" >&2; exit 1; } exit 0 ` @@ -67,8 +70,22 @@ tail -n 1 "$D/psql.stdin" | grep -q '^COMMIT;$' || exit 0 grep '^DATA ' "$D/psql.stdin" | sed 's/^DATA //' > "$D/db" ` +// fakeExec stands in for kubectl exec: the argv up to -- is its own, and none +// of the caller's environment crosses into the "container". +const fakeExec = `#!/bin/sh +D="$FAKE_DIR" +pre="" +while [ $# -gt 0 ] && [ "$1" != "--" ]; do pre="$pre $1"; shift; done +shift +printf '%s\n' "$pre" >> "$D/exec.args" +exec env -i FAKE_DIR="$D" PATH="$PATH" "$@" +` + const testURL = "postgres://felis:s3cret-pw@127.0.0.1:5432/felis?sslmode=disable" +// podConn is the in-container connection the Exec tests hand the tools. +const podConn = "host=/var/run/postgresql port=5432 dbname=felis user=felis" + type fakePG struct { dir string tools Tools @@ -92,6 +109,32 @@ func newFakePG(t *testing.T, db string) *fakePG { return f } +// newFakePod is newFakePG with the tools behind a fake kubectl exec. +func newFakePod(t *testing.T, db string) *fakePG { + t.Helper() + f := newFakePG(t, db) + ex := filepath.Join(f.dir, "kubectl") + if err := os.WriteFile(ex, []byte(fakeExec), 0o755); err != nil { + t.Fatal(err) + } + f.tools.Exec = []string{ex, "exec", "-i", "-n", "felis", "deploy/felis-postgres", "-c", "postgres", "--"} + f.tools.Conn = podConn + return f +} + +// execRuns is how many tool runs went through the fake kubectl exec. +func (f *fakePG) execRuns(t *testing.T) int { + t.Helper() + b, _ := os.ReadFile(filepath.Join(f.dir, "exec.args")) + n := 0 + for _, l := range strings.Split(string(b), "\n") { + if strings.Contains(l, "deploy/felis-postgres -c postgres") { + n++ + } + } + return n +} + func (f *fakePG) setDB(t *testing.T, s string) { t.Helper() if err := os.WriteFile(filepath.Join(f.dir, "db"), []byte(s), 0o600); err != nil { @@ -353,6 +396,74 @@ func TestBackupFailures(t *testing.T) { }) } +// The installer's database is a k3s pod and the host has no client tools: every +// tool runs behind kubectl exec, reaches the database through the container's +// socket, and moves the archive over stdout and stdin. +func TestBackupAndRestoreThroughThePod(t *testing.T) { + pg := newFakePod(t, "users: alice\n") + dir := t.TempDir() + path, err := Backup(context.Background(), BackupOptions{ + DatabaseURL: testURL, Dir: dir, Label: LabelDaily, Tools: pg.tools, Now: at(t0), Record: true, + }) + if err != nil { + t.Fatalf("Backup: %v", err) + } + m, err := Verify(path) + if err != nil { + t.Fatal(err) + } + if m.SchemaVersion != 21 || !strings.Contains(m.PGDumpVersion, "13.23") { + t.Errorf("manifest = %+v", m) + } + args, _ := os.ReadFile(filepath.Join(pg.dir, "pg_dump.args")) + if !strings.Contains(string(args), "--dbname="+podConn) || strings.Contains(string(args), "127.0.0.1") { + t.Errorf("pg_dump args = %s, want the container's socket", args) + } + if pw, _ := os.ReadFile(filepath.Join(pg.dir, "pg_dump.password")); len(pw) != 0 { + t.Errorf("PGPASSWORD reached the container: %q", pw) + } + if rec, _ := os.ReadFile(filepath.Join(pg.dir, "record.args")); !strings.Contains(string(rec), podConn) { + t.Errorf("freshness record args = %s", rec) + } + // dump, list, --version, schema version, record + if n := pg.execRuns(t); n != 5 { + t.Errorf("%d tool runs went through kubectl exec, want 5", n) + } + + pg.setDB(t, "users: alice\nusers: bob\n") + if _, err := restore(pg, dir, path, nil); err != nil { + t.Fatalf("Restore: %v", err) + } + if got := pg.db(t); got != "users: alice\n" { + t.Fatalf("db after restore = %q", got) + } + if args, _ := os.ReadFile(filepath.Join(pg.dir, "psql.args")); !strings.Contains(string(args), podConn) { + t.Errorf("replay psql args = %s", args) + } + + t.Run("a failed replay still rolls back", func(t *testing.T) { + pg.setDB(t, "current\n") + pg.flag(t, "restore_fail", "") + defer os.Remove(filepath.Join(pg.dir, "restore_fail")) + if _, err := restore(pg, dir, path, func(o *RestoreOptions) { o.SkipSafetyBackup = true }); err == nil || !strings.Contains(err.Error(), "rolled back") { + t.Fatalf("err = %v", err) + } + if got := pg.db(t); got != "current\n" { + t.Fatalf("db = %q, want it untouched", got) + } + }) + + t.Run("the ownership hint uses the container's superuser", func(t *testing.T) { + pg.flag(t, "dump_denied", "") + defer os.Remove(filepath.Join(pg.dir, "dump_denied")) + _, err := Backup(context.Background(), BackupOptions{DatabaseURL: testURL, Dir: t.TempDir(), Label: LabelDaily, Tools: pg.tools, Now: at(t0)}) + want := "sudo " + strings.Join(pg.tools.Exec, " ") + " psql -U postgres -d felis -c 'ALTER TABLE servers_preserve OWNER TO felis'" + if err == nil || !strings.Contains(err.Error(), want) { + t.Fatalf("err = %v, want %q", err, want) + } + }) +} + func TestVerifyCatchesCorruption(t *testing.T) { pg := newFakePG(t, strings.Repeat("row\n", 64)) dir := t.TempDir() diff --git a/internal/dbbackup/restore.go b/internal/dbbackup/restore.go index 441feb2..7234823 100644 --- a/internal/dbbackup/restore.go +++ b/internal/dbbackup/restore.go @@ -200,7 +200,7 @@ func Restore(ctx context.Context, o RestoreOptions) (Manifest, string, error) { if err != nil { return m, "", err } - if _, err := run(exec.CommandContext(ctx, o.Tools.pgRestore(), "--list", scratch.Name())); err != nil { + if err := listArchive(ctx, c, o.Tools, scratch.Name()); err != nil { return m, "", fmt.Errorf("the bundle's dump does not read: %w", err) } @@ -255,7 +255,7 @@ func recordNewest(ctx context.Context, c conn, t Tools, dir string) error { // otherClients counts client sessions on the database other than this one. func otherClients(ctx context.Context, c conn, t Tools) (int, error) { - out, err := run(c.command(ctx, t.psql(), "-X", "-q", "-t", "-A", "-w", "-d", c.uri, "-c", + out, err := run(t.command(ctx, c, t.psql(), "-X", "-q", "-t", "-A", "-w", "-d", t.dsn(c), "-c", "SELECT count(*) FROM pg_stat_activity WHERE datname = current_database() AND pid <> pg_backend_pid() AND backend_type = 'client backend'")) if err != nil { return 0, err @@ -272,16 +272,24 @@ const dropOwned = "DROP OWNED BY CURRENT_USER;\n" // exited cleanly: a generator that dies mid-stream leaves psql at EOF inside an // open transaction, which the server rolls back when psql disconnects. psql's // own --single-transaction would commit whatever arrived before that EOF. +// +// pg_restore reads the archive on stdin, sequentially, which is all a full +// restore needs and the only way in under Tools.Exec. func replay(ctx context.Context, c conn, t Tools, dump string) error { ctx, cancel := context.WithCancel(ctx) defer cancel() + in, err := os.Open(dump) + if err != nil { + return err + } + defer in.Close() r, w, err := os.Pipe() if err != nil { return err } var restoreErr, psqlErr bytes.Buffer - gen := exec.CommandContext(ctx, t.pgRestore(), "--no-owner", "--no-privileges", "--file=-", dump) - gen.Stdout, gen.Stderr = w, &restoreErr + gen := t.command(ctx, c, t.pgRestore(), "--no-owner", "--no-privileges", "--file=-") + gen.Stdin, gen.Stdout, gen.Stderr = in, w, &restoreErr if err := gen.Start(); err != nil { r.Close() w.Close() @@ -289,8 +297,8 @@ func replay(ctx context.Context, c conn, t Tools, dump string) error { } w.Close() - tail := &commitAfter{gen: gen} - apply := c.command(ctx, t.psql(), "-X", "-q", "-w", "-v", "ON_ERROR_STOP=1", "-d", c.uri) + tail := &commitAfter{gen: gen.Cmd} + apply := t.command(ctx, c, t.psql(), "-X", "-q", "-w", "-v", "ON_ERROR_STOP=1", "-d", t.dsn(c)) apply.Stdin = io.MultiReader(strings.NewReader("BEGIN;\n"+dropOwned), r, tail) apply.Stdout, apply.Stderr = io.Discard, &psqlErr aerr := apply.Run()