diff --git a/internal/cli/db.go b/internal/cli/db.go index 85ba2ad..4f10ce2 100644 --- a/internal/cli/db.go +++ b/internal/cli/db.go @@ -6,8 +6,6 @@ import ( "github.com/spf13/cobra" - "github.com/open-source-cloud/devstack/internal/db" - "github.com/open-source-cloud/devstack/internal/docker" "github.com/open-source-cloud/devstack/internal/orchestrate" "github.com/open-source-cloud/devstack/internal/resource" "github.com/open-source-cloud/devstack/internal/state" @@ -41,18 +39,15 @@ func newDbCmd(g *GlobalOpts) *cobra.Command { return cmd } -// defaultPgDumper is the real pg_dump/pg_restore/psql client, shelled behind the -// docker exec runner (the release binary stays CGO-free — the tools are external). -func defaultPgDumper() db.Dumper { return db.PgDumper{Runner: docker.ExecRunner{}} } - // newDbSnapshotCmd wires `db snapshot [name]` (capture) with the `ls` subcommand -// (list). A snapshot dumps ONLY the project's tenant database on the shared -// Postgres to ~/.devstack/snapshots// and records a ledger row (spec 15). +// (list). A snapshot dumps ONLY the project's tenant namespace on the shared engine +// selected by --kind (pg|redis|minio) to ~/.devstack/snapshots// and +// records a ledger row (spec 15). func newDbSnapshotCmd(g *GlobalOpts) *cobra.Command { - var project, database, instance string + var project, database, instance, kind string cmd := &cobra.Command{ Use: "snapshot [name]", - Short: "Capture a project's tenant database to the snapshot store", + Short: "Capture a project's tenant namespace (pg|redis|minio) to the snapshot store", Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error { d, closeFn, err := buildUpDeps(cmd) @@ -60,7 +55,10 @@ func newDbSnapshotCmd(g *GlobalOpts) *cobra.Command { return err } defer closeFn() - dumper := defaultPgDumper() + dumper, err := orchestrate.SelectDumper(d, kind) + if err != nil { + return err + } if err := dumper.Preflight(cmd.Context()); err != nil { return err } @@ -69,7 +67,7 @@ func newDbSnapshotCmd(g *GlobalOpts) *cobra.Command { name = args[0] } meta, err := orchestrate.Snapshot(cmd.Context(), d, dumper, orchestrate.SnapshotOptions{ - Project: project, Database: database, Instance: instance, Name: name, + Kind: kind, Project: project, Database: database, Instance: instance, Name: name, }) if err != nil { return err @@ -78,14 +76,15 @@ func newDbSnapshotCmd(g *GlobalOpts) *cobra.Command { return writeJSON(cmd, meta) } if !g.Quiet { - fmt.Fprintf(cmd.OutOrStdout(), "captured snapshot %q of %s (%d bytes)\n%s\n", meta.Name, meta.Database, meta.Size, meta.Path) + fmt.Fprintf(cmd.OutOrStdout(), "captured %s snapshot %q of %s (%d bytes)\n%s\n", meta.Kind, meta.Name, meta.Database, meta.Size, meta.Path) } return nil }, } + cmd.Flags().StringVar(&kind, "kind", "pg", "engine: pg|redis|minio") cmd.Flags().StringVar(&project, "project", "", "owner project (default: the workspace's single/first project)") - cmd.Flags().StringVar(&database, "db", "", "physical tenant database (default: the project's own database)") - cmd.Flags().StringVar(&instance, "instance", "", "shared postgres instance (default: the first postgres instance)") + cmd.Flags().StringVar(&database, "db", "", "tenant namespace: pg db / redis index / minio bucket (default: derived)") + cmd.Flags().StringVar(&instance, "instance", "", "shared instance (default: the first instance of the engine)") cmd.AddCommand(newDbSnapshotLsCmd(g)) return cmd } @@ -125,11 +124,11 @@ func newDbSnapshotLsCmd(g *GlobalOpts) *cobra.Command { } func newDbRestoreCmd(g *GlobalOpts) *cobra.Command { - var project, database, instance string + var project, database, instance, kind string var force, yes bool cmd := &cobra.Command{ Use: "restore ", - Short: "Restore a project's tenant database from a snapshot (destructive)", + Short: "Restore a project's tenant namespace (pg|redis|minio) from a snapshot (destructive)", Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { if g.JSON && !yes { @@ -140,18 +139,21 @@ func newDbRestoreCmd(g *GlobalOpts) *cobra.Command { return err } defer closeFn() - dumper := defaultPgDumper() + dumper, err := orchestrate.SelectDumper(d, kind) + if err != nil { + return err + } if err := dumper.Preflight(cmd.Context()); err != nil { return err } if !yes { - if !confirm(cmd, fmt.Sprintf("This REPLACES the tenant database from snapshot %q (current data destroyed). Type 'yes' to continue: ", args[0])) { + if !confirm(cmd, fmt.Sprintf("This REPLACES the tenant namespace from snapshot %q (current data destroyed). Type 'yes' to continue: ", args[0])) { fmt.Fprintln(cmd.OutOrStdout(), "aborted") return nil } } meta, err := orchestrate.Restore(cmd.Context(), d, dumper, orchestrate.RestoreOptions{ - Project: project, Database: database, Instance: instance, Name: args[0], Force: force, + Kind: kind, Project: project, Database: database, Instance: instance, Name: args[0], Force: force, }) if err != nil { return err @@ -165,10 +167,11 @@ func newDbRestoreCmd(g *GlobalOpts) *cobra.Command { return nil }, } + cmd.Flags().StringVar(&kind, "kind", "pg", "engine: pg|redis|minio") cmd.Flags().StringVar(&project, "project", "", "owner project (default: the workspace's single/first project)") - cmd.Flags().StringVar(&database, "db", "", "physical tenant database (default: the project's own database)") - cmd.Flags().StringVar(&instance, "instance", "", "shared postgres instance (default: the first postgres instance)") - cmd.Flags().BoolVar(&force, "force", false, "replay over a non-empty database (overwrite existing data)") + cmd.Flags().StringVar(&database, "db", "", "tenant namespace: pg db / redis index / minio bucket (default: derived)") + cmd.Flags().StringVar(&instance, "instance", "", "shared instance (default: the first instance of the engine)") + cmd.Flags().BoolVar(&force, "force", false, "replay over a non-empty namespace (overwrite existing data)") cmd.Flags().BoolVar(&yes, "yes", false, "skip the confirmation prompt (required for --json)") return cmd } diff --git a/internal/db/minio.go b/internal/db/minio.go new file mode 100644 index 0000000..09344be --- /dev/null +++ b/internal/db/minio.go @@ -0,0 +1,217 @@ +package db + +import ( + "archive/tar" + "bytes" + "context" + "fmt" + "io" + "os" + "sort" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" +) + +// This file is the MinIO (S3) snapshot/restore Dumper (spec 15). Unlike the +// pg/redis dumpers it needs NO external binary: it reuses the already-vendored, +// PURE-GO aws-sdk-go-v2 S3 client (so it stays inside the single static binary and +// there is nothing to install). A snapshot lists + gets every object in the +// project's tenant bucket and writes them into a deterministic tar archive under +// the content-addressed snapshot store; a restore reads the tar and puts each +// object back into the SAME bucket (bucket names are globally unique per instance, +// spec 15). It never touches the shared MinIO container — only the tenant bucket's +// objects — so the never-recreate guard holds. +// +// The S3 surface sits behind the S3Factory seam so unit tests run without a live +// endpoint (inject a fake S3Snapshotter); nil selects the real path-style client. + +// S3Snapshotter is the minimal object-copy surface the MinIO dumper needs: list +// the tenant bucket's keys, get one object's bytes, put one object back. The real +// aws-sdk-go-v2 client is wrapped by awsS3Snapshotter; tests inject a fake. +type S3Snapshotter interface { + ListKeys(ctx context.Context, bucket string) ([]string, error) + Get(ctx context.Context, bucket, key string) ([]byte, error) + Put(ctx context.Context, bucket, key string, body []byte) error +} + +// S3Factory builds an S3Snapshotter for a resolved tenant endpoint (the 127.0.0.1 +// overlay host/port + the instance root creds carried on ConnInfo). Injectable so +// the dumper is endpoint-free in tests; nil selects the real client. +type S3Factory func(ctx context.Context, conn ConnInfo) (S3Snapshotter, error) + +// MinioDumper snapshots/restores a tenant bucket via the S3 API. Factory nil → the +// real pure-Go path-style aws-sdk-go-v2 client. +type MinioDumper struct { + Factory S3Factory +} + +// Ensure MinioDumper satisfies the Dumper seam at compile time. +var _ Dumper = MinioDumper{} + +// Preflight always succeeds: the MinIO dumper uses the in-process pure-Go S3 +// client, so there is no external tool to probe (unlike pg/redis). Endpoint/creds +// reachability surfaces on the first List call instead. +func (MinioDumper) Preflight(context.Context) error { return nil } + +// client resolves the S3Snapshotter for a tenant endpoint (the injected factory +// or the real path-style aws-sdk-go-v2 client with the instance root creds). +func (m MinioDumper) client(ctx context.Context, conn ConnInfo) (S3Snapshotter, error) { + if m.Factory != nil { + return m.Factory(ctx, conn) + } + return newAWSS3Snapshotter(conn) +} + +// Snapshot writes every object in the tenant bucket (ConnInfo.Database) into a +// deterministic tar at outPath (keys sorted so identical bucket contents produce +// an identical archive → the content-addressed store dedupes them). Only the +// tenant bucket is read; a snapshot of project A can never read project B's data. +func (m MinioDumper) Snapshot(ctx context.Context, conn ConnInfo, outPath string) error { + c, err := m.client(ctx, conn) + if err != nil { + return err + } + bucket := conn.Database + keys, err := c.ListKeys(ctx, bucket) + if err != nil { + return fmt.Errorf("list objects in bucket %q: %w", bucket, err) + } + sort.Strings(keys) // determinism: byte-identical archive for identical contents + f, err := os.Create(outPath) + if err != nil { + return fmt.Errorf("create snapshot %q: %w", outPath, err) + } + tw := tar.NewWriter(f) + for _, key := range keys { + body, err := c.Get(ctx, bucket, key) + if err != nil { + _ = tw.Close() + _ = f.Close() + return fmt.Errorf("get object %q from %q: %w", key, bucket, err) + } + hdr := &tar.Header{Name: key, Mode: 0o644, Size: int64(len(body)), Typeflag: tar.TypeReg} + if err := tw.WriteHeader(hdr); err != nil { + _ = tw.Close() + _ = f.Close() + return fmt.Errorf("write tar header %q: %w", key, err) + } + if _, err := tw.Write(body); err != nil { + _ = tw.Close() + _ = f.Close() + return fmt.Errorf("write tar body %q: %w", key, err) + } + } + if err := tw.Close(); err != nil { + _ = f.Close() + return fmt.Errorf("close tar writer: %w", err) + } + return f.Close() +} + +// Restore reads the tar at inPath and puts each object back into the SAME tenant +// bucket. The caller has emptied/recreated the bucket (the tenant reset) before +// this runs; a plain put over existing keys overwrites in place. +func (m MinioDumper) Restore(ctx context.Context, conn ConnInfo, inPath string) error { + c, err := m.client(ctx, conn) + if err != nil { + return err + } + bucket := conn.Database + f, err := os.Open(inPath) + if err != nil { + return fmt.Errorf("open snapshot %q: %w", inPath, err) + } + defer f.Close() + tr := tar.NewReader(f) + for { + hdr, err := tr.Next() + if err == io.EOF { + break + } + if err != nil { + return fmt.Errorf("read tar %q: %w", inPath, err) + } + if hdr.Typeflag != tar.TypeReg { + continue + } + body, err := io.ReadAll(tr) + if err != nil { + return fmt.Errorf("read tar entry %q: %w", hdr.Name, err) + } + if err := c.Put(ctx, bucket, hdr.Name, body); err != nil { + return fmt.Errorf("put object %q into %q: %w", hdr.Name, bucket, err) + } + } + return nil +} + +// IsEmpty reports whether the tenant bucket has no objects (the restore-over- +// non-empty guard). +func (m MinioDumper) IsEmpty(ctx context.Context, conn ConnInfo) (bool, error) { + c, err := m.client(ctx, conn) + if err != nil { + return false, err + } + keys, err := c.ListKeys(ctx, conn.Database) + if err != nil { + return false, fmt.Errorf("list objects in bucket %q: %w", conn.Database, err) + } + return len(keys) == 0, nil +} + +// awsS3Snapshotter wraps a real *s3.Client with the small S3Snapshotter surface. +type awsS3Snapshotter struct{ c *s3.Client } + +// newAWSS3Snapshotter builds a pure-Go path-style S3 client against the tenant +// endpoint (http://Host:Port) with the instance root credentials (MinIO does not +// do virtual-host buckets — path style is required). +func newAWSS3Snapshotter(conn ConnInfo) (S3Snapshotter, error) { + cfg := aws.Config{ + Region: "us-east-1", + Credentials: credentials.NewStaticCredentialsProvider(conn.User, conn.Password, ""), + } + endpoint := fmt.Sprintf("http://%s:%d", conn.Host, conn.Port) + c := s3.NewFromConfig(cfg, func(o *s3.Options) { + o.BaseEndpoint = aws.String(endpoint) + o.UsePathStyle = true + }) + return &awsS3Snapshotter{c: c}, nil +} + +func (a *awsS3Snapshotter) ListKeys(ctx context.Context, bucket string) ([]string, error) { + var keys []string + var token *string + for { + out, err := a.c.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: aws.String(bucket), ContinuationToken: token, + }) + if err != nil { + return nil, err + } + for _, o := range out.Contents { + keys = append(keys, aws.ToString(o.Key)) + } + if out.IsTruncated == nil || !*out.IsTruncated { + return keys, nil + } + token = out.NextContinuationToken + } +} + +func (a *awsS3Snapshotter) Get(ctx context.Context, bucket, key string) ([]byte, error) { + out, err := a.c.GetObject(ctx, &s3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)}) + if err != nil { + return nil, err + } + defer out.Body.Close() + return io.ReadAll(out.Body) +} + +func (a *awsS3Snapshotter) Put(ctx context.Context, bucket, key string, body []byte) error { + _, err := a.c.PutObject(ctx, &s3.PutObjectInput{ + Bucket: aws.String(bucket), Key: aws.String(key), Body: bytes.NewReader(body), + }) + return err +} diff --git a/internal/db/minio_test.go b/internal/db/minio_test.go new file mode 100644 index 0000000..be469ca --- /dev/null +++ b/internal/db/minio_test.go @@ -0,0 +1,151 @@ +package db + +import ( + "context" + "os" + "path/filepath" + "sort" + "testing" +) + +// fakeS3 is an in-memory S3Snapshotter: a bucket→key→bytes map that records which +// verbs the dumper called (list/get/put), so tests assert the snapshot/restore +// flow without a live MinIO endpoint. +type fakeS3 struct { + objects map[string]map[string][]byte + listed int + gets int + puts int + putBytes map[string][]byte // key → last put body (restore target) +} + +func newFakeS3() *fakeS3 { + return &fakeS3{objects: map[string]map[string][]byte{}, putBytes: map[string][]byte{}} +} + +func (f *fakeS3) seed(bucket, key string, body []byte) { + if f.objects[bucket] == nil { + f.objects[bucket] = map[string][]byte{} + } + f.objects[bucket][key] = body +} + +func (f *fakeS3) ListKeys(_ context.Context, bucket string) ([]string, error) { + f.listed++ + var keys []string + for k := range f.objects[bucket] { + keys = append(keys, k) + } + sort.Strings(keys) + return keys, nil +} + +func (f *fakeS3) Get(_ context.Context, bucket, key string) ([]byte, error) { + f.gets++ + return f.objects[bucket][key], nil +} + +func (f *fakeS3) Put(_ context.Context, bucket, key string, body []byte) error { + f.puts++ + if f.objects[bucket] == nil { + f.objects[bucket] = map[string][]byte{} + } + cp := append([]byte(nil), body...) + f.objects[bucket][key] = cp + f.putBytes[key] = cp + return nil +} + +func minioConn() ConnInfo { + return ConnInfo{Host: "127.0.0.1", Port: 49000, User: "admin", Password: "secret", Database: "app-bucket"} +} + +func TestMinioDumperSnapshotRestoreRoundTrip(t *testing.T) { + src := newFakeS3() + src.seed("app-bucket", "a/one.txt", []byte("hello")) + src.seed("app-bucket", "b/two.bin", []byte{0x00, 0x01, 0x02, 0x03}) + src.seed("other-bucket", "leak.txt", []byte("SHOULD-NOT-APPEAR")) // tenant B, must never be read + + m := MinioDumper{Factory: func(context.Context, ConnInfo) (S3Snapshotter, error) { return src, nil }} + out := filepath.Join(t.TempDir(), "app.tar") + if err := m.Snapshot(context.Background(), minioConn(), out); err != nil { + t.Fatalf("Snapshot: %v", err) + } + if src.listed == 0 || src.gets != 2 { + t.Errorf("expected 1 list + 2 gets of the tenant bucket, got list=%d get=%d", src.listed, src.gets) + } + fi, err := os.Stat(out) + if err != nil || fi.Size() == 0 { + t.Fatalf("tar not written: %v", err) + } + + // Restore into a fresh (empty) target bucket via a new fake — the objects must + // come back byte-identical, and only into the SAME tenant bucket. + dst := newFakeS3() + m2 := MinioDumper{Factory: func(context.Context, ConnInfo) (S3Snapshotter, error) { return dst, nil }} + if err := m2.Restore(context.Background(), minioConn(), out); err != nil { + t.Fatalf("Restore: %v", err) + } + if dst.puts != 2 { + t.Errorf("expected 2 puts on restore, got %d", dst.puts) + } + if got := string(dst.objects["app-bucket"]["a/one.txt"]); got != "hello" { + t.Errorf("restored a/one.txt = %q, want hello", got) + } + if got := dst.objects["app-bucket"]["b/two.bin"]; string(got) != string([]byte{0x00, 0x01, 0x02, 0x03}) { + t.Errorf("restored b/two.bin = %v, want binary payload", got) + } + // Project B's object was never captured (tenant isolation). + if _, ok := dst.objects["other-bucket"]; ok { + t.Errorf("restore leaked into another tenant's bucket: %v", dst.objects) + } +} + +func TestMinioDumperSnapshotDeterministic(t *testing.T) { + seedOne := func() *fakeS3 { + f := newFakeS3() + f.seed("app-bucket", "z-last", []byte("z")) + f.seed("app-bucket", "a-first", []byte("a")) + f.seed("app-bucket", "m-mid", []byte("m")) + return f + } + m := MinioDumper{Factory: func(context.Context, ConnInfo) (S3Snapshotter, error) { return seedOne(), nil }} + dir := t.TempDir() + a := filepath.Join(dir, "a.tar") + b := filepath.Join(dir, "b.tar") + if err := m.Snapshot(context.Background(), minioConn(), a); err != nil { + t.Fatal(err) + } + if err := m.Snapshot(context.Background(), minioConn(), b); err != nil { + t.Fatal(err) + } + ab, _ := os.ReadFile(a) + bb, _ := os.ReadFile(b) + if string(ab) != string(bb) { + t.Errorf("snapshot archives differ for identical contents (key ordering not deterministic)") + } +} + +func TestMinioDumperIsEmpty(t *testing.T) { + empty := newFakeS3() + m := MinioDumper{Factory: func(context.Context, ConnInfo) (S3Snapshotter, error) { return empty, nil }} + got, err := m.IsEmpty(context.Background(), minioConn()) + if err != nil || !got { + t.Errorf("IsEmpty on empty bucket = %v (err %v), want true", got, err) + } + + full := newFakeS3() + full.seed("app-bucket", "k", []byte("v")) + m2 := MinioDumper{Factory: func(context.Context, ConnInfo) (S3Snapshotter, error) { return full, nil }} + got, err = m2.IsEmpty(context.Background(), minioConn()) + if err != nil || got { + t.Errorf("IsEmpty on non-empty bucket = %v (err %v), want false", got, err) + } +} + +func TestMinioDumperPreflightAlwaysPasses(t *testing.T) { + // Pure-Go S3 client → no external tool to probe. + if err := (MinioDumper{}).Preflight(context.Background()); err != nil { + t.Errorf("Preflight should always pass for the pure-Go MinIO dumper: %v", err) + } +} diff --git a/internal/db/redis.go b/internal/db/redis.go new file mode 100644 index 0000000..a4936e2 --- /dev/null +++ b/internal/db/redis.go @@ -0,0 +1,131 @@ +package db + +import ( + "context" + "fmt" + "os/exec" + "strconv" + "strings" +) + +// This file is the Redis snapshot/restore Dumper (spec 15). It mirrors the pg +// dumper discipline: the external `redis-cli` client is shelled behind the +// injectable Runner (nil in tests → a recording fake), the password rides +// REDISCLI_AUTH in the process env (never on the argv, so it does not leak into +// `ps`), and Preflight degrades the db verbs only (never blocks `up`) when the +// tool is absent, with a one-line remediation. +// +// BEST-EFFORT (documented on purpose): a host client cannot cheaply carve one +// tenant's logical DB out of a live shared Redis. Snapshot therefore captures a +// point-in-time RDB of the WHOLE instance via `redis-cli --rdb` (redis-cli opens +// a replication SYNC and writes the transferred RDB locally — it does NOT stop +// the shared server, so other tenants are undisturbed). Restore streams that +// artifact back through redis-cli's mass-insert pipe. A byte-faithful whole-RDB +// reload into a LIVE shared instance is not possible without a controlled restart +// + dump.rdb swap, which devstack refuses to do to a shared service (the +// never-recreate-a-stateful-shared-service guard). Callers who need per-tenant +// fidelity should run a dedicated Redis instance or a key-prefix workflow. This is +// exactly the engine spec 15 flags as "best-effort" for the shared model. + +// RedisDumper shells `redis-cli`. Runner is injectable (tests inject a recording +// fake); LookPath is injectable so Preflight is unit-testable without the binary. +type RedisDumper struct { + Runner Runner + LookPath func(string) (string, error) // nil → exec.LookPath +} + +// Ensure RedisDumper satisfies the Dumper seam at compile time. +var _ Dumper = RedisDumper{} + +// redisClientTool is the external binary the redis dumper needs on PATH. +const redisClientTool = "redis-cli" + +// Preflight verifies redis-cli is installed. Absence degrades the db verbs only +// (never blocks up), consistent with the mkcert/cloudflared external-binary +// posture (DECISIONS D11/D12). +func (r RedisDumper) Preflight(_ context.Context) error { + look := r.LookPath + if look == nil { + look = exec.LookPath + } + if _, err := look(redisClientTool); err != nil { + return &ErrToolMissing{ + Tool: redisClientTool, + Remediation: "install the Redis client tools (e.g. `apt install redis-tools`, `brew install redis`, or `dnf install redis`) so `redis-cli` is on PATH", + } + } + return nil +} + +// redisFlags builds the shared -h/-p connection flags. When ConnInfo.Database is +// a logical index (e.g. "0"), it is passed as `-n `. The password is NOT +// here — it rides REDISCLI_AUTH in the env (redisEnv). +func redisFlags(c ConnInfo) []string { + f := []string{"-h", c.Host, "-p", strconv.Itoa(c.Port)} + if c.Database != "" { + f = append(f, "-n", c.Database) + } + return f +} + +// redisEnv passes the AUTH password out-of-band so it never lands on the argv. +// An empty password (the default auth-less shared Redis) yields no env entry — +// sending AUTH to an auth-less server is itself an error. +func redisEnv(c ConnInfo) []string { + if c.Password == "" { + return nil + } + return []string{"REDISCLI_AUTH=" + c.Password} +} + +// Snapshot captures a whole-instance RDB to outPath via `redis-cli --rdb`. The +// SYNC does not stop the shared server, so other tenants are undisturbed (spec +// 15). Whole-instance capture is the documented best-effort tradeoff. +func (r RedisDumper) Snapshot(ctx context.Context, conn ConnInfo, outPath string) error { + args := append(redisFlags(conn), "--rdb", outPath) + if err := r.Runner.Run(ctx, redisEnv(conn), "", redisClientTool, args...); err != nil { + return fmt.Errorf("redis-cli --rdb: %w", err) + } + return nil +} + +// Restore streams the captured dump back through redis-cli's mass-insert pipe. +// Because the Runner has no stdin channel, the redirection is expressed through +// `sh -c` (the standard `redis-cli --pipe < dump` incantation). This is the +// best-effort restore path (see the package note): it never restarts or bounces +// the shared container. The password still rides REDISCLI_AUTH in the env. +func (r RedisDumper) Restore(ctx context.Context, conn ConnInfo, inPath string) error { + pipe := redisClientTool + " " + shellJoin(redisFlags(conn)) + " --pipe < " + shellQuote(inPath) + if err := r.Runner.Run(ctx, redisEnv(conn), "", "sh", "-c", pipe); err != nil { + return fmt.Errorf("redis-cli --pipe restore: %w", err) + } + return nil +} + +// IsEmpty reports whether the target keyspace has no keys, via `DBSIZE`. For a +// logical index it counts that index; otherwise it counts db 0. This backs the +// restore-over-non-empty guard. +func (r RedisDumper) IsEmpty(ctx context.Context, conn ConnInfo) (bool, error) { + args := append(redisFlags(conn), "DBSIZE") + out, err := r.Runner.Output(ctx, redisEnv(conn), "", redisClientTool, args...) + if err != nil { + return false, fmt.Errorf("redis-cli DBSIZE: %w", err) + } + n, perr := strconv.Atoi(strings.TrimSpace(string(out))) + if perr != nil { + return false, fmt.Errorf("parse DBSIZE %q: %w", strings.TrimSpace(string(out)), perr) + } + return n == 0, nil +} + +// shellQuote single-quotes a path for a POSIX `sh -c` string (embedded single +// quotes are escaped the standard '"'"' way). +func shellQuote(s string) string { + return "'" + strings.ReplaceAll(s, "'", `'"'"'`) + "'" +} + +// shellJoin space-joins already-safe redis-cli flags (host/port/index — no shell +// metacharacters) for the `sh -c` string. +func shellJoin(args []string) string { + return strings.Join(args, " ") +} diff --git a/internal/db/redis_test.go b/internal/db/redis_test.go new file mode 100644 index 0000000..d0aaaae --- /dev/null +++ b/internal/db/redis_test.go @@ -0,0 +1,123 @@ +package db + +import ( + "context" + "errors" + "strings" + "testing" +) + +func redisConn() ConnInfo { + return ConnInfo{Host: "127.0.0.1", Port: 46379, Password: "s3cr3t", Database: "2"} +} + +func TestRedisDumperSnapshotArgv(t *testing.T) { + r := &recRunner{} + d := RedisDumper{Runner: r} + if err := d.Snapshot(context.Background(), redisConn(), "/tmp/app.rdb"); err != nil { + t.Fatalf("Snapshot: %v", err) + } + argv := strings.Join(r.cmds[0], " ") + for _, want := range []string{"redis-cli", "-h 127.0.0.1", "-p 46379", "-n 2", "--rdb /tmp/app.rdb"} { + if !strings.Contains(argv, want) { + t.Errorf("Snapshot argv missing %q: %s", want, argv) + } + } + // Password only in the env, never on argv. + if strings.Contains(argv, "s3cr3t") { + t.Errorf("password leaked into argv: %s", argv) + } + if got := strings.Join(r.envs[0], " "); got != "REDISCLI_AUTH=s3cr3t" { + t.Errorf("env = %q, want REDISCLI_AUTH=s3cr3t", got) + } +} + +func TestRedisDumperSnapshotNoAuthNoIndex(t *testing.T) { + r := &recRunner{} + d := RedisDumper{Runner: r} + // Auth-less, whole-instance (no logical index): no -n flag, no REDISCLI_AUTH env. + conn := ConnInfo{Host: "127.0.0.1", Port: 6379} + if err := d.Snapshot(context.Background(), conn, "/tmp/all.rdb"); err != nil { + t.Fatalf("Snapshot: %v", err) + } + argv := strings.Join(r.cmds[0], " ") + if strings.Contains(argv, "-n ") { + t.Errorf("unexpected -n flag on whole-instance snapshot: %s", argv) + } + if len(r.envs[0]) != 0 { + t.Errorf("no REDISCLI_AUTH expected for an auth-less server, got %v", r.envs[0]) + } +} + +func TestRedisDumperRestoreArgv(t *testing.T) { + r := &recRunner{} + d := RedisDumper{Runner: r} + if err := d.Restore(context.Background(), redisConn(), "/tmp/app.rdb"); err != nil { + t.Fatalf("Restore: %v", err) + } + // Restore shells `sh -c "redis-cli ... --pipe < "` (no stdin channel on + // the Runner), so the redis-cli invocation is inside the -c payload. + if r.cmds[0][0] != "sh" || r.cmds[0][1] != "-c" { + t.Fatalf("restore did not shell via sh -c: %v", r.cmds[0]) + } + payload := r.cmds[0][2] + for _, want := range []string{"redis-cli", "-h 127.0.0.1", "-p 46379", "-n 2", "--pipe", "/tmp/app.rdb"} { + if !strings.Contains(payload, want) { + t.Errorf("restore payload missing %q: %s", want, payload) + } + } + if strings.Contains(payload, "s3cr3t") { + t.Errorf("password leaked into restore payload: %s", payload) + } + if got := strings.Join(r.envs[0], " "); got != "REDISCLI_AUTH=s3cr3t" { + t.Errorf("env = %q, want REDISCLI_AUTH=s3cr3t", got) + } +} + +func TestRedisDumperIsEmpty(t *testing.T) { + for _, tc := range []struct { + out string + want bool + }{ + {"0\n", true}, + {"42\n", false}, + {" 0 ", true}, + } { + r := &recRunner{output: []byte(tc.out)} + d := RedisDumper{Runner: r} + got, err := d.IsEmpty(context.Background(), redisConn()) + if err != nil { + t.Fatalf("IsEmpty(%q): %v", tc.out, err) + } + if got != tc.want { + t.Errorf("IsEmpty(%q) = %v, want %v", tc.out, got, tc.want) + } + if r.cmds[0][0] != "redis-cli" { + t.Errorf("IsEmpty shelled %q, want redis-cli", r.cmds[0][0]) + } + if last := r.cmds[0][len(r.cmds[0])-1]; last != "DBSIZE" { + t.Errorf("IsEmpty command = %q, want DBSIZE", last) + } + } +} + +func TestRedisPreflightMissingTool(t *testing.T) { + d := RedisDumper{LookPath: func(string) (string, error) { return "", errors.New("nope") }} + err := d.Preflight(context.Background()) + if err == nil { + t.Fatal("Preflight should fail when redis-cli is absent") + } + if !IsToolMissing(err) { + t.Errorf("want ErrToolMissing, got %T: %v", err, err) + } + if !strings.Contains(err.Error(), "redis-cli") || !strings.Contains(err.Error(), "redis-tools") { + t.Errorf("remediation missing: %v", err) + } +} + +func TestRedisPreflightPresent(t *testing.T) { + d := RedisDumper{LookPath: func(string) (string, error) { return "/usr/bin/redis-cli", nil }} + if err := d.Preflight(context.Background()); err != nil { + t.Errorf("Preflight should pass when redis-cli is present: %v", err) + } +} diff --git a/internal/orchestrate/snapshot.go b/internal/orchestrate/snapshot.go index 81f8159..68c4524 100644 --- a/internal/orchestrate/snapshot.go +++ b/internal/orchestrate/snapshot.go @@ -13,6 +13,7 @@ import ( "time" "github.com/open-source-cloud/devstack/internal/db" + "github.com/open-source-cloud/devstack/internal/docker" "github.com/open-source-cloud/devstack/internal/generate" "github.com/open-source-cloud/devstack/internal/lock" "github.com/open-source-cloud/devstack/internal/store" @@ -37,14 +38,16 @@ const snapshotKind = "snapshot" // SnapshotOptions selects the tenant to capture. type SnapshotOptions struct { + Kind string // engine: pg|redis|minio (default: pg) Project string // owner project (default: the workspace's single/first project) - Database string // physical tenant db (default: the project's own db) - Instance string // shared Postgres instance (default: the first postgres instance) + Database string // physical tenant namespace (pg db / redis index / minio bucket; default: derived) + Instance string // shared instance (default: the first instance of the engine) Name string // human label (default: a timestamp label) } // RestoreOptions selects the tenant + snapshot to replay. type RestoreOptions struct { + Kind string // engine: pg|redis|minio (default: pg) Project string Database string Instance string @@ -70,9 +73,58 @@ type SnapshotMeta struct { // (hyphens → underscores), matching provision.EnsureProject's naming. func pgTenantDB(project string) string { return strings.ReplaceAll(project, "-", "_") } -// resolveTenant fills in the (project, database, instance) defaults and validates -// that a shared Postgres instance exists. -func resolveTenant(d UpDeps, project, database, instance string) (proj, dbName, inst string, err error) { +// engineForKind maps the user-facing snapshot kind to the shared-template engine +// name and the stored metadata label. pg/postgres → postgres/pg; redis → redis; +// minio/s3 → minio. The label is what lands in SnapshotMeta.Kind + the dump file +// extension. +func engineForKind(kind string) (engine, label string, err error) { + switch kind { + case "", "pg", "postgres": + return "postgres", "pg", nil + case "redis": + return "redis", "redis", nil + case "minio", "s3": + return "minio", "minio", nil + default: + return "", "", fmt.Errorf("unsupported snapshot kind %q (want pg|redis|minio)", kind) + } +} + +// defaultTenant derives the default per-project namespace for an engine: the +// postgres tenant db, the minio tenant bucket (the project name), or the redis +// logical index (empty → whole instance, the best-effort default). +func defaultTenant(engine, project string) string { + switch engine { + case "postgres": + return pgTenantDB(project) + case "minio": + return project + default: // redis: no index by default (whole-instance best-effort) + return "" + } +} + +// dumpExt is the on-disk extension for a snapshot label (pg → .dump, redis → .rdb, +// minio → .tar). The extension is recorded in the sidecar so restore finds the file. +func dumpExt(label string) string { + switch label { + case "redis": + return ".rdb" + case "minio": + return ".tar" + default: + return ".dump" + } +} + +// resolveTenant fills in the (project, database, instance) defaults for the +// requested engine kind and validates that a matching shared instance exists. +// Returns the shared-template engine + the metadata label alongside the tenant. +func resolveTenant(d UpDeps, kind, project, database, instance string) (proj, dbName, inst, engine, label string, err error) { + engine, label, err = engineForKind(kind) + if err != nil { + return "", "", "", "", "", err + } proj = project if proj == "" { if names := sortedProjects(d.Model); len(names) > 0 { @@ -80,50 +132,80 @@ func resolveTenant(d UpDeps, project, database, instance string) (proj, dbName, } } if proj == "" { - return "", "", "", fmt.Errorf("no project in this workspace to snapshot") + return "", "", "", "", "", fmt.Errorf("no project in this workspace to snapshot") } if _, ok := d.Model.Projects[proj]; !ok { - return "", "", "", fmt.Errorf("project %q is not in this workspace", proj) + return "", "", "", "", "", fmt.Errorf("project %q is not in this workspace", proj) } inst = instance if inst == "" { var ok bool - inst, ok = ResolveInstance(d.Model, "postgres") + inst, ok = ResolveInstance(d.Model, engine) if !ok { - return "", "", "", fmt.Errorf("no shared postgres instance in this workspace (declare one under workspace.shared)") + return "", "", "", "", "", fmt.Errorf("no shared %s instance in this workspace (declare one under workspace.shared)", engine) } - } else if d.Model.Workspace.Shared[inst].Template != "postgres" { - return "", "", "", fmt.Errorf("shared instance %q is not a postgres engine", inst) + } else if d.Model.Workspace.Shared[inst].Template != engine { + return "", "", "", "", "", fmt.Errorf("shared instance %q is not a %s engine", inst, engine) } dbName = database if dbName == "" { - dbName = pgTenantDB(proj) + dbName = defaultTenant(engine, proj) } - return proj, dbName, inst, nil + return proj, dbName, inst, engine, label, nil } // tenantConn resolves the host-reachable admin endpoint for the tenant, reusing // the provision overlay (allocates/looks up the ledger port, applies the loopback -// overlay via compose up). Returns the ConnInfo the dumper connects with. -func tenantConn(ctx context.Context, d UpDeps, inst, dbName string) (db.ConnInfo, error) { - target, err := engineTarget(ctx, d, "postgres", inst) +// overlay via compose up). Returns the ConnInfo the engine's dumper connects with. +// Redis creds come from the instance params (auth-less by default → empty +// password, so no AUTH is sent); pg/minio use the resolved admin creds. +func tenantConn(ctx context.Context, d UpDeps, engine, inst, dbName string) (db.ConnInfo, error) { + target, err := engineTarget(ctx, d, engine, inst) if err != nil { return db.ConnInfo{}, err } - return db.ConnInfo{ - Host: target.Host, - Port: target.Port, - User: target.AdminEnv["user"], - Password: target.AdminEnv["password"], - Database: dbName, - }, nil + conn := db.ConnInfo{Host: target.Host, Port: target.Port, Database: dbName} + switch engine { + case "redis": + conn.User = "" + conn.Password = paramString(d.Model.Workspace.Shared[inst].Params, "rootPassword", "") + default: + conn.User = target.AdminEnv["user"] + conn.Password = target.AdminEnv["password"] + } + return conn, nil +} + +// SelectDumper picks the engine dumper for a snapshot kind, wired with the real +// external-tool runner (pg/redis) or the pure-Go S3 client (minio). The CLI calls +// this so `db snapshot`/`db restore` reach the right tool by target engine; tests +// inject a dumper directly. redis-cli/pg client absence surfaces via Preflight. +func SelectDumper(d UpDeps, kind string) (db.Dumper, error) { + engine, _, err := engineForKind(kind) + if err != nil { + return nil, err + } + runner := d.Runner + if runner == nil { + runner = docker.ExecRunner{} + } + switch engine { + case "postgres": + return db.PgDumper{Runner: runner}, nil + case "redis": + return db.RedisDumper{Runner: runner}, nil + case "minio": + return db.MinioDumper{}, nil + default: + return nil, fmt.Errorf("no dumper for engine %q", engine) + } } // Snapshot captures the project's tenant database to the workspace snapshot store // and records a ledger row. The dump streams OUTSIDE the flock; only the ledger // write is locked (spec 15). func Snapshot(ctx context.Context, d UpDeps, dumper db.Dumper, opt SnapshotOptions) (SnapshotMeta, error) { - proj, dbName, inst, err := resolveTenant(d, opt.Project, opt.Database, opt.Instance) + proj, dbName, inst, engine, label, err := resolveTenant(d, opt.Kind, opt.Project, opt.Database, opt.Instance) if err != nil { return SnapshotMeta{}, err } @@ -135,7 +217,7 @@ func Snapshot(ctx context.Context, d UpDeps, dumper db.Dumper, opt SnapshotOptio return SnapshotMeta{}, err } - conn, err := tenantConn(ctx, d, inst, dbName) + conn, err := tenantConn(ctx, d, engine, inst, dbName) if err != nil { return SnapshotMeta{}, err } @@ -144,7 +226,7 @@ func Snapshot(ctx context.Context, d UpDeps, dumper db.Dumper, opt SnapshotOptio if err := os.MkdirAll(dir, 0o755); err != nil { return SnapshotMeta{}, fmt.Errorf("create snapshot store: %w", err) } - dumpPath := filepath.Join(dir, name+".dump") + dumpPath := filepath.Join(dir, name+dumpExt(label)) // The dump PROCESS runs outside the flock (spec 15 — long-running). if err := dumper.Snapshot(ctx, conn, dumpPath); err != nil { @@ -156,7 +238,7 @@ func Snapshot(ctx context.Context, d UpDeps, dumper db.Dumper, opt SnapshotOptio return SnapshotMeta{}, err } meta := SnapshotMeta{ - Name: name, Project: proj, Kind: "pg", Instance: inst, Database: dbName, + Name: name, Project: proj, Kind: label, Instance: inst, Database: dbName, Digest: digest, Size: size, CreatedAt: time.Now().UTC().Format(time.RFC3339), Path: dumpPath, } if err := writeSidecar(dir, meta); err != nil { @@ -183,7 +265,7 @@ func Restore(ctx context.Context, d UpDeps, dumper db.Dumper, opt RestoreOptions if opt.Name == "" { return SnapshotMeta{}, fmt.Errorf("a snapshot name is required") } - proj, dbName, inst, err := resolveTenant(d, opt.Project, opt.Database, opt.Instance) + proj, dbName, inst, engine, _, err := resolveTenant(d, opt.Kind, opt.Project, opt.Database, opt.Instance) if err != nil { return SnapshotMeta{}, err } @@ -204,7 +286,7 @@ func Restore(ctx context.Context, d UpDeps, dumper db.Dumper, opt RestoreOptions return SnapshotMeta{}, fmt.Errorf("snapshot %q is corrupted: digest %s does not match recorded %s", opt.Name, digest, meta.Digest) } - conn, err := tenantConn(ctx, d, inst, dbName) + conn, err := tenantConn(ctx, d, engine, inst, dbName) if err != nil { return SnapshotMeta{}, err } diff --git a/internal/orchestrate/snapshot_engines_test.go b/internal/orchestrate/snapshot_engines_test.go new file mode 100644 index 0000000..a754d98 --- /dev/null +++ b/internal/orchestrate/snapshot_engines_test.go @@ -0,0 +1,263 @@ +package orchestrate + +import ( + "context" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "testing" + + "github.com/open-source-cloud/devstack/internal/config" + dbpkg "github.com/open-source-cloud/devstack/internal/db" + "github.com/open-source-cloud/devstack/internal/docker" + "github.com/open-source-cloud/devstack/internal/generate" + "github.com/open-source-cloud/devstack/internal/state" + "github.com/open-source-cloud/devstack/internal/template" + "github.com/open-source-cloud/devstack/internal/workspace" + "github.com/open-source-cloud/devstack/templates" +) + +// engineFixture builds UpDeps for a workspace whose single shared instance is the +// given engine (redis|minio), so the snapshot/restore engine-selection paths can +// be exercised with the mock docker client + a fake compose runner. +func engineFixture(t *testing.T, engine string) (UpDeps, *fakeRunner, *state.DB) { + t.Helper() + root := t.TempDir() + write := func(rel, body string) { + p := filepath.Join(root, rel) + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(p, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + } + shared := map[string]string{ + "redis": " redis: { template: redis, params: { version: \"7\" } }\n", + "minio": " minio: { template: minio, params: { rootUser: admin, rootPassword: secret } }\n", + }[engine] + write("workspace.yaml", "apiVersion: devstack/v1\nkind: Workspace\nname: demo\nshared:\n"+shared+"projects:\n - { name: app, path: app }\n") + write("app/devstack.yaml", "apiVersion: devstack/v1\nkind: Project\nname: app\nservices:\n web:\n template: node.vite\n") + + m, err := config.LoadAt(root) + if err != nil { + t.Fatalf("load: %v", err) + } + db, err := state.Open(context.Background(), filepath.Join(root, "state"), "ctx") + if err != nil { + t.Fatalf("state: %v", err) + } + t.Cleanup(func() { db.Close() }) + + mc := &docker.MockClient{ + Containers: []docker.Container{{ + ID: "eng1", Name: "devstack-shared-" + engine + "-1", State: "running", + Labels: map[string]string{generate.LabelManaged: "true", generate.LabelShared: engine}, + }}, + Details: map[string]docker.ContainerDetails{ + "eng1": {ID: "eng1", State: "running", Running: true, Health: docker.HealthHealthy}, + }, + } + src := template.NewFSSource(templates.FS) + lockPath := filepath.Join(root, "lock") + mgr := &workspace.Manager{Model: m, DB: db, Docker: mc, Source: src, LockPath: lockPath} + fr := &fakeRunner{} + d := UpDeps{ + Model: m, DB: db, Docker: mc, Manager: mgr, Source: src, + LockPath: lockPath, Runner: fr, Env: map[string]string{}, + } + return d, fr, db +} + +// redisRunner records argv and materializes the RDB file on `redis-cli --rdb`. +type redisRunner struct { + cmds [][]string + envs [][]string +} + +func (r *redisRunner) Run(_ context.Context, env []string, _, name string, args ...string) error { + r.cmds = append(r.cmds, append([]string{name}, args...)) + r.envs = append(r.envs, env) + if name == "redis-cli" { + for i, a := range args { + if a == "--rdb" && i+1 < len(args) { + _ = os.WriteFile(args[i+1], []byte("REDIS0011-fake-rdb"), 0o644) + } + } + } + return nil +} +func (r *redisRunner) Output(_ context.Context, env []string, _, name string, args ...string) ([]byte, error) { + r.cmds = append(r.cmds, append([]string{name}, args...)) + r.envs = append(r.envs, env) + return []byte("0\n"), nil +} +func (r *redisRunner) sawTool(tool string) []string { + for _, c := range r.cmds { + if c[0] == tool { + return c + } + } + return nil +} + +func TestSelectDumperByEngine(t *testing.T) { + d, _, _ := engineFixture(t, "redis") + for _, tc := range []struct { + kind string + want string // type name fragment + }{ + {"pg", "PgDumper"}, + {"postgres", "PgDumper"}, + {"redis", "RedisDumper"}, + {"minio", "MinioDumper"}, + {"s3", "MinioDumper"}, + {"", "PgDumper"}, + } { + got, err := SelectDumper(d, tc.kind) + if err != nil { + t.Fatalf("SelectDumper(%q): %v", tc.kind, err) + } + if name := typeName(got); !strings.Contains(name, tc.want) { + t.Errorf("SelectDumper(%q) = %s, want %s", tc.kind, name, tc.want) + } + } + if _, err := SelectDumper(d, "cassandra"); err == nil { + t.Error("SelectDumper should reject an unsupported engine") + } +} + +func typeName(v any) string { return fmt.Sprintf("%T", v) } + +// orchFakeS3 is an in-memory S3Snapshotter for the minio engine-selection test. +type orchFakeS3 struct { + objects map[string]map[string][]byte +} + +func newOrchFakeS3() *orchFakeS3 { + return &orchFakeS3{objects: map[string]map[string][]byte{}} +} +func (f *orchFakeS3) seed(bucket, key string, body []byte) { + if f.objects[bucket] == nil { + f.objects[bucket] = map[string][]byte{} + } + f.objects[bucket][key] = body +} +func (f *orchFakeS3) ListKeys(_ context.Context, bucket string) ([]string, error) { + var keys []string + for k := range f.objects[bucket] { + keys = append(keys, k) + } + sort.Strings(keys) + return keys, nil +} +func (f *orchFakeS3) Get(_ context.Context, bucket, key string) ([]byte, error) { + return f.objects[bucket][key], nil +} +func (f *orchFakeS3) Put(_ context.Context, bucket, key string, body []byte) error { + if f.objects[bucket] == nil { + f.objects[bucket] = map[string][]byte{} + } + f.objects[bucket][key] = append([]byte(nil), body...) + return nil +} + +func TestRedisSnapshotRestoreByEngine(t *testing.T) { + newSnapEnv(t) + d, _, ledger := engineFixture(t, "redis") + rr := &redisRunner{} + dumper := dbpkg.RedisDumper{Runner: rr, LookPath: func(string) (string, error) { return "/usr/bin/redis-cli", nil }} + + meta, err := Snapshot(context.Background(), d, dumper, SnapshotOptions{Kind: "redis", Project: "app", Name: "r1"}) + if err != nil { + t.Fatalf("Snapshot: %v", err) + } + if meta.Kind != "redis" { + t.Errorf("meta.Kind = %q, want redis", meta.Kind) + } + if !strings.HasSuffix(meta.Path, ".rdb") { + t.Errorf("redis dump path should end .rdb: %s", meta.Path) + } + rc := rr.sawTool("redis-cli") + if rc == nil { + t.Fatalf("redis-cli never ran: %v", rr.cmds) + } + joined := strings.Join(rc, " ") + for _, want := range []string{"-h 127.0.0.1", "--rdb"} { + if !strings.Contains(joined, want) { + t.Errorf("redis-cli argv missing %q: %s", want, joined) + } + } + // Ledger row recorded. + rows, _ := ledger.ProvisionedFor("app") + found := false + for _, r := range rows { + if r.Kind == snapshotKind && r.Name == "r1" { + found = true + } + } + if !found { + t.Errorf("snapshot ledger row not recorded: %v", rows) + } + + // Restore shells sh -c redis-cli --pipe. + if _, err := Restore(context.Background(), d, dumper, RestoreOptions{Kind: "redis", Project: "app", Name: "r1"}); err != nil { + t.Fatalf("Restore: %v", err) + } + sh := rr.sawTool("sh") + if sh == nil { + t.Fatalf("restore never shelled sh -c: %v", rr.cmds) + } + if !strings.Contains(strings.Join(sh, " "), "--pipe") { + t.Errorf("restore payload missing --pipe: %v", sh) + } +} + +func TestMinioSnapshotRestoreByEngine(t *testing.T) { + newSnapEnv(t) + d, _, ledger := engineFixture(t, "minio") + + src := newOrchFakeS3() + src.seed("app", "obj/one", []byte("payload-1")) + src.seed("app", "obj/two", []byte("payload-2")) + dumper := dbpkg.MinioDumper{Factory: func(context.Context, dbpkg.ConnInfo) (dbpkg.S3Snapshotter, error) { return src, nil }} + + meta, err := Snapshot(context.Background(), d, dumper, SnapshotOptions{Kind: "minio", Project: "app", Name: "m1"}) + if err != nil { + t.Fatalf("Snapshot: %v", err) + } + if meta.Kind != "minio" { + t.Errorf("meta.Kind = %q, want minio", meta.Kind) + } + if meta.Database != "app" { + t.Errorf("default minio tenant bucket = %q, want app", meta.Database) + } + if !strings.HasSuffix(meta.Path, ".tar") { + t.Errorf("minio dump path should end .tar: %s", meta.Path) + } + if _, err := os.Stat(meta.Path); err != nil { + t.Errorf("tar not written: %v", err) + } + rows, _ := ledger.ProvisionedFor("app") + found := false + for _, r := range rows { + if r.Kind == snapshotKind && r.Name == "m1" { + found = true + } + } + if !found { + t.Errorf("snapshot ledger row not recorded: %v", rows) + } + + // Restore into an empty target and confirm the two objects come back. + dst := newOrchFakeS3() + dumper2 := dbpkg.MinioDumper{Factory: func(context.Context, dbpkg.ConnInfo) (dbpkg.S3Snapshotter, error) { return dst, nil }} + if _, err := Restore(context.Background(), d, dumper2, RestoreOptions{Kind: "minio", Project: "app", Name: "m1"}); err != nil { + t.Fatalf("Restore: %v", err) + } + if got := string(dst.objects["app"]["obj/one"]); got != "payload-1" { + t.Errorf("restored obj/one = %q, want payload-1", got) + } +}