diff --git a/cmd/elastickv-snapshot-offload/main.go b/cmd/elastickv-snapshot-offload/main.go index ffbf34caf..6b25139b7 100644 --- a/cmd/elastickv-snapshot-offload/main.go +++ b/cmd/elastickv-snapshot-offload/main.go @@ -10,6 +10,7 @@ import ( "log/slog" "os" "sort" + "strconv" "strings" "github.com/bootjp/elastickv/internal/raftengine/etcd" @@ -59,6 +60,13 @@ type restoreConfig struct { manifestKey string dataDir string peerCSV string + // expectGroupRaw is the flag text; empty means the operator did not + // supply one, which is an error. Parsed into expectGroupID because + // group 0 is a real group and cannot double as "unset". + expectGroupRaw string + expectGroupID uint64 + expectCluster string + allowLegacySourceClusterless bool } func main() { @@ -123,7 +131,7 @@ func parsePublishFlags(argv []string) (*publishConfig, error) { fs.StringVar(&cfg.dataDir, "data-dir", "", "Source raft data directory containing a persisted snapshot (required)") fs.StringVar(&cfg.prefix, "prefix", "", "Object key prefix for published snapshot objects") fs.Uint64Var(&cfg.groupID, "group-id", 0, "Raft group ID recorded in the manifest") - fs.StringVar(&cfg.sourceCluster, "source-cluster", "", "Source cluster identifier (required for group 0 manifests)") + fs.StringVar(&cfg.sourceCluster, "source-cluster", "", "Source cluster identifier (required; restore matches it against the manifest)") fs.StringVar(&cfg.binaryVersion, "binary-version", "", "Binary version recorded in the manifest") fs.StringVar(&cfg.spoolDir, "spool-dir", "", "Temporary spool directory for the payload stream") if err := fs.Parse(argv); err != nil { @@ -135,8 +143,14 @@ func parsePublishFlags(argv []string) (*publishConfig, error) { if strings.TrimSpace(cfg.dataDir) == "" { return nil, errors.New("--data-dir is required") } - if cfg.groupID == 0 && strings.TrimSpace(cfg.sourceCluster) == "" { - return nil, errors.New("--source-cluster is required when --group-id is 0") + // Required for EVERY group, not just group 0. restore rejects an empty + // --expect-source-cluster outright and then compares it against the + // manifest's own value, so a manifest published without one can never + // match: a nonzero-group backup taken with this CLI would be + // unrestorable by it. Relaxing the restore side instead would give back + // the wrong-cluster restore that check exists to prevent. + if strings.TrimSpace(cfg.sourceCluster) == "" { + return nil, errors.New("--source-cluster is required") } if err := validateStoreFlags(cfg.store); err != nil { return nil, err @@ -152,6 +166,16 @@ func parseRestoreFlags(argv []string) (*restoreConfig, error) { fs.StringVar(&cfg.manifestKey, "manifest-key", "", "Object key of the snapshot manifest to restore (required)") fs.StringVar(&cfg.dataDir, "data-dir", "", "Fresh target raft data directory to create (required; must not already exist)") fs.StringVar(&cfg.peerCSV, "peers", "", "Comma-separated raft peers id=addr,id=addr (required)") + fs.StringVar(&cfg.expectCluster, "expect-source-cluster", "", + "Source cluster this data dir is for (required). The restore is refused if the "+ + "manifest was published by a different cluster -- which the group check alone "+ + "cannot catch when one bucket holds backups from several clusters.") + fs.StringVar(&cfg.expectGroupRaw, "expect-group", "", + "Raft group id this data dir is for (required). The restore is refused if the manifest "+ + "belongs to a different group, which is otherwise undetectable: nothing downstream "+ + "records the group, so the wrong group's FSM would start under this group's identity.") + fs.BoolVar(&cfg.allowLegacySourceClusterless, "allow-legacy-source-clusterless", false, + "allow an older nonzero-group manifest without source_cluster; source identity cannot be verified") if err := fs.Parse(argv); err != nil { return nil, errors.WithStack(err) } @@ -167,6 +191,17 @@ func parseRestoreFlags(argv []string) (*restoreConfig, error) { if strings.TrimSpace(cfg.peerCSV) == "" { return nil, errors.New("--peers is required") } + if strings.TrimSpace(cfg.expectGroupRaw) == "" { + return nil, errors.New("--expect-group is required") + } + if strings.TrimSpace(cfg.expectCluster) == "" { + return nil, errors.New("--expect-source-cluster is required") + } + groupID, err := strconv.ParseUint(strings.TrimSpace(cfg.expectGroupRaw), 10, 64) + if err != nil { + return nil, errors.Wrapf(err, "parse --expect-group %q", cfg.expectGroupRaw) + } + cfg.expectGroupID = groupID if err := validateStoreFlags(cfg.store); err != nil { return nil, err } @@ -263,11 +298,17 @@ func runRestore(ctx context.Context, cfg *restoreConfig, logger *slog.Logger) (r if err != nil { return err } + if cfg.allowLegacySourceClusterless { + logger.Warn("restoring with legacy source-cluster compatibility; manifest source identity may be unverifiable") + } result, err := snapshotoffload.RestorePhysicalSnapshot(ctx, snapshotoffload.RestoreOptions{ - Store: store, - ManifestKey: cfg.manifestKey, - DataDir: cfg.dataDir, - Peers: peers, + Store: store, + ManifestKey: cfg.manifestKey, + DataDir: cfg.dataDir, + Peers: peers, + ExpectGroupID: &cfg.expectGroupID, + ExpectSourceCluster: cfg.expectCluster, + AllowLegacySourceClusterless: cfg.allowLegacySourceClusterless, }) if err != nil { return errors.Wrap(err, "restore physical snapshot") diff --git a/cmd/elastickv-snapshot-offload/main_test.go b/cmd/elastickv-snapshot-offload/main_test.go index a9dedf22b..14d3c38e4 100644 --- a/cmd/elastickv-snapshot-offload/main_test.go +++ b/cmd/elastickv-snapshot-offload/main_test.go @@ -50,6 +50,10 @@ func TestSnapshotOffloadCLIPublishAndRestoreLocal(t *testing.T) { "--manifest-key", manifest.ManifestKey, "--data-dir", restoreDataDir, "--peers", "n2=127.0.0.1:12002", + // The publish above used --group-id 2, so this is the group this + // data dir is for. + "--expect-group", "2", + "--expect-source-cluster", "cluster-cli", }, io.Discard, logger) require.NoError(t, err) require.Equal(t, exitSuccess, code) @@ -67,6 +71,7 @@ func TestSnapshotOffloadCLIRequiresLocalRoot(t *testing.T) { "--store", storeLocal, "--data-dir", "data", "--group-id", "1", + "--source-cluster", "prod-tokyo", }, io.Discard, slog.New(slog.NewTextHandler(io.Discard, nil))) require.ErrorContains(t, err, "--local-root is required") require.Equal(t, exitUserErr, code) @@ -101,6 +106,7 @@ func TestSnapshotOffloadCLIS3KMSRequiresAWSKMS(t *testing.T) { "--s3-kms-key-id", "key-id", "--data-dir", "data", "--group-id", "1", + "--source-cluster", "prod-tokyo", }) require.NoError(t, err) @@ -111,6 +117,7 @@ func TestSnapshotOffloadCLIS3KMSRequiresAWSKMS(t *testing.T) { "--s3-kms-key-id", "key-id", "--data-dir", "data", "--group-id", "1", + "--source-cluster", "prod-tokyo", }) require.ErrorContains(t, err, "s3 KMS key id requires aws:kms encryption") } @@ -168,6 +175,14 @@ func TestClassifyErrorKeepsMissingSnapshotAsADataError(t *testing.T) { err: errors.Wrap(snapshotoffload.ErrInvalidOptions, "publish"), want: exitUserErr, }, + { + // The snapshot is intact; the operator named the wrong + // manifest for this data dir. Automation keys off the + // difference, so this must not be reported as bad data. + name: "wrong group's manifest", + err: errors.Wrap(snapshotoffload.ErrRestoreGroupMismatch, "restore"), + want: exitUserErr, + }, } for _, tc := range tests { @@ -177,3 +192,238 @@ func TestClassifyErrorKeepsMissingSnapshotAsADataError(t *testing.T) { }) } } + +// TestSnapshotOffloadCLIRestoreRefusesAnotherGroupsManifest is the regression +// test for a silent mis-restore. +// +// Nothing downstream of the restore records which group the data belongs to: +// the prepared artifacts carry index, term, peers and payload hash, and startup +// derives the group from the directory layout. So an operator repeating this +// command across groups and pasting the wrong manifest key produced a +// valid-looking directory that startup then loaded under a different group's +// routing identity, with no error at any point. --expect-group is the only +// place that mistake is detectable. +func TestSnapshotOffloadCLIRestoreRefusesAnotherGroupsManifest(t *testing.T) { + t.Parallel() + + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + root := t.TempDir() + objectRoot := filepath.Join(root, "objects") + sourceDataDir := seedCLISnapshot(t, root, []byte("EKVTHLC1group-2-payload"), 60, 9) + + var stdout bytes.Buffer + code, err := run(ctx, []string{ + commandPublish, + "--store", storeLocal, + "--local-root", objectRoot, + "--data-dir", sourceDataDir, + "--prefix", "cluster-cli", + "--group-id", "2", + "--source-cluster", "cluster-cli", + "--binary-version", "test-version", + }, &stdout, logger) + require.NoError(t, err) + require.Equal(t, exitSuccess, code) + + manifest, err := snapshotoffload.DecodeManifest(stdout.Bytes()) + require.NoError(t, err) + require.Equal(t, uint64(2), manifest.GroupID) + + restoreDataDir := filepath.Join(root, "restored-as-group-1") + code, err = run(ctx, []string{ + commandRestore, + "--store", storeLocal, + "--local-root", objectRoot, + "--manifest-key", manifest.ManifestKey, + "--data-dir", restoreDataDir, + "--peers", "n1=127.0.0.1:12001", + // The operator means group 1, but pasted group 2's manifest key. + "--expect-group", "1", + "--expect-source-cluster", "cluster-cli", + }, io.Discard, logger) + require.Error(t, err) + require.ErrorIs(t, err, snapshotoffload.ErrRestoreGroupMismatch) + // exitUserErr, not exitDataErr: the snapshot data is fine, the + // invocation named the wrong manifest. Automation distinguishes the two. + require.Equal(t, exitUserErr, code) + + // Nothing may be left behind: the check runs before the download and + // before the destination is created, so a mistaken key costs nothing. + _, statErr := os.Stat(restoreDataDir) + require.True(t, os.IsNotExist(statErr), + "a refused restore must not create the destination") +} + +// --expect-group is required, because defaulting it would silently accept +// whatever group the manifest happens to name -- which is the behaviour the +// flag exists to remove. +func TestSnapshotOffloadCLIRestoreRequiresAnExpectedGroup(t *testing.T) { + t.Parallel() + + _, err := parseRestoreFlags([]string{ + "--store", storeLocal, + "--local-root", "/tmp/objects", + "--manifest-key", "k", + "--data-dir", "/tmp/restored", + "--peers", "n1=127.0.0.1:12001", + }) + require.Error(t, err) + require.Contains(t, err.Error(), "--expect-group is required") +} + +// Group 0 is the dedicated TSO group, so it has to be accepted as an explicit +// value rather than treated as "unset". +func TestSnapshotOffloadCLIRestoreAcceptsGroupZero(t *testing.T) { + t.Parallel() + + cfg, err := parseRestoreFlags([]string{ + "--store", storeLocal, + "--local-root", "/tmp/objects", + "--manifest-key", "k", + "--data-dir", "/tmp/restored", + "--peers", "n1=127.0.0.1:12001", + "--expect-group", "0", + "--expect-source-cluster", "cluster-cli", + }) + require.NoError(t, err) + require.Equal(t, uint64(0), cfg.expectGroupID) +} + +func TestSnapshotOffloadCLIRestoreRejectsANonNumericGroup(t *testing.T) { + t.Parallel() + + _, err := parseRestoreFlags([]string{ + "--store", storeLocal, + "--local-root", "/tmp/objects", + "--manifest-key", "k", + "--data-dir", "/tmp/restored", + "--peers", "n1=127.0.0.1:12001", + "--expect-group", "one", + "--expect-source-cluster", "cluster-cli", + }) + require.Error(t, err) +} + +// TestSnapshotOffloadCLIRestoreRefusesAnotherClustersManifest covers the gap +// --expect-group alone leaves. +// +// One bucket can hold backups from several clusters, even under different +// prefixes. A manifest from another cluster for the SAME group id passes the +// group check, and the restore then produces a structurally valid directory +// that startup loads as this cluster's FSM. Nothing downstream records the +// source cluster, so this is the only place the mistake is detectable. +func TestSnapshotOffloadCLIRestoreRefusesAnotherClustersManifest(t *testing.T) { + t.Parallel() + + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + root := t.TempDir() + objectRoot := filepath.Join(root, "objects") + sourceDataDir := seedCLISnapshot(t, root, []byte("EKVTHLC1other-cluster-payload"), 60, 9) + + var stdout bytes.Buffer + code, err := run(ctx, []string{ + commandPublish, + "--store", storeLocal, + "--local-root", objectRoot, + "--data-dir", sourceDataDir, + "--prefix", "shared-bucket", + "--group-id", "1", + "--source-cluster", "cluster-b", + "--binary-version", "test-version", + }, &stdout, logger) + require.NoError(t, err) + require.Equal(t, exitSuccess, code) + + manifest, err := snapshotoffload.DecodeManifest(stdout.Bytes()) + require.NoError(t, err) + require.Equal(t, "cluster-b", manifest.SourceCluster) + require.Equal(t, uint64(1), manifest.GroupID) + + restoreDataDir := filepath.Join(root, "restored-into-cluster-a") + code, err = run(ctx, []string{ + commandRestore, + "--store", storeLocal, + "--local-root", objectRoot, + "--manifest-key", manifest.ManifestKey, + "--data-dir", restoreDataDir, + "--peers", "n1=127.0.0.1:12001", + // The group matches, so only the cluster identity can catch this. + "--expect-group", "1", + "--expect-source-cluster", "cluster-a", + }, io.Discard, logger) + require.Error(t, err) + require.ErrorIs(t, err, snapshotoffload.ErrRestoreSourceClusterMismatch) + require.Equal(t, exitUserErr, code) + + _, statErr := os.Stat(restoreDataDir) + require.True(t, os.IsNotExist(statErr), + "a refused restore must not create the destination") +} + +func TestSnapshotOffloadCLIRestoreRequiresAnExpectedSourceCluster(t *testing.T) { + t.Parallel() + + _, err := parseRestoreFlags([]string{ + "--store", storeLocal, + "--local-root", "/tmp/objects", + "--manifest-key", "k", + "--data-dir", "/tmp/restored", + "--peers", "n1=127.0.0.1:12001", + "--expect-group", "1", + }) + require.Error(t, err) + require.Contains(t, err.Error(), "--expect-source-cluster is required") +} + +func TestSnapshotOffloadCLIParsesLegacySourceClusterlessOverride(t *testing.T) { + t.Parallel() + + cfg, err := parseRestoreFlags([]string{ + "--store", storeLocal, + "--local-root", "/tmp/objects", + "--manifest-key", "legacy/v1/groups/1/snapshots/manifest.json", + "--data-dir", "/tmp/restored", + "--peers", "n1=127.0.0.1:12001", + "--expect-group", "1", + "--expect-source-cluster", "cluster-a", + "--allow-legacy-source-clusterless", + }) + require.NoError(t, err) + require.True(t, cfg.allowLegacySourceClusterless) +} + +// TestSnapshotOffloadCLIPublishRequiresASourceClusterForEveryGroup pins that +// publish and restore agree on this flag. +// +// It used to be required only for group 0. restore rejects an empty +// --expect-source-cluster outright and then compares it against the manifest's +// own SourceCluster, so a manifest published without one can never match: a +// nonzero-group backup taken with this CLI was unrestorable by it. Relaxing +// the restore side instead would give back the wrong-cluster restore that +// check exists to prevent. +func TestSnapshotOffloadCLIPublishRequiresASourceClusterForEveryGroup(t *testing.T) { + t.Parallel() + + for _, group := range []string{"0", "1", "7"} { + _, err := parsePublishFlags([]string{ + "--store", storeLocal, + "--local-root", "/tmp/objects", + "--data-dir", "data", + "--group-id", group, + }) + require.ErrorContains(t, err, "--source-cluster is required", + "group %s must not be able to publish an unrestorable manifest", group) + + cfg, err := parsePublishFlags([]string{ + "--store", storeLocal, + "--local-root", "/tmp/objects", + "--data-dir", "data", + "--group-id", group, + "--source-cluster", "prod-tokyo", + }) + require.NoError(t, err) + require.Equal(t, "prod-tokyo", cfg.sourceCluster) + } +} diff --git a/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md b/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md index a3d43143a..60f98e2bd 100644 --- a/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md +++ b/docs/design/2026_07_19_partial_physical_snapshot_object_offload.md @@ -1,9 +1,9 @@ # Physical Snapshot Object Offload -Status: Partial — M0/M1/M2 implemented; M3 retention/GC implemented, remaining M3 items pending +Status: Partial — M0/M1/M2 implemented; M3 publication and retention runtime, corruption drills, operations runbook and bounded three-node/single-group local recovery acceptance implemented; broader operational acceptance pending Author: bootjp Date: 2026-07-19 -Updated: 2026-07-23 +Updated: 2026-10-22 ## 1. Scope @@ -48,7 +48,13 @@ The M1 object-store-neutral substrate now adds: - `cmd/elastickv-snapshot-offload publish` and `restore` for local and S3-backed operator workflows. -The runtime scheduler remains pending. Retention/GC and corruption tests are implemented; the remaining M3 items (restore drills, multi-node acceptance, operator documentation) are pending. +The runtime scheduler is implemented and wired into main.go, opt-in via +`--snapshotOffloadBucket` (or `--snapshotOffloadLocalDir`). Retention/GC and +corruption drills from PR #1222 are integrated. GC requires a separate explicit +rollout gate; publication alone never enables deletion. A bounded three-node, +single-data-group local recovery drill has passed; remote-S3, multi-group and +long-duration operational acceptance remain outside that receipt. The runbook is at +[`../snapshot_offload_operations.md`](../snapshot_offload_operations.md). ## 2. Safety boundary @@ -111,6 +117,10 @@ upload at a time per process. Jitter spreads multi-group work. Snapshot creation cadence remains owned by the Raft engine; object offload never forces an extra state-machine snapshot in this milestone. +Each publication has a configurable deadline (`--snapshotOffloadPublishTimeout`, +default 10 minutes). Deadline expiry releases the upload slot, records a failed +attempt, and lets the scheduler retry on a later scan. + ## 5. Retention and reference lifecycle Retention is per group and combines a minimum generation count with a time @@ -221,8 +231,13 @@ enable the verifier. Before rollback or restarting any legacy publisher, make the verifier fail, stop and drain any active GC pass, and only then resume the legacy publisher. Claims cannot protect a publisher that ignores them; a process-local build version or successful claim acquisition is not evidence -that the rollout is complete. Runtime verifier wiring is part of the separate -runtime integration work; `NewGC` refuses a missing verifier. +that the rollout is complete. The runtime supplies the verifier only after the +operator sets `--snapshotOffloadGCClaimsRolloutComplete`; `NewGC` refuses a +missing verifier. GC remains disabled unless `--snapshotOffloadGCEnabled` is +also set. Passes run serially with configurable interval, timeout and policy. +Shutdown cancels and drains both workers before closing their shared store. +The rollout assertion covers standalone publishers as well as Raft members; +it is not automatic cluster capability discovery. Storage-envelope encryption protects values but not all physical keys and metadata. The external bucket therefore requires private ACLs, TLS, and @@ -238,8 +253,8 @@ applicable KMS decrypt permission). |---|---|---| | M0 | Persisted snapshot export handle, complete-payload restore preparation, focused design | Implemented in the first substrate PR | | M1 | Object client interface, S3-compatible implementation, immutable payload/manifest publication, download verification, operator CLI | Implemented: local and S3 stores, manifest schema, payload-first publish, verified restore, and publish/restore CLI | -| M2 | Leader-only per-group scheduler, metrics, jitter, concurrency bounds, cancellation and restart idempotency | Implemented: `internal/snapshotoffload/scheduler.go`. Leadership is checked before the snapshot is opened and re-checked immediately before the manifest commit via `PublishOptions.VerifyLeader`; uploads are bounded (default one per process) with interval jitter; cancellation is treated as shutdown rather than publish failure; restart idempotency comes from the object store, since publish reuses a matching committed manifest. Not yet wired into `main.go` — the runtime flags are M3. | -| M3 | Retention/GC, restore drills, corruption tests, multi-node acceptance, operational documentation | Partially implemented: the §5 two-phase retention/GC (`retention.go`) with cross-process publish/GC claims and `RetentionStore` list/delete on both the local and S3 stores. Restore corruption drills are implemented (`restore_corruption_test.go`: truncated, over-length, missing and tampered-descriptor payloads, plus a positive restore-into-fresh-dir drill). Versioned buckets are rejected unless the operator explicitly confirms noncurrent-version expiration. Multi-node acceptance and the complete operator runbook remain pending. | +| M2 | Leader-only per-group scheduler, metrics, jitter, concurrency bounds, cancellation and restart idempotency | Implemented: `internal/snapshotoffload/scheduler.go`. Leadership is checked before the snapshot is opened and re-checked immediately before the manifest commit via `PublishOptions.VerifyLeader`; uploads are bounded (default one per process) with interval jitter; cancellation is treated as shutdown rather than publish failure; restart idempotency comes from the object store, since publish reuses a matching committed manifest. Runtime wiring and flags shipped in the M3 slice below. | +| M3 | Retention/GC, restore drills, corruption tests, multi-node acceptance, operational documentation | Partially implemented: publication runtime, opt-in gated GC using the §5 implementation from #1222, metrics and operations runbook. Corruption drills cover truncated, over-length, missing and tampered payloads. GC validates versioning and requires explicit claims rollout verification. Bounded three-node/single-data-group acceptance passes periodic publication, surviving-quorum writes, actual GC and isolated replacement recovery with local FSM/adapter values and applied-index checks. Remote-S3, multi-group and long-duration acceptance are not established by that test. | The filename and header remain `partial` until M1-M3 complete the central object-offload subsystem. At that point the completion PR must use `git mv` to @@ -266,6 +281,16 @@ old path is absent before requesting review. - A multi-node drill restores each group from external objects into fresh data directories and passes adapter reads after normal Raft startup. +The bounded local acceptance drill uses the existing three-node gRPC bootstrap +fixture with one data group and an independent durable filesystem object root. +It must observe scheduler-created generations (not seeded snapshot files), keep +committing writes with one replica stopped, reclaim old generations through the +runtime GC, and restore the retained generation into an isolated replacement +cluster with fresh directories and addresses. Acceptance checks adapter values, +Raft/FSM applied indexes and a new replicated write after restore. This narrow +drill does not establish long-duration stability, remote-S3 behavior or every +multi-group topology; its execution receipt belongs in the operations runbook. + ## 10. Intentional non-goals - Parsing or duplicating the SST ingest stream owned by #1130. diff --git a/docs/snapshot_offload_operations.md b/docs/snapshot_offload_operations.md new file mode 100644 index 000000000..1db8b5a7c --- /dev/null +++ b/docs/snapshot_offload_operations.md @@ -0,0 +1,378 @@ +# Physical Snapshot Object Offload — Operations + +Runbook for the physical snapshot offload subsystem: continuous backup of +Raft snapshots to an S3-compatible object store, and disaster recovery from +those artifacts. + +Design: [`design/2026_07_19_partial_physical_snapshot_object_offload.md`](design/2026_07_19_partial_physical_snapshot_object_offload.md). + +Dependency order: merge #1245, then #1222, then this runtime integration (#1224). +Retention reuses #1222's GC and claim protocol. Publication and reclamation +have separate opt-ins; configuring a backup destination never enables deletion. + +## Scope + +Use this runbook to: + +1. enable continuous snapshot offload on a cluster, +2. verify that backups are actually being produced, +3. restore a node from a published snapshot, +4. configure retention, and understand what it will and will not delete. + +This is **physical** backup: it ships the Raft snapshot the engine already +produced. It does not force an extra state-machine snapshot, so backup +freshness is bounded by the engine's own snapshot cadence. + +## 1. What gets written + +Two object kinds under the configured prefix: + +``` +/v1/groups//snapshots/-.json manifest +/v1/payloads/sha256//.fsm payload +``` + +Payloads are **content-addressed and shared**: two groups (or two generations) +whose snapshots hash identically converge on one object. This matters for +retention — see §4. + +Manifests are self-hashing. Reusing a matching manifest advances its generation +under a claim; snapshot identity and payload remain unchanged. + +## 2. Enabling offload + +Offload is opt-in. A node with no destination configured does no offload work +and cannot fail startup on offload settings. + +```bash +elastickv \ + --snapshotOffloadBucket=my-backup-bucket \ + --snapshotOffloadRegion=ap-northeast-1 \ + --snapshotOffloadSourceCluster=prod-tokyo \ + --snapshotOffloadPrefix=elastickv \ + --snapshotOffloadServerSideEncryption=aws:kms \ + --snapshotOffloadSSEKMSKeyId=arn:aws:kms:ap-northeast-1:123456789012:key/abcd +``` + +| Flag | Meaning | +|---|---| +| `--snapshotOffloadBucket` | S3 bucket. Enables offload. | +| `--snapshotOffloadLocalDir` | Filesystem root instead of S3. **Mutually exclusive** with the bucket. | +| `--snapshotOffloadSourceCluster` | Cluster identity recorded in every manifest. Required. | +| `--snapshotOffloadPrefix` | Key prefix for all artifacts. | +| `--snapshotOffloadRegion` / `--snapshotOffloadEndpoint` / `--snapshotOffloadProfile` / `--snapshotOffloadForcePathStyle` | S3 addressing and credentials. | +| `--snapshotOffloadServerSideEncryption` / `--snapshotOffloadSSEKMSKeyId` | `AES256` or `aws:kms`. KMS aliases are rejected; pass an ARN or bare key ID. | +| `--snapshotOffloadInterval` | Scan cadence. Default 15m. | +| `--snapshotOffloadJitter` | Spread across groups. Default: a quarter of the interval. | +| `--snapshotOffloadConcurrency` | Concurrent uploads per process. Default 1. | +| `--snapshotOffloadPublishTimeout` | Deadline for each publication, including spooling and store operations. Default 10m; must be positive. | +| `--snapshotOffloadSpoolDir` | Where payloads are spooled before upload. Needs room for the largest snapshot. | + +**A misconfigured offload refuses to start the node.** That is deliberate: an +operator who configured a backup destination and silently received no backups +is worse off than one whose node failed loudly. + +### Security requirements + +The bucket holds physical keys and metadata. Storage-envelope encryption +protects *values*, not all keys and metadata, so the bucket itself must be +protected: + +- private ACLs — anonymous read or write is a deployment failure, +- TLS, +- server-side encryption (SSE-S3 or SSE-KMS), +- credentials scoped to `list`/`get`/`put`/`delete` **below the prefix only**, +- secrets supplied by file or environment, never in process arguments. + +## 3. Verifying that backups are happening + +Only the current leader of a group publishes; followers skip. On a healthy +three-node group, exactly one node reports publishes and two report +`not_leader`. + +```promql +# Backup freshness — the number that matters. Alert if it stops advancing. +elastickv_snapshot_offload_last_published_index + +# Backups are failing. Any sustained rate is paging-grade. +rate(elastickv_snapshot_offload_failed_total[15m]) + +# Routine skips. Expected on followers and unchanged snapshots. +rate(elastickv_snapshot_offload_skipped_total[15m]) +``` + +Skip reasons and what they mean: + +| Reason | Meaning | Action | +|---|---|---| +| `not_leader` | This node does not lead the group. | None — expected on followers. | +| `already_published` | Snapshot unchanged since this process last published it. | None. | +| `no_persisted_snapshot` | The group has not produced a snapshot yet. | None on a young cluster. Investigate if it persists on a busy group. | +| `already_in_flight` | Another scan is publishing this group. | None. | +| `object_claimed` | A publisher or GC holds the same object claim. | Retry on the next scan; inspect persistent orphan claims before recovering ownership. | +| `leadership_unknown` | Engine unavailable, typically during shutdown. | None if the node is stopping. | + +**A group whose `last_published_index` never advances has no backups**, even +though nothing is failing. Alert on staleness, not only on errors. + +## 4. Retention + +GC is off by default. Upgrade every server and standalone CLI publisher sharing +the destination/prefix to the claims protocol, stop and drain all legacy +publications, then enable one designated retention worker with both: + +```text +--snapshotOffloadGCEnabled=true +--snapshotOffloadGCClaimsRolloutComplete=true +``` + +The second setting is an explicit operator assertion about the entire +destination/prefix, not automatic membership discovery. Without it, enabling GC +fails startup. The runtime supplies #1222's mandatory verifier before every GC +pass. Before starting any legacy publisher or rolling back, disable GC, stop +and drain its worker, and only then resume legacy publication. Stopping the +server cancels publication and retention, waits for both workers, then closes +the local store descriptor. + +| Flag | Default | Meaning | +|---|---|---| +| `--snapshotOffloadGCInterval` | `1h` | Delay after each serial GC pass. | +| `--snapshotOffloadGCTimeout` | `10m` | Deadline for a pass; failures log and retry on the next interval. | +| `--snapshotOffloadGCMinGenerations` | `3` | Minimum retained generations per group. | +| `--snapshotOffloadGCMaxAge` | `336h` | Retain manifests inside this age window too. | +| `--snapshotOffloadGCPayloadGrace` | `24h` | Delay before an unreferenced payload can be marked. | +| `--snapshotOffloadGCMinMarkAge` | `24h` | Delay between marking and reclaiming an unchanged payload. | +| `--snapshotOffloadAllowVersionedWithLifecycle` | `false` | Assert noncurrent-version expiration is configured on a versioned bucket. | + +All policy values, interval and timeout must be positive. GC also requires an +offload destination. Startup validates bucket policy when GC is enabled; every +pass validates it again before scanning or deleting. Read-only restore and +publication-only workers do not require `s3:GetBucketVersioning`. + +Retention is per group, and runs in two phases. + +**Phase 1 — manifests.** Keeps `MinGenerations` newest per group plus anything +inside `MaxAge`, and always keeps a group's newest valid manifest regardless of +both. A group can never be left with no restore point. + +**Phase 2 — payloads.** Rebuilds the live set from **every surviving manifest +in the whole prefix** — not per group, because payloads are shared — then +reclaims only unreferenced objects, using **two-pass mark-and-sweep**: a pass +marks an eligible payload, and only a later pass, with the object unchanged and +the mark older than `MinMarkAge`, deletes it. + +Publishers and GC coordinate with storage-visible object claims because +content-preserving reuse cannot be detected by a general-purpose S3 ETag +precondition. The two-pass delay adds protection against stale evidence. +**`MinMarkAge` must exceed your longest plausible publish.** + +Retention refuses to delete anything when it cannot prove the live set: + +- a malformed manifest anywhere in the prefix → payload reclamation is skipped + entirely, and the malformed object is preserved for inspection, +- a listing or pagination failure → no deletes at all, +- an object under the payload prefix that does not parse as a payload key → + left alone. + +If `PayloadPhaseSkipped` is set with malformed manifests reported, fix or +remove the malformed object; storage will not be reclaimed until you do. + +### Versioned buckets + +Retention deletes by key. On a bucket with **S3 versioning enabled**, a keyed +delete only writes a delete marker: the bytes survive as a noncurrent version +that later listings cannot see, so GC reports successful reclamation while +storage grows without bound. + +GC refuses Enabled/Suspended versioning unless noncurrent-version expiration +has been configured and `--snapshotOffloadAllowVersionedWithLifecycle=true` +acknowledges it. Unknown status or missing `s3:GetBucketVersioning` permission +aborts GC. Restore requires object reads and applicable KMS decrypt permission, +not lifecycle acknowledgement or bucket-versioning access. + +## 5. Restore + +Restore is **offline** and targets an **absent** data directory. It refuses to +overwrite an existing one — that guard is what protects an operator who +mistakenly points a restore at a live node. + +```bash +# 1. Find the generation to restore. +elastickv-snapshot-offload publish --help # same store flags as below + +# 2. Restore each group into ITS OWN directory (see the path rule below). +elastickv-snapshot-offload restore \ + --store=s3 --s3-bucket=my-backup-bucket --s3-region=ap-northeast-1 \ + --manifest-key='elastickv/v1/groups/1/snapshots/00000000000000004211-00000000000000000007.json' \ + --data-dir=/var/lib/elastickv/n1/group-1 \ + --expect-group=1 \ + --expect-source-cluster=prod-tokyo \ + --peers='n1=10.0.0.1:50051,n2=10.0.0.2:50051,n3=10.0.0.3:50051' + +# 3. Repeat for every group the node hosts, then start it normally. +``` + +### Every flag that names the target must name the SAME target + +`--manifest-key`, `--data-dir`, `--expect-group`, `--expect-source-cluster` +and `--peers` are five independent statements about what you are restoring. +Most of them used to be uncheckable against each other: + +- **`--expect-source-cluster` is required.** The group check is not enough + when one bucket holds backups from several clusters, even under + different prefixes: another cluster's manifest for the *same* group id + passes it, and the restore produces a valid-looking directory that + startup loads as this cluster's FSM. Nothing downstream records the + source cluster, so this is the only place that mistake is detectable. + It must match the `--source-cluster` the publish used. + +- **`--expect-group` is required and is the only cross-check.** Nothing in + the restored directory records which group the data came from — the + artifacts carry index, term, peers and payload hash, and startup derives + the group from the directory layout. So pasting group 2's manifest key + into a group-1 restore produced a valid-looking directory that startup + then loaded as group 1: the wrong physical FSM under another group's + routing identity, with no error anywhere. The restore is now refused, + before the download and before the destination is created, so a mistaken + key costs nothing and leaves nothing behind. +- **`--peers` must be that group's peer map, not group 1's.** Each group + has its own listener addresses from `--raftGroups` / `--raftGroupPeers`, + and restore persists the supplied peers into that group's data + directory. Copying the `:50051` endpoints from the example above into + every invocation leaves the other groups trying to reach the wrong Raft + endpoints, and they never form a quorum. Re-derive the peer list per + group the same way you re-derive the manifest key and the data dir. + +### The `--data-dir` path must match what the server will open + +`--data-dir` is the **per-group** directory, not the node's `--raftDir`. +The server derives it as: + +The rule the server actually applies (`groupDataDir`) is: + +- **group 0 always** gets `//group-0`. +- **every other group** gets `group-` **only when the node hosts more + than one _data_ group**, and `/` otherwise. + +"More than one data group" is the exact condition, because +`dataGroupsNeedMultiDirs` counts data groups and **excludes group 0**: + +| Deployment (`--raftGroups`) | Group | Directory | +|---|---|---| +| two or more data groups, e.g. `1,2` | any group *G* | `//group-` | +| a single data group, e.g. `1` | 1 | `/` | +| dedicated TSO **plus one** data group, e.g. `0,1` | 0 | `//group-0` | +| dedicated TSO **plus one** data group, e.g. `0,1` | 1 | `/` — **not** `group-1` | +| dedicated TSO plus two or more data groups, e.g. `0,1,2` | any group *G* | `//group-` | + +The fourth row is the one that catches people out. A node running the +dedicated TSO group alongside a single data group *looks* multi-group — +two entries in `--raftGroups` — but only group 0 is parked under +`group-0`; the data group still opens `/` directly. +Restoring it into `group-1` puts the data where startup never looks. + +That is the general failure in both directions: a directory the server +does not open is not an error, it is an empty group. Startup finds +nothing, the restore is silently ignored, and the node comes back with +only the groups you happened to place correctly. **Check the table per +group before each invocation** rather than assuming one rule for the +whole node. + +### Transport and spool hygiene + +- **Snapshot offload requests must use HTTPS.** The node validates an explicit + `--snapshotOffloadEndpoint` and also rejects plaintext URLs resolved by the + AWS SDK configuration chain. Redirects must preserve both HTTPS and the + endpoint host. For a local MinIO during development, pass + `--snapshotOffloadAllowInsecureEndpoint` explicitly; host-changing redirects + remain prohibited. The AWS SDK's CA bundle and credential-provider settings + are preserved. +- **Stale spool files are cleaned at startup.** A publish spools the + payload to `/.snapshot-offload-spool` (or + `--snapshotOffloadSpoolDir`) and removes it on completion; a process + killed mid-publish leaves the file behind. Offload startup removes + leftovers older than 24 hours before scheduling. Use separate spool + directories when a publish can take longer than 24 hours. Budget the spool volume for + one in-flight payload per concurrent upload + (`--snapshotOffloadConcurrency`). + +Restore verifies exact length and SHA-256 before the payload is accepted, then +fsyncs and atomically renames it into place. Any integrity failure leaves the +destination **absent** rather than half-written. + +Target membership (`--peers`) is explicit operator input, not copied from the +source. That is what makes recovery onto replacement addresses possible, while +the source membership stays in the manifest for audit. + +Exit codes: `0` success, `1` invalid invocation, `2` missing or invalid +snapshot data. Automation should distinguish these. + +Backups created by older versions may have an empty `source_cluster` for +nonzero groups. They remain fail-closed by default because their source cannot +be verified. After independently verifying provenance, an operator may add +`--allow-legacy-source-clusterless` to one restore invocation; the command +logs a warning and still requires `--expect-source-cluster`. + +## 6. Failure modes + +| Symptom | Cause | Action | +|---|---|---| +| `last_published_index` frozen, no failures | Node is not the leader, or the engine has produced no new snapshot. | Confirm which node leads the group; check the engine's snapshot cadence. | +| Sustained `failed_total` | Object store unreachable, credentials expired, bucket policy denies writes. | Check the scheduler's log line — it carries the error the metric deliberately omits. | +| Storage grows despite retention | Versioned bucket without a lifecycle rule (§4), or reclamation blocked by a malformed manifest. | Add the lifecycle rule; inspect reported malformed manifests. | +| Restore fails with an integrity error | Payload truncated, over-length, or the manifest was edited. | Restore an older generation; the destination was left absent, so nothing was damaged. | +| Restore refuses to run | Destination directory already exists. | Restore into a fresh path. Never delete a live data dir to make room. | + +## 7. Limits + +- Backup freshness is bounded by the Raft engine's snapshot cadence; offload + never forces an extra snapshot. +- Losing a group's leadership mid-publish can leave an unreferenced payload, + which retention reclaims. It can never leave a committed manifest. +- Mark state is per-process and in memory. A restart delays reclamation by one + pass; it never advances it. + +## 8. Bounded local recovery acceptance + +`TestSnapshotOffloadThreeNodeRecoveryAcceptance` reuses the real bootstrap +fixture, not mocked leadership or seeded snapshot files. It runs three etcd/Raft +nodes with Pebble FSMs, loopback gRPC adapters and one data group. The durable +filesystem object root is independent of all source and replacement data dirs. + +The test starts the actual periodic offload runtime on the source nodes, +observes a committed generation, stops one follower, and keeps writing through +the surviving quorum. It restarts the publishers and enables runtime retention +on one worker with the claims-rollout acknowledgement. Retention must actually +remove the old manifest **and** its payload; writes continue after reclamation. +No direct publication, forced snapshot, fake timestamp or direct deletion is +used. A few checkpoint writes advance the fixture's normal snapshot threshold +past the asserted user data; offload itself never forces a snapshot. + +After stopping all publishers/GC and source nodes, the test reopens the object +root and restores the retained latest generation into absent data dirs for an +isolated three-node cluster with new IDs and addresses. Group/source identity, +target peer metadata and payload SHA are checked. All replacement nodes must +expose the data in their **local FSM**, report Raft and durable FSM applied indexes +at least the manifest index, pass RawKV adapter reads, and apply a new replicated +write. Local checks are necessary because an adapter can forward to the leader. + +Reproduce with bounded concurrency and a hard test-process deadline: + +```sh +GOMAXPROCS=4 go test -race . \ + -run '^TestSnapshotOffloadThreeNodeRecoveryAcceptance$' \ + -count=3 -timeout=3m -v +``` + +Local receipt, 2026-10-23 JST, Go 1.26.6/darwin-arm64: all three race runs passed +in 28.769 seconds total. Each observed periodic snapshot indexes `17 -> 33 -> 49`, +reclaimed the first generation, and verified 36 user keys (in addition to +checkpoint writes). Each replacement restored index 49, initially reported +Raft applied index 50 and durable FSM applied index 49, and read all 37 asserted +keys after the post-restore replicated write. Per-run elapsed times were +9.37, 8.53 and 8.99 seconds. + +All data, object roots and sockets are ephemeral/local; the test does not +deploy, contact production, delete production data or provision paid cloud +resources. It is **not** a 30-day burn-in, remote-S3 acceptance, a multi-group/TSO +topology drill or operational closure. The design remains `partial`. diff --git a/internal/snapshotoffload/manifest.go b/internal/snapshotoffload/manifest.go index a215a3c7a..c2ba37f11 100644 --- a/internal/snapshotoffload/manifest.go +++ b/internal/snapshotoffload/manifest.go @@ -19,10 +19,23 @@ const ( var ( ErrInvalidOptions = errors.New("snapshot offload: invalid options") - ErrIntegrity = errors.New("snapshot offload: integrity check failed") - ErrObjectConflict = errors.New("snapshot offload: object conflict") - ErrObjectNotFound = errors.New("snapshot offload: object not found") - ErrObjectClaimed = errors.New("snapshot offload: object claimed by another operation") + + // ErrRestoreGroupMismatch reports a restore whose manifest belongs to a + // different Raft group than the operator named. It is deliberately its + // own sentinel: an operator repeating the restore command across groups + // needs to see "wrong group", not a generic invalid-options error. + ErrRestoreGroupMismatch = errors.New("snapshot offload: manifest group does not match the requested group") + + // ErrRestoreSourceClusterMismatch reports a restore whose manifest was + // published by a different cluster than the operator named. Its own + // sentinel, like the group mismatch: "another cluster's backup" and + // "another group's backup" are different operator mistakes with + // different fixes. + ErrRestoreSourceClusterMismatch = errors.New("snapshot offload: manifest source cluster does not match the requested cluster") + ErrIntegrity = errors.New("snapshot offload: integrity check failed") + ErrObjectConflict = errors.New("snapshot offload: object conflict") + ErrObjectNotFound = errors.New("snapshot offload: object not found") + ErrObjectClaimed = errors.New("snapshot offload: object claimed by another operation") // ErrObjectModified is returned by DeleteObjectIfUnmodified when // the object changed after the caller validated it. For retention diff --git a/internal/snapshotoffload/offload_test.go b/internal/snapshotoffload/offload_test.go index 5189fee75..801ab7e12 100644 --- a/internal/snapshotoffload/offload_test.go +++ b/internal/snapshotoffload/offload_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "fmt" "io" "os" "path/filepath" @@ -59,6 +60,8 @@ func TestPublishAndRestorePhysicalSnapshotRoundTrip(t *testing.T) { Peers: []etcdraftengine.Peer{ {NodeID: 9, ID: "n9", Address: "127.0.0.1:19009"}, }, + ExpectGroupID: expectGroup(manifest.GroupID), + ExpectSourceCluster: manifest.SourceCluster, }) require.NoError(t, err) require.Equal(t, int64(len(payload)), result.PayloadBytes) @@ -96,10 +99,12 @@ func TestRestoreRejectsCorruptPayloadAndLeavesDestinationAbsent(t *testing.T) { restoreDataDir := filepath.Join(root, "restored") _, err = RestorePhysicalSnapshot(ctx, RestoreOptions{ - Store: store, - ManifestKey: manifest.ManifestKey, - DataDir: restoreDataDir, - Peers: singlePeer(), + Store: store, + ManifestKey: manifest.ManifestKey, + DataDir: restoreDataDir, + Peers: singlePeer(), + ExpectGroupID: expectGroup(manifest.GroupID), + ExpectSourceCluster: manifest.SourceCluster, }) require.ErrorIs(t, err, ErrIntegrity) _, statErr := os.Stat(restoreDataDir) @@ -317,26 +322,30 @@ func TestDecodeManifestAcceptsJointConsensusConfState(t *testing.T) { require.Equal(t, manifest.ConfState, decoded.ConfState) } -func TestPublishRejectsGroupZeroWithoutSourceClusterBeforePayload(t *testing.T) { - ctx := context.Background() - root := t.TempDir() - payload := []byte("EKVTHLC1payload-invalid-group-zero") - sourceDataDir := seedPhysicalSnapshot(t, root, payload, 18, 11, singlePeer()) - store := newTestLocalStore(t, filepath.Join(root, "objects")) - - _, err := PublishPersistedSnapshot(ctx, PublishOptions{ - Store: store, - DataDir: sourceDataDir, - Prefix: "cluster-a", - GroupID: 0, - }) - require.ErrorIs(t, err, ErrInvalidOptions) - - payloadKey, err := payloadKey("cluster-a", hexSHA256Bytes(payload)) - require.NoError(t, err) - _, ok, err := store.HeadObject(ctx, payloadKey) - require.NoError(t, err) - require.False(t, ok) +func TestPublishRejectsEveryGroupWithoutSourceClusterBeforePayload(t *testing.T) { + for _, groupID := range []uint64{0, 7} { + t.Run(fmt.Sprintf("group-%d", groupID), func(t *testing.T) { + ctx := context.Background() + root := t.TempDir() + payload := []byte("EKVTHLC1payload-invalid-source-cluster") + sourceDataDir := seedPhysicalSnapshot(t, root, payload, 18, 11, singlePeer()) + store := newTestLocalStore(t, filepath.Join(root, "objects")) + + _, err := PublishPersistedSnapshot(ctx, PublishOptions{ + Store: store, + DataDir: sourceDataDir, + Prefix: "cluster-a", + GroupID: groupID, + }) + require.ErrorIs(t, err, ErrInvalidOptions) + + payloadKey, err := payloadKey("cluster-a", hexSHA256Bytes(payload)) + require.NoError(t, err) + _, ok, err := store.HeadObject(ctx, payloadKey) + require.NoError(t, err) + require.False(t, ok) + }) + } } func TestPublishUsesDataDirLocalSpoolByDefault(t *testing.T) { @@ -405,10 +414,12 @@ func TestRestoreInlineManifestRejectsStaleSelfHashBeforePayloadDownload(t *testi tracked := &countingObjectStore{ObjectStore: store} _, err = RestorePhysicalSnapshot(ctx, RestoreOptions{ - Store: tracked, - Manifest: &tampered, - DataDir: filepath.Join(root, "restored"), - Peers: singlePeer(), + Store: tracked, + Manifest: &tampered, + DataDir: filepath.Join(root, "restored"), + Peers: singlePeer(), + ExpectGroupID: expectGroup(tampered.GroupID), + ExpectSourceCluster: tampered.SourceCluster, }) require.ErrorIs(t, err, ErrIntegrity) require.Zero(t, tracked.getObjectCalls) @@ -509,10 +520,12 @@ func TestRestorePreflightsExistingDestinationBeforePayloadDownload(t *testing.T) require.NoError(t, os.Mkdir(restoreDataDir, 0o755)) _, err = RestorePhysicalSnapshot(ctx, RestoreOptions{ - Store: store, - ManifestKey: manifest.ManifestKey, - DataDir: restoreDataDir, - Peers: singlePeer(), + Store: store, + ManifestKey: manifest.ManifestKey, + DataDir: restoreDataDir, + Peers: singlePeer(), + ExpectGroupID: expectGroup(manifest.GroupID), + ExpectSourceCluster: manifest.SourceCluster, }) require.ErrorIs(t, err, etcdraftengine.ErrExternalSnapshotRestoreExists) } @@ -540,6 +553,8 @@ func TestRestoreRejectsInvalidPeersBeforePayloadDownload(t *testing.T) { Peers: []etcdraftengine.Peer{ {NodeID: 0, ID: "n0", Address: "127.0.0.1:12000"}, }, + ExpectGroupID: expectGroup(manifest.GroupID), + ExpectSourceCluster: manifest.SourceCluster, }) require.ErrorIs(t, err, ErrInvalidOptions) require.Zero(t, tracked.getObjectCalls) @@ -563,10 +578,12 @@ func TestRestoreHonorsCancelledContextBeforePayloadDownload(t *testing.T) { tracked := &countingObjectStore{ObjectStore: store} _, err = RestorePhysicalSnapshot(ctx, RestoreOptions{ - Store: tracked, - Manifest: manifest, - DataDir: filepath.Join(root, "restored"), - Peers: singlePeer(), + Store: tracked, + Manifest: manifest, + DataDir: filepath.Join(root, "restored"), + Peers: singlePeer(), + ExpectGroupID: expectGroup(manifest.GroupID), + ExpectSourceCluster: manifest.SourceCluster, }) require.ErrorIs(t, err, context.Canceled) require.Zero(t, tracked.getObjectCalls) @@ -806,3 +823,208 @@ func TestPublishVerifiesLeadershipAfterTheManifestAbsenceProbe(t *testing.T) { require.NotContains(t, verifiedAfter, "put-manifest", "and before the manifest object is created") } + +// expectGroup is the RestoreOptions.ExpectGroupID helper. The field is a +// pointer because group 0 is a real group (the dedicated TSO group), so zero +// cannot double as "not supplied". +func expectGroup(groupID uint64) *uint64 { + return &groupID +} + +// TestCleanStaleSpoolFilesRemovesOrphansFromAKilledProcess covers the spool +// leak. +// +// spoolExport removes its file with a defer, which never runs if the process is +// killed or the host loses power mid-publish. Nothing else removed them, so +// repeated crashes during large snapshots accumulated full payload copies until +// the spool volume filled. +func TestCleanStaleSpoolFilesRemovesOrphansFromAKilledProcess(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + orphan := filepath.Join(dir, "elastickv-snapshot-offload-123456.fsm") + require.NoError(t, os.WriteFile(orphan, []byte("payload"), 0o600)) + + // Files that are not spool files must be left alone: the spool dir can be + // shared, and deleting an unrelated file would be far worse than the leak. + unrelated := filepath.Join(dir, "keep-me.txt") + require.NoError(t, os.WriteFile(unrelated, []byte("data"), 0o600)) + manifest := filepath.Join(dir, "manifest.json") + require.NoError(t, os.WriteFile(manifest, []byte("{}"), 0o600)) + + removed, err := CleanStaleSpoolFiles(dir, time.Now().Add(time.Minute)) + require.NoError(t, err) + require.Equal(t, []string{orphan}, removed) + + require.NoFileExists(t, orphan) + require.FileExists(t, unrelated) + require.FileExists(t, manifest) +} + +// A file newer than the cutoff may belong to a concurrently running publish, so +// it must survive: deleting it would break a live upload to fix a leak. +func TestCleanStaleSpoolFilesSparesAFileNewerThanTheCutoff(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + inFlight := filepath.Join(dir, "elastickv-snapshot-offload-inflight.fsm") + require.NoError(t, os.WriteFile(inFlight, []byte("partial"), 0o600)) + + removed, err := CleanStaleSpoolFiles(dir, time.Now().Add(-time.Hour)) + require.NoError(t, err) + require.Empty(t, removed) + require.FileExists(t, inFlight) +} + +// A missing spool dir is not an error: offload may never have run. +func TestCleanStaleSpoolFilesToleratesAMissingDirectory(t *testing.T) { + t.Parallel() + + removed, err := CleanStaleSpoolFiles(filepath.Join(t.TempDir(), "absent"), time.Now()) + require.NoError(t, err) + require.Empty(t, removed) +} + +func TestCleanStaleSpoolFilesRequiresADirectory(t *testing.T) { + t.Parallel() + + _, err := CleanStaleSpoolFiles(" ", time.Now()) + require.ErrorIs(t, err, ErrInvalidOptions) +} + +// TestSpoolDirForMatchesWhatAPublishUses pins that a caller cleaning the spool +// dir targets the same directory a publish writes into -- a cleaner pointed at +// the wrong path silently does nothing. +func TestSpoolDirForMatchesWhatAPublishUses(t *testing.T) { + t.Parallel() + + require.Equal(t, + filepath.Join("/srv", "raft", ".snapshot-offload-spool"), + SpoolDirFor("", "/srv/raft/n1"), + "the default spool dir is a sibling of the data dir, so it shares its filesystem "+ + "and the spooled payload can be renamed rather than copied") + require.Equal(t, filepath.Clean("/spool"), SpoolDirFor("/spool", "/srv/raft/n1"), + "an explicit spool dir wins") +} + +// TestRestoreRejectsAnotherClustersManifestBeforeTheDownload pins that the +// cluster check, like the group check, runs before the payload download and +// before the destination is created. +func TestRestoreRejectsAnotherClustersManifestBeforeTheDownload(t *testing.T) { + t.Parallel() + + ctx := context.Background() + root := t.TempDir() + payload := []byte("EKVTHLC1payload-for-cross-cluster-restore") + sourceDataDir := seedPhysicalSnapshot(t, root, payload, 31, 11, singlePeer()) + store := newTestLocalStore(t, filepath.Join(root, "objects")) + manifest, err := PublishPersistedSnapshot(ctx, PublishOptions{ + Store: store, + DataDir: sourceDataDir, + Prefix: "shared-bucket", + GroupID: 1, + SourceCluster: "cluster-b", + }) + require.NoError(t, err) + + tracked := &countingObjectStore{ObjectStore: store} + dest := filepath.Join(root, "restored") + _, err = RestorePhysicalSnapshot(ctx, RestoreOptions{ + Store: tracked, + Manifest: manifest, + DataDir: dest, + Peers: singlePeer(), + // The group MATCHES; only the cluster identity differs, which is the + // case the group check alone cannot catch. + ExpectGroupID: expectGroup(manifest.GroupID), + ExpectSourceCluster: "cluster-a", + AllowLegacySourceClusterless: true, + }) + require.ErrorIs(t, err, ErrRestoreSourceClusterMismatch) + require.Zero(t, tracked.getObjectCalls, + "the payload must not be downloaded for a manifest from another cluster") + _, statErr := os.Stat(dest) + require.True(t, os.IsNotExist(statErr), "the destination must not be created") +} + +func TestRestoreLegacySourceClusterlessManifestRequiresExplicitOverride(t *testing.T) { + t.Parallel() + + ctx := context.Background() + root := t.TempDir() + payload := []byte("EKVTHLC1legacy-source-clusterless-payload") + sourceDataDir := seedPhysicalSnapshot(t, root, payload, 32, 12, singlePeer()) + store := newTestLocalStore(t, filepath.Join(root, "objects")) + manifest, err := PublishPersistedSnapshot(ctx, PublishOptions{ + Store: store, + DataDir: sourceDataDir, + Prefix: "legacy", + GroupID: 1, + SourceCluster: "cluster-a", + }) + require.NoError(t, err) + + legacy := *manifest + legacy.SourceCluster = "" + _, legacy.ManifestSHA256, err = legacy.MarshalCanonical() + require.NoError(t, err) + + base := RestoreOptions{ + Store: store, + Manifest: &legacy, + Peers: singlePeer(), + ExpectGroupID: expectGroup(legacy.GroupID), + ExpectSourceCluster: "cluster-a", + } + withoutOverride := base + withoutOverride.DataDir = filepath.Join(root, "without-override") + _, err = RestorePhysicalSnapshot(ctx, withoutOverride) + require.ErrorIs(t, err, ErrRestoreSourceClusterMismatch) + + withOverride := base + withOverride.DataDir = filepath.Join(root, "with-override") + withOverride.AllowLegacySourceClusterless = true + _, err = RestorePhysicalSnapshot(ctx, withOverride) + require.NoError(t, err) +} + +// TestCleanStaleSpoolFilesTreatsTheDirectoryAsAPathNotAPattern is the glob +// regression. +// +// filepath.Glob interprets the whole joined path as a pattern, so a configured +// spool dir containing a metacharacter — /tmp/spool[12] — matched /tmp/spool1 +// and /tmp/spool2 and deleted the spool files under THOSE, while the directory +// the operator actually configured was never examined. A configured path is a +// path. +func TestCleanStaleSpoolFilesTreatsTheDirectoryAsAPathNotAPattern(t *testing.T) { + t.Parallel() + + root := t.TempDir() + // The configured directory, whose name contains glob metacharacters. + configured := filepath.Join(root, "spool[12]") + require.NoError(t, os.MkdirAll(configured, 0o750)) + mine := filepath.Join(configured, "elastickv-snapshot-offload-mine.fsm") + require.NoError(t, os.WriteFile(mine, []byte("mine"), 0o600)) + + // Directories the pattern would have matched instead. Their files must be + // untouched: they belong to someone else. + var bystanders []string + for _, name := range []string{"spool1", "spool2"} { + dir := filepath.Join(root, name) + require.NoError(t, os.MkdirAll(dir, 0o750)) + f := filepath.Join(dir, "elastickv-snapshot-offload-theirs.fsm") + require.NoError(t, os.WriteFile(f, []byte("theirs"), 0o600)) + bystanders = append(bystanders, f) + } + + removed, err := CleanStaleSpoolFiles(configured, time.Now().Add(time.Minute)) + require.NoError(t, err) + require.Equal(t, []string{mine}, removed, + "only the configured directory's own spool file may be removed") + + require.NoFileExists(t, mine) + for _, f := range bystanders { + require.FileExists(t, f, + "a directory that merely MATCHES the configured name is not the configured directory") + } +} diff --git a/internal/snapshotoffload/publish.go b/internal/snapshotoffload/publish.go index cfede7ffd..2c5522c8e 100644 --- a/internal/snapshotoffload/publish.go +++ b/internal/snapshotoffload/publish.go @@ -376,13 +376,112 @@ func validatePublishOptions(opts PublishOptions) error { return errors.Wrap(ErrInvalidOptions, "object store is required") case stringsTrim(opts.DataDir) == "": return errors.Wrap(ErrInvalidOptions, "data dir is required") - case opts.GroupID == 0 && stringsTrim(opts.SourceCluster) == "": - return errors.Wrap(ErrInvalidOptions, "source cluster is required for group 0 manifests") + case stringsTrim(opts.SourceCluster) == "": + return errors.Wrap(ErrInvalidOptions, "source cluster is required") default: return nil } } +// spoolFilePattern is the glob matching spool files spoolExport creates. +// Exported via CleanStaleSpoolFiles so a caller can identify them without +// duplicating the name. +const spoolFilePattern = "elastickv-snapshot-offload-*.fsm" + +// CleanStaleSpoolFiles removes spool files left behind by a previous process. +// +// spoolExport removes its file with a defer, which never runs if the process +// is killed or the host loses power mid-publish. Nothing else removed them, so +// repeated crashes during large snapshots accumulated full payload copies +// until the spool volume filled. +// +// olderThan guards against deleting a file a CONCURRENTLY RUNNING publish is +// still writing: only files whose mtime predates the cutoff are removed. A +// caller at startup must still use a real stale-age cutoff because another +// process may share the directory and be publishing concurrently. +// +// Returns the files removed. Individual failures are collected rather than +// aborting, because one undeletable file must not leave the rest to +// accumulate. +func CleanStaleSpoolFiles(spoolDir string, olderThan time.Time) ([]string, error) { + spoolDir = stringsTrim(spoolDir) + if spoolDir == "" { + return nil, errors.Wrap(ErrInvalidOptions, "spool dir is required") + } + // The directory is READ, not globbed. filepath.Glob interprets the whole + // joined path as a pattern, so a configured spool dir containing a glob + // metacharacter -- /tmp/spool[12] -- would match /tmp/spool1 and + // /tmp/spool2 and delete files under THOSE, while the real + // /tmp/spool[12] was never examined. A configured path is a path, never a + // pattern; only the entry names are matched against the pattern. + spoolDir = filepath.Clean(spoolDir) + entries, err := os.ReadDir(spoolDir) + if err != nil { + if os.IsNotExist(err) { + // Offload may never have run here. + return nil, nil + } + return nil, errors.Wrapf(err, "read spool dir %s", spoolDir) + } + var ( + removed []string + failures []error + examined int + ) + for _, entry := range entries { + matched, err := filepath.Match(spoolFilePattern, entry.Name()) + if err != nil { + return removed, errors.Wrapf(err, "match spool pattern %q", spoolFilePattern) + } + if !matched { + continue + } + examined++ + path := filepath.Join(spoolDir, entry.Name()) + gone, err := removeStaleSpoolFile(path, olderThan) + if err != nil { + failures = append(failures, err) + continue + } + if gone { + removed = append(removed, path) + } + } + if len(failures) > 0 { + return removed, errors.Wrapf(errors.Join(failures...), + "clean spool dir %s: %d of %d files failed", spoolDir, len(failures), examined) + } + return removed, nil +} + +// removeStaleSpoolFile removes one spool file if it is a regular file older +// than the cutoff, reporting whether it was removed. +// +// A file that vanished between the glob and here is not a failure: another +// process's cleanup, or the owning publish finishing, is the expected race. +func removeStaleSpoolFile(path string, olderThan time.Time) (bool, error) { + info, err := os.Stat(path) + if err != nil { + if os.IsNotExist(err) { + return false, nil + } + return false, errors.Wrapf(err, "stat spool file %s", path) + } + if !info.Mode().IsRegular() || !info.ModTime().Before(olderThan) { + return false, nil + } + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return false, errors.Wrapf(err, "remove spool file %s", path) + } + return true, nil +} + +// SpoolDirFor reports the spool directory a publish with these options uses, +// so a caller can clean it without re-deriving the default. +func SpoolDirFor(spoolDir, dataDir string) string { + return publishSpoolDir(PublishOptions{SpoolDir: spoolDir, DataDir: dataDir}) +} + func publishSpoolDir(opts PublishOptions) string { if stringsTrim(opts.SpoolDir) != "" { return filepath.Clean(opts.SpoolDir) @@ -394,7 +493,7 @@ func spoolExport(ctx context.Context, export *etcdraftengine.PersistedSnapshotEx if err := os.MkdirAll(spoolDir, localStoreDirPerm); err != nil { return nil, "", 0, errors.WithStack(err) } - tmp, err := os.CreateTemp(spoolDir, "elastickv-snapshot-offload-*.fsm") + tmp, err := os.CreateTemp(spoolDir, spoolFilePattern) if err != nil { return nil, "", 0, errors.WithStack(err) } diff --git a/internal/snapshotoffload/restore.go b/internal/snapshotoffload/restore.go index 0a45e819b..c00036813 100644 --- a/internal/snapshotoffload/restore.go +++ b/internal/snapshotoffload/restore.go @@ -20,6 +20,39 @@ type RestoreOptions struct { Manifest *Manifest DataDir string Peers []etcdraftengine.Peer + + // ExpectGroupID is the Raft group the operator believes this data + // directory belongs to. Required. + // + // Nothing downstream carries the group's identity: the restored + // artifacts record index, term, peers and payload hash, but the group + // comes only from the manifest, and startup derives the group from the + // directory layout instead. So a group-2 manifest restored into a + // group-1 data directory produces a perfectly valid-looking + // directory that startup then loads as group 1 -- the wrong physical + // FSM under another group's routing identity, with no error anywhere. + // The only place that mistake can be caught is here, against what the + // operator says they intended. + // + // A pointer because group 0 is a real group (the dedicated TSO group), + // so zero cannot double as "unset". + ExpectGroupID *uint64 + + // ExpectSourceCluster is the cluster the operator believes this manifest + // came from. Required. + // + // The group check alone is not enough when one bucket holds backups from + // several clusters, even under different prefixes: another cluster's + // manifest for the SAME group id passes it, and the restore produces a + // structurally valid directory that startup then loads as this cluster's + // FSM. Nothing downstream records the source cluster either, so this is + // the only place the mistake is detectable. + ExpectSourceCluster string + + // AllowLegacySourceClusterless permits restoring a legacy nonzero-group + // manifest created before source-cluster became mandatory. The caller must + // opt in explicitly because the manifest cannot prove its source cluster. + AllowLegacySourceClusterless bool } const ( @@ -89,6 +122,16 @@ func prepareRestorePayload(ctx context.Context, opts RestoreOptions) (Manifest, if err := validateManifest(manifest); err != nil { return Manifest{}, "", nil, err } + // Before the download and before the destination exists, so a + // mistaken manifest key costs nothing and leaves nothing behind. + if err := checkRestoreGroup(manifest, opts.ExpectGroupID); err != nil { + return Manifest{}, "", nil, err + } + if err := checkRestoreSourceCluster( + manifest, opts.ExpectSourceCluster, opts.AllowLegacySourceClusterless, + ); err != nil { + return Manifest{}, "", nil, err + } if err := checkRestorePreflight(ctx, opts.DataDir); err != nil { return Manifest{}, "", nil, err } @@ -210,11 +253,48 @@ func validateRestoreOptions(opts RestoreOptions) error { return errors.Wrap(ErrInvalidOptions, "data dir is required") case len(opts.Peers) == 0: return errors.Wrap(ErrInvalidOptions, "restore peers are required") + case opts.ExpectGroupID == nil: + return errors.Wrap(ErrInvalidOptions, "expected raft group id is required") + case stringsTrim(opts.ExpectSourceCluster) == "": + return errors.Wrap(ErrInvalidOptions, "expected source cluster is required") default: return validateRestorePeers(opts.Peers) } } +// checkRestoreGroup rejects a manifest belonging to a different group than +// the operator asked to restore. +func checkRestoreGroup(manifest Manifest, expect *uint64) error { + if expect == nil { + return errors.Wrap(ErrInvalidOptions, "expected raft group id is required") + } + if manifest.GroupID != *expect { + return errors.Wrapf(ErrRestoreGroupMismatch, + "manifest %s belongs to group %d, not the requested group %d", + manifest.ManifestKey, manifest.GroupID, *expect) + } + return nil +} + +// checkRestoreSourceCluster rejects a manifest published by a different +// cluster than the operator named. +func checkRestoreSourceCluster(manifest Manifest, expect string, allowLegacySourceClusterless bool) error { + expect = stringsTrim(expect) + if expect == "" { + return errors.Wrap(ErrInvalidOptions, "expected source cluster is required") + } + manifestSource := stringsTrim(manifest.SourceCluster) + if manifestSource == "" && manifest.GroupID != 0 && allowLegacySourceClusterless { + return nil + } + if manifestSource != expect { + return errors.Wrapf(ErrRestoreSourceClusterMismatch, + "manifest %s was published by cluster %q, not the requested %q", + manifest.ManifestKey, manifest.SourceCluster, expect) + } + return nil +} + func validateRestorePeers(peers []etcdraftengine.Peer) error { seenNodeIDs := make(map[uint64]struct{}, len(peers)) seenIDs := make(map[string]struct{}, len(peers)) diff --git a/internal/snapshotoffload/restore_corruption_test.go b/internal/snapshotoffload/restore_corruption_test.go index e773645e4..1a5d6c48e 100644 --- a/internal/snapshotoffload/restore_corruption_test.go +++ b/internal/snapshotoffload/restore_corruption_test.go @@ -38,10 +38,11 @@ func publishForRestoreDrill(t *testing.T, root string, payload []byte) (*LocalSt func requireRestoreFailsClosed(t *testing.T, store *LocalStore, manifestKey, dataDir string) error { t.Helper() _, err := RestorePhysicalSnapshot(context.Background(), RestoreOptions{ - Store: store, - ManifestKey: manifestKey, - DataDir: dataDir, - Peers: singlePeer(), + Store: store, + ManifestKey: manifestKey, + DataDir: dataDir, + Peers: singlePeer(), + ExpectGroupID: expectGroup(1), ExpectSourceCluster: "cluster-a", }) require.Error(t, err) _, statErr := os.Stat(dataDir) @@ -169,10 +170,11 @@ func TestRestoreDrillSucceedsIntoAFreshDirectory(t *testing.T) { restoreDataDir := filepath.Join(root, "restored") result, err := RestorePhysicalSnapshot(context.Background(), RestoreOptions{ - Store: store, - ManifestKey: manifest.ManifestKey, - DataDir: restoreDataDir, - Peers: singlePeer(), + Store: store, + ManifestKey: manifest.ManifestKey, + DataDir: restoreDataDir, + Peers: singlePeer(), + ExpectGroupID: expectGroup(1), ExpectSourceCluster: "cluster-a", }) require.NoError(t, err) require.NotNil(t, result) @@ -182,10 +184,11 @@ func TestRestoreDrillSucceedsIntoAFreshDirectory(t *testing.T) { // than overwrite: the preflight is what protects an operator who // re-runs the drill against a live node's data dir. _, err = RestorePhysicalSnapshot(context.Background(), RestoreOptions{ - Store: store, - ManifestKey: manifest.ManifestKey, - DataDir: restoreDataDir, - Peers: singlePeer(), + Store: store, + ManifestKey: manifest.ManifestKey, + DataDir: restoreDataDir, + Peers: singlePeer(), + ExpectGroupID: expectGroup(1), ExpectSourceCluster: "cluster-a", }) require.Error(t, err, "restore must refuse a destination that already exists") } diff --git a/internal/snapshotoffload/s3_store.go b/internal/snapshotoffload/s3_store.go index 98234f642..c9cfe29f9 100644 --- a/internal/snapshotoffload/s3_store.go +++ b/internal/snapshotoffload/s3_store.go @@ -22,7 +22,7 @@ import ( const ( s3MetadataSHA256 = "elastickv-sha256" - s3LoadConfigOptionCapHint = 4 + s3LoadConfigOptionCapHint = 5 s3ConditionalWriteRetries = 3 s3MaxSinglePutBytes = int64(5 * 1024 * 1024 * 1024) s3DefaultMultipartPart = int64(64 * 1024 * 1024) @@ -58,6 +58,7 @@ type S3StoreConfig struct { ServerSideEncryption string SSEKMSKeyID string DisableChecksumHeaders bool + HTTPClientDecorator func(aws.HTTPClient) (aws.HTTPClient, error) // AllowVersionedBucketWithLifecycle is an explicit operator assertion // that the bucket has a noncurrent-version expiration lifecycle. The // GC rejects Enabled/Suspended versioning by default because keyed @@ -86,16 +87,11 @@ func NewS3Store(ctx context.Context, cfg S3StoreConfig) (*S3Store, error) { } client := cfg.Client if client == nil { - awsCfg, err := loadS3AWSConfig(ctx, cfg) + var err error + client, err = newS3ObjectClient(ctx, cfg) if err != nil { return nil, err } - client = s3.NewFromConfig(awsCfg, func(o *s3.Options) { - o.UsePathStyle = cfg.ForcePathStyle - if stringsTrim(cfg.Endpoint) != "" { - o.BaseEndpoint = aws.String(stringsTrim(cfg.Endpoint)) - } - }) } bucket := stringsTrim(cfg.Bucket) return &S3Store{ @@ -110,6 +106,26 @@ func NewS3Store(ctx context.Context, cfg S3StoreConfig) (*S3Store, error) { }, nil } +func newS3ObjectClient(ctx context.Context, cfg S3StoreConfig) (*s3.Client, error) { + awsCfg, err := loadS3AWSConfig(ctx, cfg) + if err != nil { + return nil, err + } + if cfg.HTTPClientDecorator != nil { + client, err := cfg.HTTPClientDecorator(awsCfg.HTTPClient) + if err != nil { + return nil, errors.Wrap(err, "configure s3 http client") + } + awsCfg.HTTPClient = client + } + return s3.NewFromConfig(awsCfg, func(o *s3.Options) { + o.UsePathStyle = cfg.ForcePathStyle + if stringsTrim(cfg.Endpoint) != "" { + o.BaseEndpoint = aws.String(stringsTrim(cfg.Endpoint)) + } + }), nil +} + // ValidateRetention checks versioning immediately before GC, without imposing // bucket-policy permissions or lifecycle requirements on read-only restores. func (s *S3Store) ValidateRetention(ctx context.Context) error { diff --git a/internal/snapshotoffload/s3_store_test.go b/internal/snapshotoffload/s3_store_test.go index f1a0d1754..cbe54407a 100644 --- a/internal/snapshotoffload/s3_store_test.go +++ b/internal/snapshotoffload/s3_store_test.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "math" + "net/http" "sort" "strings" "sync" @@ -49,11 +50,36 @@ func TestPublishAndRestorePhysicalSnapshotRoundTripWithS3Store(t *testing.T) { Peers: []etcdraftengine.Peer{ {NodeID: 2, ID: "n2", Address: "127.0.0.1:12002"}, }, + ExpectGroupID: expectGroup(manifest.GroupID), + ExpectSourceCluster: manifest.SourceCluster, }) require.NoError(t, err) require.Equal(t, manifest.Payload.SHA256, result.PayloadSHA256) } +func TestS3StoreUsesConfiguredHTTPClientOnlyForS3(t *testing.T) { + client := &http.Client{} + cfg := S3StoreConfig{ + Bucket: "snapshots", + Region: "us-east-1", + ServerSideEncryption: "AES256", + AccessKeyID: "test-access-key", + SecretAccessKey: "test-secret-key", + HTTPClientDecorator: func(aws.HTTPClient) (aws.HTTPClient, error) { + return client, nil + }, + } + awsCfg, err := loadS3AWSConfig(context.Background(), cfg) + require.NoError(t, err) + require.NotEqual(t, client, awsCfg.HTTPClient, + "credential providers must retain their own transport for IMDS and ECS") + store, err := NewS3Store(context.Background(), cfg) + require.NoError(t, err) + s3Client, ok := store.client.(*s3.Client) + require.True(t, ok) + require.Same(t, client, s3Client.Options().HTTPClient) +} + func TestS3RestoreDoesNotRequireBucketVersioningPermission(t *testing.T) { t.Parallel() for _, status := range []types.BucketVersioningStatus{"", types.BucketVersioningStatusEnabled, "FutureVersioningMode"} { @@ -73,6 +99,7 @@ func TestS3RestoreDoesNotRequireBucketVersioningPermission(t *testing.T) { readerStore := newTestS3Store(t, client) result, err := RestorePhysicalSnapshot(ctx, RestoreOptions{ Store: readerStore, ManifestKey: manifest.ManifestKey, DataDir: root + "/restored", + ExpectGroupID: expectGroup(3), ExpectSourceCluster: "cluster-s3", Peers: []etcdraftengine.Peer{{NodeID: 2, ID: "n2", Address: "127.0.0.1:12002"}}, }) require.NoError(t, err) diff --git a/internal/snapshotoffload/scheduler.go b/internal/snapshotoffload/scheduler.go index cb165ce0a..118a291da 100644 --- a/internal/snapshotoffload/scheduler.go +++ b/internal/snapshotoffload/scheduler.go @@ -32,6 +32,8 @@ type Scheduler struct { observer SchedulerObserver logger *slog.Logger now func() time.Time + + publishTimeout time.Duration // published caches the highest index this process has published per group. // It is an optimisation only: restart idempotency comes from the object // store, not from this map. @@ -78,12 +80,19 @@ func (nopSchedulerObserver) ObserveSnapshotOffloadFailed(uint64, error) // Default scheduling parameters from §4. const ( - DefaultSchedulerInterval = 15 * time.Minute - DefaultSchedulerConcurrency = 1 + DefaultSchedulerInterval = 15 * time.Minute + DefaultSchedulerConcurrency = 1 + DefaultSchedulerPublishTimeout = 10 * time.Minute ) type SchedulerOption func(*Scheduler) +// WithSchedulerPublishTimeout bounds one publication, including store requests +// and spooling, so a stalled group cannot prevent later scans. +func WithSchedulerPublishTimeout(d time.Duration) SchedulerOption { + return func(s *Scheduler) { s.publishTimeout = d } +} + func WithSchedulerInterval(d time.Duration) SchedulerOption { return func(s *Scheduler) { if d > 0 { @@ -160,6 +169,8 @@ func NewScheduler(store PublishStore, groups []OffloadGroup, prefix, sourceClust now: time.Now, published: make(map[uint64]uint64), inFlight: make(map[uint64]struct{}), + + publishTimeout: DefaultSchedulerPublishTimeout, } for _, opt := range opts { opt(s) @@ -191,6 +202,8 @@ func (s *Scheduler) validate() error { return errors.Wrap(ErrInvalidOptions, "snapshot offload scheduler requires an object store") case s.sourceName == "": return errors.Wrap(ErrInvalidOptions, "snapshot offload scheduler requires a source cluster name") + case s.publishTimeout <= 0: + return errors.Wrap(ErrInvalidOptions, "snapshot offload publication timeout must be positive") } // Both leadership callbacks are mandatory. Treating a nil callback // as "leader" would let a miswired scheduler publish from a @@ -335,7 +348,9 @@ func (s *Scheduler) publishGroupBounded(ctx context.Context, group OffloadGroup) return } defer s.endGroup(group.GroupID) - s.publishGroup(ctx, group) + bounded, cancel := context.WithTimeout(ctx, s.publishTimeout) + defer cancel() + s.publishGroup(bounded, group) } // staggeredGroup pairs a group with the absolute instant its work may @@ -400,7 +415,7 @@ func (s *Scheduler) publishGroup(ctx context.Context, group OffloadGroup) { SkipIfNotNewerThan: s.publishedIndex(group.GroupID), }) if err != nil { - if errors.Is(err, context.Canceled) || ctx.Err() != nil { + if errors.Is(err, context.Canceled) || errors.Is(ctx.Err(), context.Canceled) { return } if errors.Is(err, ErrSnapshotNotNewer) { diff --git a/internal/snapshotoffload/scheduler_test.go b/internal/snapshotoffload/scheduler_test.go index 3ea109b23..d30dd7b51 100644 --- a/internal/snapshotoffload/scheduler_test.go +++ b/internal/snapshotoffload/scheduler_test.go @@ -50,6 +50,63 @@ func (o *recordingObserver) snapshot() ([]uint64, []string, []error) { const schedulerTestIndex = 42 +type stalledHeadStore struct { + PublishStore + stall atomic.Bool +} + +func TestSchedulerRejectsNonpositivePublishTimeout(t *testing.T) { + t.Parallel() + store := newTestLocalStore(t, t.TempDir()) + for _, timeout := range []time.Duration{0, -time.Second} { + _, err := NewScheduler(store, nil, "", "cluster-a", "test", + WithSchedulerPublishTimeout(timeout)) + require.ErrorIs(t, err, ErrInvalidOptions) + } +} + +func (s *stalledHeadStore) HeadObject(ctx context.Context, key string) (ObjectInfo, bool, error) { + if s.stall.Load() { + <-ctx.Done() + return ObjectInfo{}, false, ctx.Err() + } + return s.PublishStore.HeadObject(ctx, key) +} + +func TestSchedulerPublicationTimeoutReleasesScanAndAllowsRetry(t *testing.T) { + t.Parallel() + root := t.TempDir() + dataDir := seedSchedulerGroup(t, root, "group") + store := &stalledHeadStore{PublishStore: newTestLocalStore(t, filepath.Join(root, "objects"))} + store.stall.Store(true) + observer := &recordingObserver{} + scheduler := newTestScheduler(t, store, []OffloadGroup{{ + GroupID: 7, DataDir: dataDir, + IsLeader: func() bool { return true }, + VerifyLeader: func(context.Context) error { return nil }, + }}, WithSchedulerPublishTimeout(500*time.Millisecond), WithSchedulerObserver(observer)) + + done := make(chan struct{}) + go func() { + scheduler.SyncOnce(context.Background()) + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("a stalled store must not block the scan indefinitely") + } + _, _, failures := observer.snapshot() + require.Len(t, failures, 1) + require.ErrorIs(t, failures[0], context.DeadlineExceeded) + require.Zero(t, scheduler.LastPublishedIndex(7)) + + store.stall.Store(false) + scheduler.publishTimeout = 5 * time.Second + scheduler.SyncOnce(context.Background()) + require.Equal(t, uint64(schedulerTestIndex), scheduler.LastPublishedIndex(7)) +} + func seedSchedulerGroup(t *testing.T, root, name string) string { t.Helper() dir := filepath.Join(root, name) diff --git a/main.go b/main.go index 0b144fbc7..44a101c26 100644 --- a/main.go +++ b/main.go @@ -874,6 +874,17 @@ func startDistributionStartup(in distributionStartupInput) (distributionStartup, } startMonitoringCollectors(in.ctx, in.metricsRegistry, in.runtimes, in.clock, in.encWiring.cache) startFSMCompactorIfEnabled(in.ctx, in.eg, in.runtimes, in.readTracker) + // §4 physical snapshot offload. Opt-in, and a hard error when + // configured-but-unbuildable: an operator who set a backup + // destination and silently got no backups is worse off than one + // whose node refused to start. + if err := startSnapshotOffload( + in.ctx, in.eg, in.runtimes, *raftDir, in.raftID, in.cfg.multi, + in.metricsRegistry.SnapshotOffloadObserver(), slog.Default(), + ); err != nil { + in.cancel() + return distributionStartup{}, err + } return distributionStartup{ defaultRuntime: defaultRuntime, distServer: distServer, diff --git a/main_snapshot_offload.go b/main_snapshot_offload.go new file mode 100644 index 000000000..f50cb7102 --- /dev/null +++ b/main_snapshot_offload.go @@ -0,0 +1,426 @@ +package main + +import ( + "context" + "flag" + "log/slog" + "net/http" + "net/url" + "os" + "path" + "strings" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http" + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/bootjp/elastickv/internal/snapshotoffload" + "github.com/cockroachdb/errors" + "golang.org/x/sync/errgroup" +) + +const ( + snapshotOffloadMaxRedirects = 10 + snapshotOffloadStaleSpoolAge = 24 * time.Hour + snapshotOffloadSpoolDirPerm = 0o755 +) + +// Physical snapshot object offload (design doc §4 / §7). Opt-in: the +// whole subsystem stays dormant unless --snapshotOffloadBucket (S3) or +// --snapshotOffloadLocalDir (filesystem) is set. +// +// The design requires that only a group's current leader publishes, so +// each group contributes both a cheap pre-check and a pre-commit +// leadership re-verification; the scheduler bounds the latter itself. +var ( + snapshotOffloadBucket = flag.String("snapshotOffloadBucket", "", + "S3 bucket for physical snapshot offload; empty to disable") + snapshotOffloadLocalDir = flag.String("snapshotOffloadLocalDir", "", + "filesystem root for physical snapshot offload; an alternative to --snapshotOffloadBucket, mainly for testing") + snapshotOffloadPrefix = flag.String("snapshotOffloadPrefix", "", + "key prefix below which snapshot artifacts are written") + snapshotOffloadRegion = flag.String("snapshotOffloadRegion", "", + "AWS region for the snapshot offload bucket") + snapshotOffloadEndpoint = flag.String("snapshotOffloadEndpoint", "", + "custom S3 endpoint for snapshot offload; empty uses the AWS default") + snapshotOffloadProfile = flag.String("snapshotOffloadProfile", "", + "shared-credentials profile for snapshot offload") + snapshotOffloadForcePathStyle = flag.Bool("snapshotOffloadForcePathStyle", false, + "use path-style addressing for the snapshot offload endpoint") + snapshotOffloadSSE = flag.String("snapshotOffloadServerSideEncryption", "", + "server-side encryption mode for snapshot objects (AES256 or aws:kms)") + snapshotOffloadSSEKMSKeyID = flag.String("snapshotOffloadSSEKMSKeyId", "", + "KMS key ARN when --snapshotOffloadServerSideEncryption is aws:kms") + snapshotOffloadInterval = flag.Duration("snapshotOffloadInterval", snapshotoffload.DefaultSchedulerInterval, + "how often to scan local groups for a publishable snapshot") + snapshotOffloadJitter = flag.Duration("snapshotOffloadJitter", 0, + "random spread applied to the offload schedule; zero uses a quarter of the interval") + snapshotOffloadConcurrency = flag.Int("snapshotOffloadConcurrency", snapshotoffload.DefaultSchedulerConcurrency, + "maximum concurrent snapshot uploads for this process") + snapshotOffloadPublishTimeout = flag.Duration("snapshotOffloadPublishTimeout", snapshotoffload.DefaultSchedulerPublishTimeout, + "deadline for each physical snapshot publication") + snapshotOffloadSpoolDir = flag.String("snapshotOffloadSpoolDir", "", + "directory for snapshot spool files; empty uses the data dir's filesystem") + snapshotOffloadSourceCluster = flag.String("snapshotOffloadSourceCluster", "", + "source cluster identity recorded in every manifest; required when offload is enabled") +) + +// snapshotOffloadAllowInsecureEndpoint is the narrow development opt-in for a +// plaintext offload endpoint. Off by default, and named so it cannot be +// mistaken for a tuning knob. +var snapshotOffloadAllowInsecureEndpoint = flag.Bool("snapshotOffloadAllowInsecureEndpoint", false, + "allow a plaintext http:// --snapshotOffloadEndpoint. DEVELOPMENT ONLY: snapshot "+ + "payloads and session credentials would cross the network unencrypted.") + +// rejectPlaintextOffloadEndpoint refuses an http:// endpoint unless the +// operator explicitly opted in. +// +// The design and the runbook both require TLS for the external bucket, and +// this path carries whole snapshot payloads plus the session credentials used +// to write them. An endpoint that silently downgraded to plaintext is the kind +// of misconfiguration that is invisible until someone captures the traffic, so +// it fails startup instead. +func rejectPlaintextOffloadEndpoint() error { + endpoint := strings.TrimSpace(*snapshotOffloadEndpoint) + if endpoint == "" { + return nil + } + parsed, err := url.Parse(endpoint) + if err != nil { + return errors.Wrapf(err, "parse --snapshotOffloadEndpoint %q", endpoint) + } + // A scheme-less host:port is ambiguous rather than known-plaintext, but + // the AWS SDK resolves it as http, so it is refused too. + allowedScheme := strings.EqualFold(parsed.Scheme, "https") || + (*snapshotOffloadAllowInsecureEndpoint && strings.EqualFold(parsed.Scheme, "http")) + if !allowedScheme || parsed.Hostname() == "" || parsed.Opaque != "" { + return errors.Wrapf(snapshotoffload.ErrInvalidOptions, + "--snapshotOffloadEndpoint %q is not a valid https URL; snapshot payloads and credentials "+ + "would cross the network unencrypted. Pass an https:// endpoint, or "+ + "--snapshotOffloadAllowInsecureEndpoint for local development only", + endpoint) + } + return nil +} + +func snapshotOffloadHTTPClient(base aws.HTTPClient) (aws.HTTPClient, error) { + var client http.Client + switch resolved := base.(type) { + case nil: + client.Transport = awshttp.NewBuildableClient().GetTransport() + case *awshttp.BuildableClient: + client.Transport = resolved.GetTransport() + client.Timeout = resolved.GetTimeout() + case *http.Client: + client = *resolved + default: + return nil, errors.Wrapf(snapshotoffload.ErrInvalidOptions, + "unsupported snapshot offload SDK http client %T", base) + } + transport := client.Transport + if transport == nil { + transport = http.DefaultTransport + } + allowInsecure := *snapshotOffloadAllowInsecureEndpoint + client.Transport = snapshotOffloadHTTPSTransport{next: transport, allowInsecure: allowInsecure} + client.CheckRedirect = func(req *http.Request, via []*http.Request) error { + return checkSnapshotOffloadRedirect(req, via, allowInsecure) + } + return &client, nil +} + +type snapshotOffloadHTTPSTransport struct { + next http.RoundTripper + allowInsecure bool +} + +func (t snapshotOffloadHTTPSTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if !snapshotOffloadRequestAllowed(req, t.allowInsecure) { + return nil, errors.Wrap(snapshotoffload.ErrInvalidOptions, + "snapshot offload request must use https") + } + if t.next == nil { + return nil, errors.Wrap(snapshotoffload.ErrInvalidOptions, + "snapshot offload transport is required") + } + response, err := t.next.RoundTrip(req) + return response, errors.Wrap(err, "snapshot offload request") +} + +func snapshotOffloadRequestAllowed(req *http.Request, allowInsecure bool) bool { + if req == nil || req.URL == nil { + return false + } + return strings.EqualFold(req.URL.Scheme, "https") || + (allowInsecure && strings.EqualFold(req.URL.Scheme, "http")) +} + +func checkSnapshotOffloadRedirect(req *http.Request, via []*http.Request, allowInsecure bool) error { + if !snapshotOffloadRequestAllowed(req, allowInsecure) { + return errors.Wrap(snapshotoffload.ErrInvalidOptions, + "snapshot offload redirect must preserve https") + } + if len(via) >= snapshotOffloadMaxRedirects { + return errors.New("snapshot offload stopped after 10 redirects") + } + if req.Response == nil { + return errors.New("snapshot offload redirect response is required") + } + if req.Response.StatusCode != http.StatusTemporaryRedirect && + req.Response.StatusCode != http.StatusPermanentRedirect { + return http.ErrUseLastResponse + } + if len(via) > 0 && !strings.EqualFold(via[len(via)-1].URL.Host, req.URL.Host) { + return errors.Wrap(snapshotoffload.ErrInvalidOptions, + "snapshot offload redirect must preserve the endpoint host") + } + return nil +} + +func validateSnapshotOffloadPrefix(prefix string) error { + prefix = strings.Trim(strings.TrimSpace(prefix), "/") + if prefix == "" { + return nil + } + cleaned := path.Clean(prefix) + if cleaned == ".." || strings.HasPrefix(cleaned, "../") { + return errors.Wrapf(snapshotoffload.ErrInvalidOptions, + "invalid --snapshotOffloadPrefix %q", prefix) + } + return nil +} + +func validateSnapshotOffloadSpoolDir() error { + dir := strings.TrimSpace(*snapshotOffloadSpoolDir) + if dir == "" { + return nil + } + if err := os.MkdirAll(dir, snapshotOffloadSpoolDirPerm); err != nil { + return errors.Wrap(err, "snapshot offload: create spool dir") + } + info, err := os.Stat(dir) + if err != nil { + return errors.Wrap(err, "snapshot offload: stat spool dir") + } + if !info.IsDir() { + return errors.Wrapf(snapshotoffload.ErrInvalidOptions, + "--snapshotOffloadSpoolDir %q is not a directory", dir) + } + return nil +} + +// snapshotOffloadEnabled reports whether the operator configured a +// destination. Checked before any other offload flag is validated so a +// node that never opts in cannot fail startup on offload config. +func snapshotOffloadEnabled() bool { + return strings.TrimSpace(*snapshotOffloadBucket) != "" || + strings.TrimSpace(*snapshotOffloadLocalDir) != "" +} + +// buildSnapshotOffloadStore constructs the configured object store. +// +// Bucket and local dir are mutually exclusive: accepting both would +// leave which destination actually receives the artifacts ambiguous, +// and a backup written to the wrong place is discovered only when a +// restore is attempted. +func buildSnapshotOffloadStore(ctx context.Context) (snapshotOffloadStore, error) { + bucket := strings.TrimSpace(*snapshotOffloadBucket) + localDir := strings.TrimSpace(*snapshotOffloadLocalDir) + if bucket != "" && localDir != "" { + return nil, errors.Wrap(snapshotoffload.ErrInvalidOptions, + "--snapshotOffloadBucket and --snapshotOffloadLocalDir are mutually exclusive") + } + if localDir != "" { + store, err := snapshotoffload.NewLocalStore(localDir) + if err != nil { + return nil, errors.Wrap(err, "snapshot offload: local store") + } + return store, nil + } + if err := rejectPlaintextOffloadEndpoint(); err != nil { + return nil, err + } + store, err := snapshotoffload.NewS3Store(ctx, snapshotoffload.S3StoreConfig{ + Bucket: bucket, + Region: strings.TrimSpace(*snapshotOffloadRegion), + Endpoint: strings.TrimSpace(*snapshotOffloadEndpoint), + Profile: strings.TrimSpace(*snapshotOffloadProfile), + ForcePathStyle: *snapshotOffloadForcePathStyle, + ServerSideEncryption: strings.TrimSpace(*snapshotOffloadSSE), + SSEKMSKeyID: strings.TrimSpace(*snapshotOffloadSSEKMSKeyID), + HTTPClientDecorator: snapshotOffloadHTTPClient, + AllowVersionedBucketWithLifecycle: *snapshotOffloadAllowVersionedLifecycle, + }) + if err != nil { + return nil, errors.Wrap(err, "snapshot offload: s3 store") + } + return store, nil +} + +// cleanStaleOffloadSpool removes leftover spool files from every directory +// this process would publish through. +// +// Files younger than snapshotOffloadStaleSpoolAge may belong to another +// process sharing the configured spool directory and are left untouched. +func cleanStaleOffloadSpool( + runtimes []*raftGroupRuntime, + raftDir, raftID string, + multi bool, + logger *slog.Logger, +) { + configured := strings.TrimSpace(*snapshotOffloadSpoolDir) + seen := make(map[string]struct{}) + cutoff := time.Now().Add(-snapshotOffloadStaleSpoolAge) + for _, group := range snapshotOffloadGroups(runtimes, raftDir, raftID, multi) { + dir := snapshotoffload.SpoolDirFor(configured, group.DataDir) + if _, done := seen[dir]; done { + continue + } + seen[dir] = struct{}{} + removed, err := snapshotoffload.CleanStaleSpoolFiles(dir, cutoff) + if err != nil { + logger.Warn("snapshot offload: could not clean stale spool files", + slog.String("spool_dir", dir), slog.Any("err", err)) + } + if len(removed) > 0 { + logger.Info("snapshot offload: removed stale spool files", + slog.String("spool_dir", dir), slog.Int("files", len(removed))) + } + } +} + +// snapshotOffloadGroups uses snapshotEngine() for leadership callbacks so +// scheduler reads are synchronized with runtime shutdown. +func snapshotOffloadGroups( + runtimes []*raftGroupRuntime, raftDir, raftID string, multi bool, +) []snapshotoffload.OffloadGroup { + groups := make([]snapshotoffload.OffloadGroup, 0, len(runtimes)) + for _, rt := range runtimes { + if rt == nil { + continue + } + groups = append(groups, snapshotoffload.OffloadGroup{ + GroupID: rt.spec.id, + DataDir: groupDataDir(raftDir, raftID, rt.spec.id, multi), + IsLeader: snapshotOffloadIsLeader(rt), + VerifyLeader: snapshotOffloadVerifyLeader(rt), + }) + } + return groups +} + +func snapshotOffloadIsLeader(rt *raftGroupRuntime) func() bool { + return func() bool { + engine := rt.snapshotEngine() + return engine != nil && engine.State() == raftengine.StateLeader + } +} + +// snapshotOffloadVerifyLeader is the §4 pre-commit re-verification: a +// multi-gigabyte spool takes long enough to lose an election, so +// leadership must hold at the instant the manifest commits, not merely +// when the snapshot was opened. +func snapshotOffloadVerifyLeader(rt *raftGroupRuntime) func(context.Context) error { + return func(ctx context.Context) error { + engine := rt.snapshotEngine() + if engine == nil { + return errors.Wrap(snapshotoffload.ErrInvalidOptions, + "snapshot offload: raft engine closed") + } + verifier, ok := engine.(interface { + VerifyLeader(context.Context) error + }) + if !ok { + return errors.Wrap(snapshotoffload.ErrInvalidOptions, + "snapshot offload: raft engine cannot verify leadership") + } + return errors.Wrap(verifier.VerifyLeader(ctx), "snapshot offload: verify leadership") + } +} + +// startSnapshotOffload wires and starts the scheduler when offload is +// configured. It returns an error rather than logging and continuing: +// an operator who configured a backup destination and got no backups +// is worse off than one whose node refused to start. +func startSnapshotOffload( + ctx context.Context, + eg *errgroup.Group, + runtimes []*raftGroupRuntime, + raftDir, raftID string, + multi bool, + observer snapshotoffload.SchedulerObserver, + logger *slog.Logger, +) (retErr error) { + if *snapshotOffloadGCEnabled && !snapshotOffloadEnabled() { + return errors.Wrap(snapshotoffload.ErrInvalidOptions, "snapshot retention requires an offload destination") + } + if !snapshotOffloadEnabled() { + return nil + } + if err := validateSnapshotOffloadPrefix(*snapshotOffloadPrefix); err != nil { + return err + } + if err := validateSnapshotOffloadSpoolDir(); err != nil { + return err + } + store, err := buildSnapshotOffloadStore(ctx) + if err != nil { + return err + } + handedOff := false + defer func() { + if !handedOff { + retErr = errors.CombineErrors(retErr, closeSnapshotOffloadStore(store)) + } + }() + + scheduler, err := snapshotoffload.NewScheduler( + store, + snapshotOffloadGroups(runtimes, raftDir, raftID, multi), + strings.TrimSpace(*snapshotOffloadPrefix), + strings.TrimSpace(*snapshotOffloadSourceCluster), + buildVersion(), + snapshotOffloadSchedulerOptions(observer, logger)..., + ) + if err != nil { + return errors.Wrap(err, "snapshot offload: scheduler") + } + retention, err := buildSnapshotOffloadRetention(ctx, store, logger) + if err != nil { + return err + } + + // Clean only aged leftovers before starting this process's publishers. + cleanStaleOffloadSpool(runtimes, raftDir, raftID, multi, logger) + + logger.Info("snapshot offload enabled", + slog.Int("groups", len(runtimes)), + slog.Duration("interval", *snapshotOffloadInterval), + slog.Int("concurrency", *snapshotOffloadConcurrency)) + + eg.Go(func() error { + // Run returns only on context cancellation; a failing group is + // retried on the next tick rather than tearing the process + // down, because an object-store outage must not stop serving. + return runSnapshotOffloadWorkers(ctx, store, scheduler, retention) + }) + handedOff = true + return nil +} + +func snapshotOffloadSchedulerOptions(observer snapshotoffload.SchedulerObserver, logger *slog.Logger) []snapshotoffload.SchedulerOption { + opts := []snapshotoffload.SchedulerOption{ + snapshotoffload.WithSchedulerInterval(*snapshotOffloadInterval), + snapshotoffload.WithSchedulerConcurrency(*snapshotOffloadConcurrency), + snapshotoffload.WithSchedulerPublishTimeout(*snapshotOffloadPublishTimeout), + snapshotoffload.WithSchedulerObserver(observer), + snapshotoffload.WithSchedulerLogger(logger), + } + if *snapshotOffloadJitter > 0 { + opts = append(opts, snapshotoffload.WithSchedulerJitter(*snapshotOffloadJitter)) + } + if dir := strings.TrimSpace(*snapshotOffloadSpoolDir); dir != "" { + opts = append(opts, snapshotoffload.WithSchedulerSpoolDir(dir)) + } + return opts +} diff --git a/main_snapshot_offload_acceptance_test.go b/main_snapshot_offload_acceptance_test.go new file mode 100644 index 000000000..23ecd227b --- /dev/null +++ b/main_snapshot_offload_acceptance_test.go @@ -0,0 +1,262 @@ +package main + +import ( + "context" + "fmt" + "math" + "os" + "slices" + "strings" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/raftengine" + etcdraftengine "github.com/bootjp/elastickv/internal/raftengine/etcd" + "github.com/bootjp/elastickv/internal/snapshotoffload" + pb "github.com/bootjp/elastickv/proto" + "github.com/stretchr/testify/require" + "golang.org/x/sync/errgroup" +) + +const ( + offloadAcceptanceNodeCount = 3 + offloadAcceptanceBatchSize = 12 + offloadAcceptanceTimeout = 20 * time.Second + offloadAcceptancePoll = 25 * time.Millisecond +) + +// Reuses the real Raft/Pebble/gRPC bootstrap fixture: no prebuilt snapshot, +// mocked leadership, in-memory object backend or manual publication call. +func TestSnapshotOffloadThreeNodeRecoveryAcceptance(t *testing.T) { + t.Setenv("ELASTICKV_RAFT_SNAPSHOT_COUNT", "4") + configureOffloadAcceptance(t) + baseDir, endpoints, nodes := startBootstrapE2ECluster(t, t.TempDir(), offloadAcceptanceNodeCount, 1, raftEngineEtcd) + t.Cleanup(func() { closeBootstrapE2ENodes(t, nodes) }) + waitForBootstrapClusterConfig(t, nodes, bootstrapExpectedConfiguration(endpoints)) + leader := waitForSingleLeader(t, nodes) + groupID := nodes[leader].runtimes[0].spec.id + objects, err := snapshotoffload.NewLocalStore(*snapshotOffloadLocalDir) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, objects.Close()) }) + clients, conns := rawKVClients(t, endpoints) + t.Cleanup(func() { closeGRPCConns(conns) }) + + workers := make([]func(), len(nodes)) + for i, node := range nodes { + workers[i] = startOffloadAcceptanceWorker(t, baseDir, node) + } + expected := make(map[string]string) + writeOffloadAcceptanceBatch(t, clients[leader], expected, 0) + first := checkpointOffloadAcceptance(t, objects, nodes[leader], clients[leader], groupID) + require.Len(t, first.ConfState.Voters, offloadAcceptanceNodeCount) + + // Lose one replica without changing membership. The surviving quorum must + // still serve writes while publication and reclamation are running. + follower := (leader + 1) % len(nodes) + workers[follower]() + require.NoError(t, nodes[follower].close()) + nodes[follower] = nil + for i, node := range nodes { + if node == nil { + continue + } + workers[i]() + *snapshotOffloadGCEnabled = i == leader + workers[i] = startOffloadAcceptanceWorker(t, baseDir, node) + } + writeOffloadAcceptanceBatch(t, clients[leader], expected, offloadAcceptanceBatchSize) + second := checkpointOffloadAcceptance(t, objects, nodes[leader], clients[leader], groupID) + require.Greater(t, second.SnapshotIndex, first.SnapshotIndex) + require.NotEqual(t, first.Payload.Key, second.Payload.Key) + require.Eventually(t, func() bool { + _, manifestExists, manifestErr := objects.HeadObject(context.Background(), first.ManifestKey) + _, payloadExists, payloadErr := objects.HeadObject(context.Background(), first.Payload.Key) + return manifestErr == nil && payloadErr == nil && !manifestExists && !payloadExists + }, offloadAcceptanceTimeout, offloadAcceptancePoll, "runtime GC must actually reclaim the old manifest and payload") + writeOffloadAcceptanceBatch(t, clients[leader], expected, 2*offloadAcceptanceBatchSize) + latest := checkpointOffloadAcceptance(t, objects, nodes[leader], clients[leader], groupID) + require.Greater(t, latest.SnapshotIndex, second.SnapshotIndex) + for i, node := range nodes { + if node != nil { + requireOffloadAcceptanceLocalValues(t, node, expected) + requireOffloadAcceptanceValues(t, clients[i], expected) + } + } + for _, stop := range workers { + stop() + } + closeBootstrapE2ENodes(t, nodes) + clear(nodes) + // Re-open the object root after all publishers/GC stop. Recovery must use + // durable bytes, not a live publisher's memory or the source's data dir. + require.NoError(t, objects.Close()) + objects, err = snapshotoffload.NewLocalStore(*snapshotOffloadLocalDir) + require.NoError(t, err) + latest = waitForOffloadAcceptanceManifest(t, objects, groupID, latest.SnapshotIndex) + t.Logf("source: three voters, one stopped; %d verified user keys plus checkpoint writes; periodic indexes %d -> %d -> %d; old manifest/payload reclaimed", + len(expected), first.SnapshotIndex, second.SnapshotIndex, latest.SnapshotIndex) + restoreOffloadAcceptanceCluster(t, objects, latest, expected) +} + +func configureOffloadAcceptance(t *testing.T) { + t.Helper() + configureSnapshotGC(t) + withOffloadFlags(t, "", t.TempDir()) + interval, jitter, timeout := *snapshotOffloadInterval, *snapshotOffloadJitter, *snapshotOffloadPublishTimeout + t.Cleanup(func() { + *snapshotOffloadInterval, *snapshotOffloadJitter, *snapshotOffloadPublishTimeout = interval, jitter, timeout + }) + *snapshotOffloadInterval, *snapshotOffloadJitter, *snapshotOffloadPublishTimeout = offloadAcceptancePoll, time.Nanosecond, 5*time.Second + *snapshotOffloadGCEnabled = false + *snapshotOffloadGCInterval = offloadAcceptancePoll + *snapshotOffloadGCPayloadGrace, *snapshotOffloadGCMinMarkAge = 100*time.Millisecond, 100*time.Millisecond +} + +func startOffloadAcceptanceWorker(t *testing.T, baseDir string, node *bootstrapE2ENode) func() { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + workers, workerCtx := errgroup.WithContext(ctx) + require.NoError(t, startSnapshotOffload(workerCtx, workers, node.runtimes, baseDir, node.id, false, nil, testLogger(t))) + stop := func() { + cancel() + require.NoError(t, workers.Wait()) + } + t.Cleanup(stop) + return stop +} + +func writeOffloadAcceptanceBatch(t *testing.T, client pb.RawKVClient, expected map[string]string, first int) { + t.Helper() + for i := first; i < first+offloadAcceptanceBatchSize; i++ { + key, value := fmt.Sprintf("offload-acceptance-%03d", i), fmt.Sprintf("committed-value-%03d", i) + require.NoError(t, rawPutWithTimeout(client, []byte(key), []byte(value))) + expected[key] = value + } +} + +func requireOffloadAcceptanceValues(t *testing.T, client pb.RawKVClient, expected map[string]string) { + t.Helper() + // A follower can learn the new leader after the election is visible on + // the winner. Bound startup convergence rather than assuming it is atomic. + require.Eventually(t, func() bool { + for key, value := range expected { + response, err := rawGetWithTimeout(client, []byte(key), bootstrapE2ERPCRequestTimeout) + if err != nil || !response.Exists || string(response.Value) != value { + return false + } + } + return true + }, offloadAcceptanceTimeout, offloadAcceptancePoll, "adapter did not return every expected value") +} + +func requireOffloadAcceptanceLocalValues(t *testing.T, node *bootstrapE2ENode, expected map[string]string) { + t.Helper() + // Adapter reads can forward to the leader. Check each replica's local FSM + // too, so forwarding or a healthy peer cannot hide missing restored data. + require.Eventually(t, func() bool { + for key, value := range expected { + got, err := node.runtimes[0].store.GetAt(context.Background(), []byte(key), math.MaxUint64) + if err != nil || string(got) != value { + return false + } + } + return true + }, offloadAcceptanceTimeout, offloadAcceptancePoll, "%s local FSM did not contain every expected value", node.id) +} + +func checkpointOffloadAcceptance(t *testing.T, objects *snapshotoffload.LocalStore, node *bootstrapE2ENode, client pb.RawKVClient, groupID uint64) snapshotoffload.Manifest { + t.Helper() + minimum := node.engine().Status().AppliedIndex + // The fixture has no idle HLC-renewal loop. Advance the real log through + // the next snapshot threshold so the last asserted user write is covered. + for range node.engine().SnapshotEvery() { + require.NoError(t, rawPutWithTimeout(client, []byte("offload-checkpoint"), []byte("barrier"))) + } + return waitForOffloadAcceptanceManifest(t, objects, groupID, minimum) +} + +func waitForOffloadAcceptanceManifest(t *testing.T, objects *snapshotoffload.LocalStore, groupID, minimum uint64) snapshotoffload.Manifest { + t.Helper() + var latest snapshotoffload.Manifest + require.Eventually(t, func() bool { + refs, err := objects.ListObjects(context.Background(), fmt.Sprintf("cluster-a/v1/groups/%d/snapshots/", groupID)) + if err != nil || len(refs) == 0 { + return false + } + slices.SortFunc(refs, func(a, b snapshotoffload.ObjectRef) int { + return strings.Compare(b.Key, a.Key) + }) + latest, err = snapshotoffload.LoadManifest(context.Background(), objects, refs[0].Key) + return err == nil && latest.SnapshotIndex >= minimum + }, offloadAcceptanceTimeout, offloadAcceptancePoll, "periodic publication did not reach applied index %d", minimum) + require.Equal(t, groupID, latest.GroupID) + require.Equal(t, "cluster-a", latest.SourceCluster) + return latest +} + +func restoreOffloadAcceptanceCluster(t *testing.T, objects *snapshotoffload.LocalStore, manifest snapshotoffload.Manifest, expected map[string]string) { + t.Helper() + root := t.TempDir() + endpoints, listeners := allocateBootstrapE2EEndpoints(t, offloadAcceptanceNodeCount) + t.Cleanup(func() { closeBootstrapE2EListeners(listeners) }) + peers := make([]etcdraftengine.Peer, len(endpoints)) + for i := range endpoints { + endpoints[i].id = fmt.Sprintf("replacement-%d", i+1) + peers[i] = etcdraftengine.Peer{ + NodeID: etcdraftengine.DeriveNodeID(endpoints[i].id), ID: endpoints[i].id, Address: endpoints[i].raftAddr, + Suffrage: etcdraftengine.SuffrageVoter, + } + } + for _, endpoint := range endpoints { + dest := groupDataDir(root, endpoint.id, manifest.GroupID, false) + _, err := os.Stat(dest) + require.True(t, os.IsNotExist(err)) + result, err := snapshotoffload.RestorePhysicalSnapshot(context.Background(), snapshotoffload.RestoreOptions{ + Store: objects, ManifestKey: manifest.ManifestKey, DataDir: dest, Peers: peers, + ExpectGroupID: &manifest.GroupID, ExpectSourceCluster: "cluster-a", + }) + require.NoError(t, err) + require.Equal(t, manifest.Payload.SHA256, result.PayloadSHA256) + storedPeers, exists, err := etcdraftengine.LoadPersistedPeers(dest) + require.NoError(t, err) + require.True(t, exists) + require.Len(t, storedPeers, offloadAcceptanceNodeCount) + for _, peer := range storedPeers { + require.Contains(t, peers, peer) + } + } + nodes := make([]*bootstrapE2ENode, len(endpoints)) + t.Cleanup(func() { closeBootstrapE2ENodes(t, nodes) }) + for i, endpoint := range endpoints { + node, err := startBootstrapE2ENode(root, endpoint, listeners[i], false, "", raftEngineEtcd) + require.NoError(t, err) + nodes[i] = node + } + waitForBootstrapClusterConfig(t, nodes, bootstrapExpectedConfiguration(endpoints)) + leader := waitForSingleLeader(t, nodes) + clients, conns := rawKVClients(t, endpoints) + t.Cleanup(func() { closeGRPCConns(conns) }) + for i, node := range nodes { + require.GreaterOrEqual(t, node.engine().Status().AppliedIndex, manifest.SnapshotIndex) + reader, ok := node.runtimes[0].stateMachine.(raftengine.AppliedIndexReader) + require.True(t, ok) + index, present, err := reader.LastAppliedIndex() + require.NoError(t, err) + require.True(t, present) + require.GreaterOrEqual(t, index, manifest.SnapshotIndex) + requireOffloadAcceptanceLocalValues(t, node, expected) + requireOffloadAcceptanceValues(t, clients[i], expected) + t.Logf("replacement %s: restored index=%d applied=%d durable_applied=%d keys=%d", node.id, manifest.SnapshotIndex, node.engine().Status().AppliedIndex, index, len(expected)) + } + require.NoError(t, rawPutWithTimeout(clients[leader], []byte("after-offload-restore"), []byte("replicated"))) + committed := nodes[leader].engine().Status().AppliedIndex + expected["after-offload-restore"] = "replicated" + for i, node := range nodes { + requireOffloadAcceptanceLocalValues(t, node, expected) + requireOffloadAcceptanceValues(t, clients[i], expected) + require.Eventually(t, func() bool { + return node.engine().Status().AppliedIndex >= committed + }, offloadAcceptanceTimeout, offloadAcceptancePoll) + } + t.Logf("recovery: %d nodes read all %d keys and accepted a new replicated write", len(nodes), len(expected)) +} diff --git a/main_snapshot_offload_gc.go b/main_snapshot_offload_gc.go new file mode 100644 index 000000000..939395a65 --- /dev/null +++ b/main_snapshot_offload_gc.go @@ -0,0 +1,137 @@ +package main + +import ( + "context" + "flag" + "io" + "log/slog" + "strings" + "time" + + "github.com/bootjp/elastickv/internal/snapshotoffload" + "github.com/cockroachdb/errors" + "golang.org/x/sync/errgroup" +) + +const defaultSnapshotRetentionTimeout = 10 * time.Minute + +var ( + snapshotOffloadGCEnabled = flag.Bool("snapshotOffloadGCEnabled", false, "enable snapshot retention on this worker after completing the publisher claims rollout") + snapshotOffloadGCClaimsComplete = flag.Bool("snapshotOffloadGCClaimsRolloutComplete", false, "operator assertion: every publisher sharing the destination/prefix uses claims and all legacy publications have drained") + snapshotOffloadGCInterval = flag.Duration("snapshotOffloadGCInterval", time.Hour, "interval between serial snapshot retention passes") + snapshotOffloadGCTimeout = flag.Duration("snapshotOffloadGCTimeout", defaultSnapshotRetentionTimeout, "deadline for one snapshot retention pass") + snapshotOffloadGCMinGenerations = flag.Int("snapshotOffloadGCMinGenerations", snapshotoffload.DefaultMinGenerations, "minimum snapshot generations retained per Raft group") + snapshotOffloadGCMaxAge = flag.Duration("snapshotOffloadGCMaxAge", snapshotoffload.DefaultMaxAge, "age window retained in addition to minimum generations") + snapshotOffloadGCPayloadGrace = flag.Duration("snapshotOffloadGCPayloadGrace", snapshotoffload.DefaultPayloadGrace, "minimum age of an unreferenced snapshot payload before marking") + snapshotOffloadGCMinMarkAge = flag.Duration("snapshotOffloadGCMinMarkAge", snapshotoffload.DefaultMinMarkAge, "minimum delay between marking and reclaiming a snapshot payload") + snapshotOffloadAllowVersionedLifecycle = flag.Bool("snapshotOffloadAllowVersionedWithLifecycle", false, "allow retention in a versioned S3 bucket after confirming noncurrent-version expiration") +) + +type snapshotOffloadStore interface { + snapshotoffload.PublishStore + snapshotoffload.RetentionStore +} + +type snapshotOffloadRetention struct { + gc *snapshotoffload.GC + interval time.Duration + timeout time.Duration + logger *slog.Logger +} + +func buildSnapshotOffloadRetention(ctx context.Context, store snapshotOffloadStore, logger *slog.Logger) (*snapshotOffloadRetention, error) { + if !*snapshotOffloadGCEnabled { + return nil, nil + } + if err := validateSnapshotRetentionConfig(); err != nil { + return nil, err + } + confirmed := *snapshotOffloadGCClaimsComplete + verify := func(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return errors.WithStack(err) + } + if !confirmed { + return errors.Wrap(snapshotoffload.ErrInvalidOptions, "snapshot retention requires --snapshotOffloadGCClaimsRolloutComplete after all legacy publishers are stopped and drained") + } + return nil + } + if err := verify(ctx); err != nil { + return nil, err + } + gc, err := snapshotoffload.NewGC(snapshotoffload.GCOptions{ + Store: store, Prefix: strings.TrimSpace(*snapshotOffloadPrefix), + Policy: snapshotoffload.RetentionPolicy{ + MinGenerations: *snapshotOffloadGCMinGenerations, MaxAge: *snapshotOffloadGCMaxAge, + PayloadGrace: *snapshotOffloadGCPayloadGrace, MinMarkAge: *snapshotOffloadGCMinMarkAge, + }, + VerifyPublishersUseClaims: verify, Logger: logger, + }) + if err != nil { + return nil, errors.Wrap(err, "snapshot retention: create GC") + } + bounded, cancel := context.WithTimeout(ctx, *snapshotOffloadGCTimeout) + defer cancel() + if err := store.ValidateRetention(bounded); err != nil { + return nil, errors.Wrap(err, "snapshot retention: validate bucket policy") + } + return &snapshotOffloadRetention{gc: gc, interval: *snapshotOffloadGCInterval, timeout: *snapshotOffloadGCTimeout, logger: logger}, nil +} + +func validateSnapshotRetentionConfig() error { + if *snapshotOffloadGCInterval <= 0 || *snapshotOffloadGCTimeout <= 0 || + *snapshotOffloadGCMinGenerations < 1 || *snapshotOffloadGCMaxAge <= 0 || + *snapshotOffloadGCPayloadGrace <= 0 || *snapshotOffloadGCMinMarkAge <= 0 { + return errors.Wrap(snapshotoffload.ErrInvalidOptions, "snapshot retention intervals, deadlines and policy must be positive") + } + return nil +} + +func (r *snapshotOffloadRetention) run(ctx context.Context) error { + for { + if ctx.Err() != nil { + return errors.WithStack(ctx.Err()) + } + bounded, cancel := context.WithTimeout(ctx, r.timeout) + result, err := r.gc.RunOnce(bounded) + cancel() + if ctx.Err() != nil { + return errors.WithStack(ctx.Err()) + } + if err != nil { + r.logger.WarnContext(ctx, "snapshot retention pass failed", "err", err) + } else { + r.logger.InfoContext(ctx, "snapshot retention pass completed", + "manifests_deleted", len(result.ManifestsDeleted), "payloads_deleted", len(result.PayloadsDeleted), + "payloads_awaiting_sweep", result.PayloadsAwaitingSweep, "skip_reason", result.SkipReason) + } + timer := time.NewTimer(r.interval) + select { + case <-ctx.Done(): + timer.Stop() + return nil + case <-timer.C: + } + } +} + +func closeSnapshotOffloadStore(store snapshotOffloadStore) error { + if closer, ok := store.(io.Closer); ok { + return errors.Wrap(closer.Close(), "close snapshot offload store") + } + return nil +} + +func runSnapshotOffloadWorkers(ctx context.Context, store snapshotOffloadStore, scheduler *snapshotoffload.Scheduler, retention *snapshotOffloadRetention) (retErr error) { + defer func() { retErr = errors.CombineErrors(retErr, closeSnapshotOffloadStore(store)) }() + workers, childCtx := errgroup.WithContext(ctx) + workers.Go(func() error { return scheduler.Run(childCtx) }) + if retention != nil { + workers.Go(func() error { return retention.run(childCtx) }) + } + err := workers.Wait() + if ctx.Err() != nil { + return nil + } + return errors.Wrap(err, "snapshot offload workers") +} diff --git a/main_snapshot_offload_gc_test.go b/main_snapshot_offload_gc_test.go new file mode 100644 index 000000000..a7e4c4f1f --- /dev/null +++ b/main_snapshot_offload_gc_test.go @@ -0,0 +1,150 @@ +package main + +import ( + "context" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + etcdraftengine "github.com/bootjp/elastickv/internal/raftengine/etcd" + "github.com/bootjp/elastickv/internal/snapshotoffload" + "github.com/stretchr/testify/require" + "golang.org/x/sync/errgroup" +) + +func configureSnapshotGC(t *testing.T) { + t.Helper() + enabled, confirmed := *snapshotOffloadGCEnabled, *snapshotOffloadGCClaimsComplete + interval, timeout := *snapshotOffloadGCInterval, *snapshotOffloadGCTimeout + minimum, age := *snapshotOffloadGCMinGenerations, *snapshotOffloadGCMaxAge + grace, mark := *snapshotOffloadGCPayloadGrace, *snapshotOffloadGCMinMarkAge + prefix, cluster := *snapshotOffloadPrefix, *snapshotOffloadSourceCluster + t.Cleanup(func() { + *snapshotOffloadGCEnabled, *snapshotOffloadGCClaimsComplete = enabled, confirmed + *snapshotOffloadGCInterval, *snapshotOffloadGCTimeout = interval, timeout + *snapshotOffloadGCMinGenerations, *snapshotOffloadGCMaxAge = minimum, age + *snapshotOffloadGCPayloadGrace, *snapshotOffloadGCMinMarkAge = grace, mark + *snapshotOffloadPrefix, *snapshotOffloadSourceCluster = prefix, cluster + }) + *snapshotOffloadGCEnabled, *snapshotOffloadGCClaimsComplete = true, true + *snapshotOffloadGCInterval, *snapshotOffloadGCTimeout = 10*time.Millisecond, time.Second + *snapshotOffloadGCMinGenerations, *snapshotOffloadGCMaxAge = 1, time.Nanosecond + *snapshotOffloadGCPayloadGrace, *snapshotOffloadGCMinMarkAge = time.Nanosecond, time.Nanosecond + *snapshotOffloadPrefix, *snapshotOffloadSourceCluster = "cluster-a", "cluster-a" +} + +func TestSnapshotOffloadGCRuntimeRequiresExplicitRolloutAndDestination(t *testing.T) { + configureSnapshotGC(t) + withOffloadFlags(t, "", "") + err := startSnapshotOffload(context.Background(), nil, nil, t.TempDir(), "n1", false, nil, testLogger(t)) + require.ErrorContains(t, err, "requires an offload destination") + + *snapshotOffloadLocalDir = t.TempDir() + *snapshotOffloadGCClaimsComplete = false + err = startSnapshotOffload(context.Background(), nil, nil, t.TempDir(), "n1", false, nil, testLogger(t)) + require.ErrorContains(t, err, "snapshotOffloadGCClaimsRolloutComplete") + *snapshotOffloadGCEnabled = false + store, err := buildSnapshotOffloadStore(context.Background()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, closeSnapshotOffloadStore(store)) }) + retention, err := buildSnapshotOffloadRetention(context.Background(), store, testLogger(t)) + require.NoError(t, err) + require.Nil(t, retention, "publication-only configuration must never enable reclamation") + *snapshotOffloadGCEnabled, *snapshotOffloadGCClaimsComplete = true, true + *snapshotOffloadGCInterval = 0 + _, err = buildSnapshotOffloadRetention(context.Background(), store, testLogger(t)) + require.ErrorIs(t, err, snapshotoffload.ErrInvalidOptions) +} + +func seedRuntimeOffloadSnapshot(t *testing.T, store snapshotoffload.PublishStore, index uint64, payload string) *snapshotoffload.Manifest { + t.Helper() + root := t.TempDir() + input := filepath.Join(root, "input.fsm") + require.NoError(t, os.WriteFile(input, []byte(payload), 0o600)) + dataDir := filepath.Join(root, "raft") + _, err := etcdraftengine.PreparePhysicalSnapshotRestore(etcdraftengine.PhysicalSnapshotRestoreOptions{ + InputFSMPath: input, DataDir: dataDir, Index: index, Term: 2, + Peers: []etcdraftengine.Peer{{NodeID: 1, ID: "n1", Address: "127.0.0.1:12001"}}, + }) + require.NoError(t, err) + manifest, err := snapshotoffload.PublishPersistedSnapshot(context.Background(), snapshotoffload.PublishOptions{ + Store: store, DataDir: dataDir, Prefix: "cluster-a", GroupID: 1, + SourceCluster: "cluster-a", CreatedAt: time.Now().Add(-time.Hour), + }) + require.NoError(t, err) + return manifest +} + +func TestSnapshotOffloadRuntimeReclaimsAndRestoresRetainedSnapshot(t *testing.T) { + configureSnapshotGC(t) + root := t.TempDir() + withOffloadFlags(t, "", root) + objects, err := snapshotoffload.NewLocalStore(root) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, objects.Close()) }) + stale := seedRuntimeOffloadSnapshot(t, objects, 10, "EKVTHLC1old-runtime-payload") + latest := seedRuntimeOffloadSnapshot(t, objects, 20, "EKVTHLC1latest-runtime-payload") + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + workers, workerCtx := errgroup.WithContext(ctx) + require.NoError(t, startSnapshotOffload(workerCtx, workers, nil, t.TempDir(), "n1", false, nil, testLogger(t))) + require.Eventually(t, func() bool { + _, exists, headErr := objects.HeadObject(context.Background(), stale.Payload.Key) + return headErr == nil && !exists + }, 5*time.Second, 10*time.Millisecond) + cancel() + require.NoError(t, workers.Wait()) + _, exists, err := objects.HeadObject(context.Background(), stale.ManifestKey) + require.NoError(t, err) + require.False(t, exists) + groupID := uint64(1) + result, err := snapshotoffload.RestorePhysicalSnapshot(context.Background(), snapshotoffload.RestoreOptions{ + Store: objects, ManifestKey: latest.ManifestKey, DataDir: filepath.Join(t.TempDir(), "restored"), + ExpectGroupID: &groupID, ExpectSourceCluster: "cluster-a", + Peers: []etcdraftengine.Peer{{NodeID: 2, ID: "n2", Address: "127.0.0.1:12002"}}, + }) + require.NoError(t, err) + require.Equal(t, latest.Payload.SHA256, result.PayloadSHA256) +} + +type blockingRetentionStore struct { + snapshotOffloadStore + calls atomic.Int32 +} + +func (s *blockingRetentionStore) ValidateRetention(ctx context.Context) error { + if s.calls.Add(1) == 1 { + <-ctx.Done() + return ctx.Err() + } + return s.snapshotOffloadStore.ValidateRetention(ctx) +} + +func TestSnapshotOffloadGCTimeoutRetriesAndShutdownClosesStore(t *testing.T) { + store, err := snapshotoffload.NewLocalStore(t.TempDir()) + require.NoError(t, err) + backend := &blockingRetentionStore{snapshotOffloadStore: store} + gc, err := snapshotoffload.NewGC(snapshotoffload.GCOptions{ + Store: backend, Prefix: "cluster-a", VerifyPublishersUseClaims: func(ctx context.Context) error { return ctx.Err() }, + }) + require.NoError(t, err) + retention := &snapshotOffloadRetention{gc: gc, interval: time.Millisecond, timeout: 10 * time.Millisecond, logger: testLogger(t)} + scheduler, err := snapshotoffload.NewScheduler(store, nil, "cluster-a", "cluster-a", "test") + require.NoError(t, err) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + done := make(chan error, 1) + go func() { done <- runSnapshotOffloadWorkers(ctx, store, scheduler, retention) }() + require.Eventually(t, func() bool { return backend.calls.Load() >= 2 }, time.Second, time.Millisecond) + cancel() + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("snapshot workers did not stop after cancellation") + } + _, _, err = store.HeadObject(context.Background(), "closed") + require.ErrorIs(t, err, snapshotoffload.ErrInvalidOptions) +} diff --git a/main_snapshot_offload_test.go b/main_snapshot_offload_test.go new file mode 100644 index 000000000..0e6e8ad8f --- /dev/null +++ b/main_snapshot_offload_test.go @@ -0,0 +1,468 @@ +package main + +import ( + "context" + "encoding/pem" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/bootjp/elastickv/internal/snapshotoffload" + "github.com/cockroachdb/errors" + "github.com/stretchr/testify/require" +) + +// withOffloadFlags sets the offload flags for one test and restores +// them afterwards. The flags are process globals, so a test that left +// them set would enable offload for every later test in the package. +func withOffloadFlags(t *testing.T, bucket, localDir string) { + t.Helper() + origBucket, origLocal := *snapshotOffloadBucket, *snapshotOffloadLocalDir + *snapshotOffloadBucket, *snapshotOffloadLocalDir = bucket, localDir + t.Cleanup(func() { + *snapshotOffloadBucket, *snapshotOffloadLocalDir = origBucket, origLocal + }) +} + +// TestSnapshotOffloadIsOptIn pins that a node which configured no +// destination does no offload work and cannot fail startup on offload +// configuration. +func TestSnapshotOffloadIsOptIn(t *testing.T) { + withOffloadFlags(t, "", "") + require.False(t, snapshotOffloadEnabled()) + require.NoError(t, startSnapshotOffload( + context.Background(), nil, nil, t.TempDir(), "n1", false, nil, testLogger(t))) +} + +// TestSnapshotOffloadRejectsAmbiguousDestination guards against +// accepting both a bucket and a local dir: which destination actually +// receives the artifacts would be ambiguous, and a backup written to +// the wrong place is discovered only when a restore is attempted. +func TestSnapshotOffloadRejectsAmbiguousDestination(t *testing.T) { + withOffloadFlags(t, "some-bucket", t.TempDir()) + require.True(t, snapshotOffloadEnabled()) + + _, err := buildSnapshotOffloadStore(context.Background()) + require.Error(t, err) + require.True(t, errors.Is(err, snapshotoffload.ErrInvalidOptions)) + require.ErrorContains(t, err, "mutually exclusive") +} + +func TestSnapshotOffloadBuildsALocalStore(t *testing.T) { + root := t.TempDir() + withOffloadFlags(t, "", root) + + store, err := buildSnapshotOffloadStore(context.Background()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, closeSnapshotOffloadStore(store)) }) + require.NotNil(t, store) + _, ok := store.(*snapshotoffload.LocalStore) + require.True(t, ok) +} + +// TestSnapshotOffloadGroupsCarryPerGroupDataDirs pins that each group +// is pointed at its own Raft data dir. Publishing a group's snapshot +// from another group's directory would ship the wrong state under the +// right manifest identity. +func TestSnapshotOffloadGroupsCarryPerGroupDataDirs(t *testing.T) { + raftDir := t.TempDir() + runtimes := []*raftGroupRuntime{ + {spec: groupSpec{id: 1}}, + {spec: groupSpec{id: 2}}, + nil, // a nil runtime must be skipped, not panic + } + + groups := snapshotOffloadGroups(runtimes, raftDir, "n1", true) + require.Len(t, groups, 2) + + seen := map[uint64]string{} + for _, g := range groups { + require.NotNil(t, g.IsLeader, "every group must carry both leadership callbacks") + require.NotNil(t, g.VerifyLeader) + seen[g.GroupID] = g.DataDir + } + require.Equal(t, filepath.Join(raftDir, "n1", "group-1"), seen[1]) + require.Equal(t, filepath.Join(raftDir, "n1", "group-2"), seen[2]) + require.NotEqual(t, seen[1], seen[2]) +} + +// TestSnapshotOffloadLeadershipFailsClosedOnAClosedEngine covers +// shutdown: the scheduler outlives startup and races Close(), so a +// runtime whose engine has been cleared must report "not leader" +// rather than panic or, worse, publish. +func TestSnapshotOffloadLeadershipFailsClosedOnAClosedEngine(t *testing.T) { + rt := &raftGroupRuntime{spec: groupSpec{id: 7}} // engine never set + + require.False(t, snapshotOffloadIsLeader(rt)(), + "a closed engine must never look like a leader") + + err := snapshotOffloadVerifyLeader(rt)(context.Background()) + require.Error(t, err) + require.True(t, errors.Is(err, snapshotoffload.ErrInvalidOptions)) +} + +// TestStartSnapshotOffloadRejectsIncompleteConfiguration pins that a +// configured-but-invalid offload fails startup rather than logging and +// leaving the operator with no backups. +func TestStartSnapshotOffloadRejectsIncompleteConfiguration(t *testing.T) { + withOffloadFlags(t, "", t.TempDir()) + origCluster := *snapshotOffloadSourceCluster + *snapshotOffloadSourceCluster = " " // whitespace-only: no identity + t.Cleanup(func() { *snapshotOffloadSourceCluster = origCluster }) + + err := startSnapshotOffload( + context.Background(), nil, + []*raftGroupRuntime{{spec: groupSpec{id: 1}}}, + t.TempDir(), "n1", false, nil, testLogger(t)) + require.Error(t, err) + require.True(t, errors.Is(err, snapshotoffload.ErrInvalidOptions)) +} + +// testLogger discards output so a test that exercises the enabled path +// does not spam the run. +func testLogger(t *testing.T) *slog.Logger { + t.Helper() + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// TestRunbookRestorePathsMatchGroupDataDir keeps the operations +// runbook's `--data-dir` table honest against the function the server +// actually uses. +// +// A wrong path here is not a cosmetic doc bug: an operator following +// it during disaster recovery restores into a directory the server +// never opens, startup finds the per-group directories empty, and the +// restore is silently ignored. +func TestRunbookRestorePathsMatchGroupDataDir(t *testing.T) { + t.Parallel() + + const raftDir = "/var/lib/elastickv" + const raftID = "n1" + + tests := []struct { + name string + groupID uint64 + multi bool + want string + }{ + {name: "multi-group", groupID: 1, multi: true, want: "/var/lib/elastickv/n1/group-1"}, + {name: "multi-group higher id", groupID: 7, multi: true, want: "/var/lib/elastickv/n1/group-7"}, + {name: "single group", groupID: 1, multi: false, want: "/var/lib/elastickv/n1"}, + {name: "single node group zero", groupID: 0, multi: false, want: "/var/lib/elastickv/n1/group-0"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + require.Equal(t, tc.want, groupDataDir(raftDir, raftID, tc.groupID, tc.multi), + "docs/snapshot_offload_operations.md documents this path for restore") + }) + } +} + +// TestRunbookRestorePathsFollowFromTheGroupTopology pins the runbook table the +// way an operator actually reads it: from --raftGroups to a directory. +// +// TestRunbookRestorePathsMatchGroupDataDir above takes `multi` as an input, so +// it cannot catch the case where a reader derives the wrong `multi` in the first +// place — and that is the case that bites. dataGroupsNeedMultiDirs counts DATA +// groups and excludes group 0, so a node running the dedicated TSO group +// alongside a single data group has two entries in --raftGroups but is NOT +// multi-dir: group 0 lands in group-0 while the data group opens +// / directly. Restoring that data group into group-1 puts it +// where startup never looks, and an empty group is not an error. +func TestRunbookRestorePathsFollowFromTheGroupTopology(t *testing.T) { + t.Parallel() + + const ( + raftDir = "/var/lib/elastickv" + raftID = "n1" + ) + spec := func(ids ...uint64) []groupSpec { + out := make([]groupSpec, 0, len(ids)) + for _, id := range ids { + out = append(out, groupSpec{id: id, address: "127.0.0.1:50051"}) + } + return out + } + + for _, tc := range []struct { + name string + groups []groupSpec + groupID uint64 + want string + }{ + { + name: "two data groups: each gets its own dir", + groups: spec(1, 2), groupID: 1, + want: "/var/lib/elastickv/n1/group-1", + }, + { + name: "two data groups: the second one too", + groups: spec(1, 2), groupID: 2, + want: "/var/lib/elastickv/n1/group-2", + }, + { + name: "a single data group opens the node dir", + groups: spec(1), groupID: 1, + want: "/var/lib/elastickv/n1", + }, + { + name: "dedicated TSO plus one data group: group 0 is always group-0", + groups: spec(0, 1), groupID: 0, + want: "/var/lib/elastickv/n1/group-0", + }, + { + // The row that catches people out: two --raftGroups entries but + // only one DATA group, so this is not a multi-dir deployment. + name: "dedicated TSO plus one data group: the data group is NOT group-1", + groups: spec(0, 1), groupID: 1, + want: "/var/lib/elastickv/n1", + }, + { + name: "dedicated TSO plus two data groups is multi-dir again", + groups: spec(0, 1, 2), groupID: 1, + want: "/var/lib/elastickv/n1/group-1", + }, + { + name: "dedicated TSO plus two data groups: group 0 unchanged", + groups: spec(0, 1, 2), groupID: 0, + want: "/var/lib/elastickv/n1/group-0", + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + // requested=true is the operator asking for per-group dirs; the + // topology decides whether that takes effect. + multi := effectiveMultiDataDirs(tc.groups, true) + require.Equal(t, tc.want, groupDataDir(raftDir, raftID, tc.groupID, multi), + "docs/snapshot_offload_operations.md documents this path for restore") + }) + } +} + +// TestRejectPlaintextOffloadEndpoint pins the transport requirement. +// +// The design and the runbook both require TLS for the external bucket, and this +// path carries whole snapshot payloads plus the credentials used to write them. +// An http:// endpoint was forwarded straight to the AWS client, so the +// downgrade was invisible until someone captured the traffic. +func TestRejectPlaintextOffloadEndpoint(t *testing.T) { + restore := func(endpoint string, allow bool) func() { + prevEndpoint := *snapshotOffloadEndpoint + prevAllow := *snapshotOffloadAllowInsecureEndpoint + *snapshotOffloadEndpoint = endpoint + *snapshotOffloadAllowInsecureEndpoint = allow + return func() { + *snapshotOffloadEndpoint = prevEndpoint + *snapshotOffloadAllowInsecureEndpoint = prevAllow + } + } + + for _, tc := range []struct { + name string + endpoint string + allow bool + wantErr bool + }{ + {name: "https is accepted", endpoint: "https://s3.example.com", wantErr: false}, + {name: "uppercase HTTPS is accepted", endpoint: "HTTPS://s3.example.com", wantErr: false}, + {name: "empty means the AWS default, which is https", endpoint: "", wantErr: false}, + {name: "https without a host is refused", endpoint: "https://", wantErr: true}, + {name: "opaque https URL is refused", endpoint: "https:example.com", wantErr: true}, + {name: "malformed development endpoint is refused", endpoint: "http://", allow: true, wantErr: true}, + {name: "http is refused", endpoint: "http://s3.example.com", wantErr: true}, + { + // The AWS SDK resolves a scheme-less endpoint as http, so it is + // refused rather than assumed safe. + name: "scheme-less host:port is refused", endpoint: "s3.example.com:9000", wantErr: true, + }, + { + name: "http with the explicit development opt-in is allowed", + endpoint: "http://127.0.0.1:9000", allow: true, wantErr: false, + }, + } { + t.Run(tc.name, func(t *testing.T) { + defer restore(tc.endpoint, tc.allow)() + err := rejectPlaintextOffloadEndpoint() + if tc.wantErr { + require.Error(t, err, "endpoint %q", tc.endpoint) + require.ErrorIs(t, err, snapshotoffload.ErrInvalidOptions) + return + } + require.NoError(t, err, "endpoint %q", tc.endpoint) + }) + } +} + +func TestSnapshotOffloadRedirectPolicy(t *testing.T) { + t.Parallel() + + origin, err := http.NewRequestWithContext(context.Background(), http.MethodPut, + "https://s3.example.com/object", nil) + require.NoError(t, err) + + redirect := func(rawURL string, status int) *http.Request { + t.Helper() + u, parseErr := url.Parse(rawURL) + require.NoError(t, parseErr) + return &http.Request{ + Method: http.MethodPut, + URL: u, + Header: make(http.Header), + Response: &http.Response{StatusCode: status}, + } + } + + httpsRedirect := redirect("https://s3.example.com/next", http.StatusTemporaryRedirect) + require.NoError(t, checkSnapshotOffloadRedirect(httpsRedirect, []*http.Request{origin}, false)) + + plaintextRedirect := redirect("http://s3.example.com/next", http.StatusTemporaryRedirect) + require.ErrorIs(t, checkSnapshotOffloadRedirect(plaintextRedirect, []*http.Request{origin}, false), + snapshotoffload.ErrInvalidOptions) + + unsupportedRedirect := redirect("https://s3.example.com/next", http.StatusFound) + require.ErrorIs(t, checkSnapshotOffloadRedirect(unsupportedRedirect, []*http.Request{origin}, false), + http.ErrUseLastResponse) + + crossHost := redirect("https://other.example.com/next", http.StatusPermanentRedirect) + require.ErrorIs(t, checkSnapshotOffloadRedirect(crossHost, []*http.Request{origin}, false), + snapshotoffload.ErrInvalidOptions) +} + +func TestSnapshotOffloadInsecureClientStillRestrictsRedirects(t *testing.T) { + previous := *snapshotOffloadAllowInsecureEndpoint + *snapshotOffloadAllowInsecureEndpoint = true + t.Cleanup(func() { *snapshotOffloadAllowInsecureEndpoint = previous }) + + base := &http.Client{Transport: &recordingRoundTripper{}} + wrapped, err := snapshotOffloadHTTPClient(base) + require.NoError(t, err) + client, ok := wrapped.(*http.Client) + require.True(t, ok) + origin, err := http.NewRequestWithContext(context.Background(), http.MethodPut, + "http://s3.example.com/object", nil) + require.NoError(t, err) + response, err := client.Do(origin) + require.NoError(t, err) + require.NoError(t, response.Body.Close()) + redirect := origin.Clone(context.Background()) + redirect.URL, err = url.Parse("http://other.s3.example.com/object") + require.NoError(t, err) + redirect.Response = &http.Response{StatusCode: http.StatusTemporaryRedirect} + require.ErrorIs(t, client.CheckRedirect(redirect, []*http.Request{origin}), + snapshotoffload.ErrInvalidOptions) +} + +func TestSnapshotOffloadPreservesSDKCABundle(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + })) + t.Cleanup(server.Close) + caPath := filepath.Join(t.TempDir(), "ca.pem") + require.NoError(t, os.WriteFile(caPath, pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", Bytes: server.Certificate().Raw, + }), 0o600)) + t.Setenv("AWS_CA_BUNDLE", caPath) + t.Setenv("AWS_EC2_METADATA_DISABLED", "true") + previous := *snapshotOffloadAllowInsecureEndpoint + *snapshotOffloadAllowInsecureEndpoint = false + t.Cleanup(func() { *snapshotOffloadAllowInsecureEndpoint = previous }) + + var resolved aws.HTTPClient + store, err := snapshotoffload.NewS3Store(context.Background(), snapshotoffload.S3StoreConfig{ + Bucket: "snapshot-test", Region: "us-east-1", Endpoint: server.URL, + ForcePathStyle: true, ServerSideEncryption: "AES256", + AccessKeyID: "test-key", SecretAccessKey: "test-secret", + HTTPClientDecorator: func(base aws.HTTPClient) (aws.HTTPClient, error) { + resolved = base + return snapshotOffloadHTTPClient(base) + }, + }) + require.NoError(t, err) + require.NotNil(t, resolved, "the SDK must supply the CA-configured client") + _, found, err := store.HeadObject(context.Background(), "snapshot") + require.NoError(t, err, "the guarded client must trust the SDK CA bundle") + require.False(t, found) +} + +type recordingRoundTripper struct { + called bool +} + +func (r *recordingRoundTripper) RoundTrip(*http.Request) (*http.Response, error) { + r.called = true + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader("")), + }, nil +} + +func TestSnapshotOffloadTransportRejectsSDKResolvedPlaintextEndpoint(t *testing.T) { + t.Parallel() + + next := &recordingRoundTripper{} + transport := snapshotOffloadHTTPSTransport{next: next} + request, err := http.NewRequestWithContext(context.Background(), http.MethodPut, + "http://s3.example.com/object", nil) + require.NoError(t, err) + + _, err = transport.RoundTrip(request) + require.ErrorIs(t, err, snapshotoffload.ErrInvalidOptions) + require.False(t, next.called, "plaintext must be rejected before credentials or payload reach the transport") + + request.URL.Scheme = "https" + response, err := transport.RoundTrip(request) + require.NoError(t, err) + require.True(t, next.called) + require.NoError(t, response.Body.Close()) +} + +func TestSnapshotOffloadRejectsEscapingPrefixBeforeScheduling(t *testing.T) { + t.Parallel() + + for _, prefix := range []string{"..", "../backup", "snapshots/../../backup", "/../backup"} { + require.ErrorIs(t, validateSnapshotOffloadPrefix(prefix), snapshotoffload.ErrInvalidOptions, + "prefix %q", prefix) + } + for _, prefix := range []string{"", "/", "snapshots", "/snapshots/group-a/"} { + require.NoError(t, validateSnapshotOffloadPrefix(prefix), "prefix %q", prefix) + } +} + +func TestSnapshotOffloadRejectsSpoolPathThatIsAFile(t *testing.T) { + spoolFile := filepath.Join(t.TempDir(), "spool") + require.NoError(t, os.WriteFile(spoolFile, []byte("not a directory"), 0o600)) + + previous := *snapshotOffloadSpoolDir + *snapshotOffloadSpoolDir = spoolFile + t.Cleanup(func() { *snapshotOffloadSpoolDir = previous }) + + require.Error(t, validateSnapshotOffloadSpoolDir()) +} + +func TestSnapshotOffloadCleanupSparesRecentSharedSpoolFiles(t *testing.T) { + spoolDir := t.TempDir() + spoolFile := filepath.Join(spoolDir, "elastickv-snapshot-offload-shared.fsm") + require.NoError(t, os.WriteFile(spoolFile, []byte("in flight"), 0o600)) + + previous := *snapshotOffloadSpoolDir + *snapshotOffloadSpoolDir = spoolDir + t.Cleanup(func() { *snapshotOffloadSpoolDir = previous }) + runtimes := []*raftGroupRuntime{{spec: groupSpec{id: 1}}} + + cleanStaleOffloadSpool(runtimes, t.TempDir(), "n1", false, testLogger(t)) + require.FileExists(t, spoolFile) + + stale := time.Now().Add(-snapshotOffloadStaleSpoolAge - time.Hour) + require.NoError(t, os.Chtimes(spoolFile, stale, stale)) + cleanStaleOffloadSpool(runtimes, t.TempDir(), "n1", false, testLogger(t)) + require.NoFileExists(t, spoolFile) +} diff --git a/monitoring/registry.go b/monitoring/registry.go index 2ae7698dd..74f377618 100644 --- a/monitoring/registry.go +++ b/monitoring/registry.go @@ -31,6 +31,7 @@ type Registry struct { coldStartObs *ColdStartObserver tso *TSOMetrics tsoObserver *TSOObserver + snapOffload *SnapshotOffloadMetrics encryption *EncryptionMetrics encryptionObs *EncryptionStateObserver redisTTLInline *RedisTTLInlineMigrationMetrics @@ -66,6 +67,7 @@ func NewRegistry(nodeID string, nodeAddress string) *Registry { r.coldStartObs = newColdStartObserver(r.coldStart) r.tso = newTSOMetrics(registerer) r.tsoObserver = newTSOObserver(r.tso) + r.snapOffload = newSnapshotOffloadMetrics(registerer) r.encryption = newEncryptionMetrics(registerer) r.encryptionObs = newEncryptionStateObserver(r.encryption) r.redisTTLInline = newRedisTTLInlineMigrationMetrics(registerer) @@ -308,6 +310,16 @@ func (r *Registry) TSOObserver() *TSOObserver { return r.tsoObserver } +// SnapshotOffloadObserver returns the physical snapshot offload +// scheduler's metrics observer. Passed to the scheduler through +// snapshotoffload.WithSchedulerObserver. +func (r *Registry) SnapshotOffloadObserver() *SnapshotOffloadMetrics { + if r == nil { + return nil + } + return r.snapOffload +} + // EncryptionObserver returns the data-at-rest encryption observer // backed by this registry. The storage layer receives it through // store.WithEncryptionObserver and calls it on every envelope emit diff --git a/monitoring/snapshot_offload.go b/monitoring/snapshot_offload.go new file mode 100644 index 000000000..a73cbb486 --- /dev/null +++ b/monitoring/snapshot_offload.go @@ -0,0 +1,161 @@ +package monitoring + +import ( + "strconv" + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +// snapshotPayloadBucketBase is the smallest payload-size histogram +// bucket (1 MiB). Snapshots below it are rounding error next to the +// multi-gigabyte cases the histogram exists to show. +const ( + snapshotPayloadBucketBase = 1 << 20 // 1 MiB + snapshotPayloadBucketFactor = 4 + snapshotPayloadBucketCount = 8 // 1 MiB through ~16 GiB + snapshotOffloadGroupIDLabel = "group_id" + snapshotOffloadReasonLabel = "reason" +) + +// SnapshotOffloadMetrics exposes the physical snapshot offload +// scheduler's outcomes (design doc §4). +// +// group_id is a label on every series: its cardinality is the number +// of Raft groups this process hosts, which is bounded by deployment +// topology rather than by traffic. skip reason is a closed set owned +// by the scheduler. +type SnapshotOffloadMetrics struct { + published *prometheus.CounterVec + skipped *prometheus.CounterVec + failed *prometheus.CounterVec + lastPublishIndex *prometheus.GaugeVec + publishSeconds *prometheus.HistogramVec + payloadBytes *prometheus.HistogramVec +} + +func newSnapshotOffloadMetrics(registerer prometheus.Registerer) *SnapshotOffloadMetrics { + m := &SnapshotOffloadMetrics{ + published: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "elastickv_snapshot_offload_published_total", + Help: "Total physical snapshots published to the object store, by Raft group.", + }, + []string{snapshotOffloadGroupIDLabel}, + ), + skipped: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "elastickv_snapshot_offload_skipped_total", + Help: "Total offload scans that published nothing, by Raft group and reason. Routine on a follower or an unchanged snapshot.", + }, + []string{snapshotOffloadGroupIDLabel, snapshotOffloadReasonLabel}, + ), + failed: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "elastickv_snapshot_offload_failed_total", + Help: "Total offload attempts that failed, by Raft group. A sustained rate means backups are not being taken.", + }, + []string{snapshotOffloadGroupIDLabel}, + ), + lastPublishIndex: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "elastickv_snapshot_offload_last_published_index", + Help: "Raft index of the most recent snapshot this process published, by group. Staleness here is the backup-freshness signal.", + }, + []string{snapshotOffloadGroupIDLabel}, + ), + publishSeconds: prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "elastickv_snapshot_offload_publish_seconds", + Help: "Wall time to spool, upload and commit one snapshot.", + Buckets: []float64{0.5, 1, 5, 15, 30, 60, 300, 900, 1800, 3600}, + }, + []string{snapshotOffloadGroupIDLabel}, + ), + payloadBytes: prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "elastickv_snapshot_offload_payload_bytes", + Help: "Size of each published snapshot payload.", + Buckets: prometheus.ExponentialBuckets( + snapshotPayloadBucketBase, + snapshotPayloadBucketFactor, + snapshotPayloadBucketCount, + ), + }, + []string{snapshotOffloadGroupIDLabel}, + ), + } + registerer.MustRegister( + m.published, + m.skipped, + m.failed, + m.lastPublishIndex, + m.publishSeconds, + m.payloadBytes, + ) + return m +} + +// ObserveSnapshotOffloadPublished records one successful publication. +func (m *SnapshotOffloadMetrics) ObserveSnapshotOffloadPublished( + groupID, index uint64, payloadBytes int64, elapsed time.Duration, +) { + if m == nil { + return + } + label := snapshotOffloadGroupLabel(groupID) + m.published.WithLabelValues(label).Inc() + m.lastPublishIndex.WithLabelValues(label).Set(float64(index)) + m.publishSeconds.WithLabelValues(label).Observe(max(0, elapsed).Seconds()) + m.payloadBytes.WithLabelValues(label).Observe(float64(max(int64(0), payloadBytes))) +} + +// ObserveSnapshotOffloadSkipped records a scan that published nothing. +func (m *SnapshotOffloadMetrics) ObserveSnapshotOffloadSkipped(groupID uint64, reason string) { + if m == nil { + return + } + m.skipped.WithLabelValues(snapshotOffloadGroupLabel(groupID), normalizeSnapshotOffloadSkip(reason)).Inc() +} + +// ObserveSnapshotOffloadFailed records a failed attempt. The error is +// deliberately not a label: its text is unbounded, and a per-message +// series would let one recurring failure explode the metric's +// cardinality. Diagnosis comes from the scheduler's log line. +func (m *SnapshotOffloadMetrics) ObserveSnapshotOffloadFailed(groupID uint64, _ error) { + if m == nil { + return + } + m.failed.WithLabelValues(snapshotOffloadGroupLabel(groupID)).Inc() +} + +func snapshotOffloadGroupLabel(groupID uint64) string { + return strconv.FormatUint(groupID, 10) +} + +// Skip reasons emitted by the scheduler. +const ( + snapshotOffloadSkipNotLeader = "not_leader" + snapshotOffloadSkipAlreadyPublished = "already_published" + snapshotOffloadSkipNoSnapshot = "no_persisted_snapshot" + snapshotOffloadSkipInFlight = "already_in_flight" + snapshotOffloadSkipUnknownLeader = "leadership_unknown" + snapshotOffloadSkipObjectClaimed = "object_claimed" + snapshotOffloadSkipUnknown = "unknown" +) + +// normalizeSnapshotOffloadSkip keeps the reason label inside the +// scheduler's closed set. +func normalizeSnapshotOffloadSkip(reason string) string { + switch reason { + case snapshotOffloadSkipNotLeader, + snapshotOffloadSkipAlreadyPublished, + snapshotOffloadSkipNoSnapshot, + snapshotOffloadSkipInFlight, + snapshotOffloadSkipUnknownLeader, + snapshotOffloadSkipObjectClaimed: + return reason + default: + return snapshotOffloadSkipUnknown + } +} diff --git a/monitoring/snapshot_offload_test.go b/monitoring/snapshot_offload_test.go new file mode 100644 index 000000000..a81204c36 --- /dev/null +++ b/monitoring/snapshot_offload_test.go @@ -0,0 +1,99 @@ +package monitoring + +import ( + "errors" + "strings" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" +) + +func TestSnapshotOffloadMetricsRecordOutcomes(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + m := newSnapshotOffloadMetrics(reg) + + m.ObserveSnapshotOffloadPublished(7, 4211, 5<<20, 12*time.Second) + m.ObserveSnapshotOffloadSkipped(7, snapshotOffloadSkipNotLeader) + m.ObserveSnapshotOffloadSkipped(7, snapshotOffloadSkipAlreadyPublished) + m.ObserveSnapshotOffloadFailed(9, errors.New("object store unavailable")) + + require.NoError(t, testutil.GatherAndCompare( + reg, + strings.NewReader(` +# HELP elastickv_snapshot_offload_published_total Total physical snapshots published to the object store, by Raft group. +# TYPE elastickv_snapshot_offload_published_total counter +elastickv_snapshot_offload_published_total{group_id="7"} 1 +# HELP elastickv_snapshot_offload_last_published_index Raft index of the most recent snapshot this process published, by group. Staleness here is the backup-freshness signal. +# TYPE elastickv_snapshot_offload_last_published_index gauge +elastickv_snapshot_offload_last_published_index{group_id="7"} 4211 +# HELP elastickv_snapshot_offload_failed_total Total offload attempts that failed, by Raft group. A sustained rate means backups are not being taken. +# TYPE elastickv_snapshot_offload_failed_total counter +elastickv_snapshot_offload_failed_total{group_id="9"} 1 +# HELP elastickv_snapshot_offload_skipped_total Total offload scans that published nothing, by Raft group and reason. Routine on a follower or an unchanged snapshot. +# TYPE elastickv_snapshot_offload_skipped_total counter +elastickv_snapshot_offload_skipped_total{group_id="7",reason="already_published"} 1 +elastickv_snapshot_offload_skipped_total{group_id="7",reason="not_leader"} 1 +`), + "elastickv_snapshot_offload_published_total", + "elastickv_snapshot_offload_last_published_index", + "elastickv_snapshot_offload_failed_total", + "elastickv_snapshot_offload_skipped_total", + )) +} + +// TestSnapshotOffloadMetricsBoundTheSkipReasonLabel is the cardinality +// guard: an unrecognised reason must collapse rather than mint a +// series per distinct string. +func TestSnapshotOffloadMetricsBoundTheSkipReasonLabel(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + m := newSnapshotOffloadMetrics(reg) + + m.ObserveSnapshotOffloadSkipped(1, "something-new") + m.ObserveSnapshotOffloadSkipped(1, "something-else") + m.ObserveSnapshotOffloadSkipped(1, "") + m.ObserveSnapshotOffloadSkipped(1, "object_claimed") + + require.Equal(t, 2, testutil.CollectAndCount(m.skipped)) + require.InDelta(t, 1.0, + testutil.ToFloat64(m.skipped.WithLabelValues("1", snapshotOffloadSkipObjectClaimed)), 0.0001) + require.InDelta(t, 3.0, + testutil.ToFloat64(m.skipped.WithLabelValues("1", snapshotOffloadSkipUnknown)), 0.0001) +} + +// TestSnapshotOffloadMetricsDoNotLabelByError pins that the failure +// counter carries no error text: messages are unbounded, and one +// recurring failure would otherwise explode the metric's cardinality. +func TestSnapshotOffloadMetricsDoNotLabelByError(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + m := newSnapshotOffloadMetrics(reg) + + for i := range 20 { + m.ObserveSnapshotOffloadFailed(1, errors.New(strings.Repeat("x", i+1))) + } + require.Equal(t, 1, testutil.CollectAndCount(m.failed), + "distinct error texts must not create distinct series") +} + +func TestSnapshotOffloadMetricsNilReceiverIsInert(t *testing.T) { + t.Parallel() + + var m *SnapshotOffloadMetrics + require.NotPanics(t, func() { + m.ObserveSnapshotOffloadPublished(1, 2, 3, time.Second) + m.ObserveSnapshotOffloadSkipped(1, "x") + m.ObserveSnapshotOffloadFailed(1, errors.New("boom")) + }) + require.NotNil(t, NewRegistry("n1", "127.0.0.1:1").SnapshotOffloadObserver()) + + var nilRegistry *Registry + require.Nil(t, nilRegistry.SnapshotOffloadObserver()) +}