From 552356f1184ad9e13ad1103d01c41073293f7e15 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 22:20:35 +0900 Subject: [PATCH 1/5] Wire snapshot offload into the server runtime Validate enabled offload configuration, constrain S3 transport without changing credential discovery, preserve explicit legacy restore, and publish bounded metrics with an operations runbook. --- cmd/elastickv-snapshot-offload/main.go | 55 ++- cmd/elastickv-snapshot-offload/main_test.go | 250 +++++++++++ ...artial_physical_snapshot_object_offload.md | 16 +- docs/snapshot_offload_operations.md | 294 +++++++++++++ internal/snapshotoffload/manifest.go | 19 +- internal/snapshotoffload/offload_test.go | 294 +++++++++++-- internal/snapshotoffload/publish.go | 105 ++++- internal/snapshotoffload/restore.go | 80 ++++ internal/snapshotoffload/s3_store.go | 7 +- internal/snapshotoffload/s3_store_test.go | 24 ++ main.go | 11 + main_snapshot_offload.go | 381 ++++++++++++++++ main_snapshot_offload_test.go | 407 ++++++++++++++++++ monitoring/registry.go | 12 + monitoring/snapshot_offload.go | 157 +++++++ monitoring/snapshot_offload_test.go | 96 +++++ 16 files changed, 2153 insertions(+), 55 deletions(-) create mode 100644 docs/snapshot_offload_operations.md create mode 100644 main_snapshot_offload.go create mode 100644 main_snapshot_offload_test.go create mode 100644 monitoring/snapshot_offload.go create mode 100644 monitoring/snapshot_offload_test.go diff --git a/cmd/elastickv-snapshot-offload/main.go b/cmd/elastickv-snapshot-offload/main.go index efdd1640d..71e652b1d 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" @@ -58,6 +59,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() { @@ -122,7 +130,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 { @@ -134,8 +142,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 @@ -151,6 +165,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) } @@ -166,6 +190,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 } @@ -258,11 +293,17 @@ func runRestore(ctx context.Context, cfg *restoreConfig, logger *slog.Logger) er 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 2956620aa..2bd15978a 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 pending +Status: Partial — M0/M1/M2 implemented; M3 runtime wiring and the operations runbook implemented; retention/GC, restore drills, multi-node acceptance, and the versioned-bucket decision pending Author: bootjp Date: 2026-07-19 -Updated: 2026-07-23 +Updated: 2026-09-25 ## 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 and retention/GC remain pending. +The runtime scheduler is implemented and wired into main.go, opt-in via +`--snapshotOffloadBucket` (or `--snapshotOffloadLocalDir`). Retention/GC is +implemented in PR #1222, which remains unmerged and is not included here. +Retention/GC integration, +restore drills, multi-node acceptance, and the §7 versioned-bucket decision +remain pending on this branch; the operator runbook is at +[`../snapshot_offload_operations.md`](../snapshot_offload_operations.md). ## 2. Safety boundary @@ -164,8 +170,8 @@ permissions below the configured prefix. |---|---|---| | 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 | 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. Shipped on this branch: the runtime wiring (`main_snapshot_offload.go`) that runs the scheduler in-process, its metrics (`monitoring/snapshot_offload.go`), and the operational runbook (`docs/snapshot_offload_operations.md`). The §5 two-phase retention/GC is **implemented in PR #1222 but remains unmerged and is not included here**. Until it lands, published manifests and payloads are never reclaimed by this branch, so operators need an external bucket lifecycle policy. Also pending: restore corruption drills, multi-node acceptance, and the §7 versioned-bucket decision, which is why the `partial` marker stays. | 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 diff --git a/docs/snapshot_offload_operations.md b/docs/snapshot_offload_operations.md new file mode 100644 index 000000000..3355ce06d --- /dev/null +++ b/docs/snapshot_offload_operations.md @@ -0,0 +1,294 @@ +# 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). + +> **Note:** §4 (Retention) describes the retention/GC subsystem implemented in +> PR #1222. That PR remains unmerged and its implementation is not included +> here. Everything else here is live once this change ships. + +## 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 immutable and self-hashing. A manifest names exactly one payload. + +## 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. | +| `--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. | +| `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 + +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. + +The second pass exists because a publisher reusing a payload rewrites identical +bytes, which no general-purpose S3 precondition can detect (`If-Match` compares +a content-derived ETag; `IfMatchLastModifiedTime` is directory-buckets only). +**`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. + +**A versioned backup bucket requires a noncurrent-version expiration lifecycle +rule.** Whether to instead enumerate versions directly, or refuse versioned +buckets at startup, is an open decision. + +## 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. +- **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. diff --git a/internal/snapshotoffload/manifest.go b/internal/snapshotoffload/manifest.go index ca30a7686..51b37744e 100644 --- a/internal/snapshotoffload/manifest.go +++ b/internal/snapshotoffload/manifest.go @@ -19,9 +19,22 @@ 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") + + // 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") // ErrNoPersistedSnapshot reports that the LOCAL data dir has no // persisted snapshot yet. It is deliberately distinct from diff --git a/internal/snapshotoffload/offload_test.go b/internal/snapshotoffload/offload_test.go index a8942cadd..8098dc8d1 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" @@ -58,6 +59,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) @@ -95,10 +98,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) @@ -288,26 +293,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) { @@ -376,10 +385,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) @@ -480,10 +491,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) } @@ -511,6 +524,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) @@ -534,10 +549,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) @@ -724,3 +741,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 d53914c0f..95289c9ce 100644 --- a/internal/snapshotoffload/publish.go +++ b/internal/snapshotoffload/publish.go @@ -308,13 +308,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) @@ -326,7 +425,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/s3_store.go b/internal/snapshotoffload/s3_store.go index f8819d8df..3cf82be09 100644 --- a/internal/snapshotoffload/s3_store.go +++ b/internal/snapshotoffload/s3_store.go @@ -6,6 +6,7 @@ import ( "encoding/base64" "encoding/hex" "io" + "net/http" "strings" "time" @@ -21,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) @@ -54,6 +55,7 @@ type S3StoreConfig struct { ServerSideEncryption string SSEKMSKeyID string DisableChecksumHeaders bool + HTTPClient *http.Client } type S3Store struct { @@ -81,6 +83,9 @@ func NewS3Store(ctx context.Context, cfg S3StoreConfig) (*S3Store, error) { } client = s3.NewFromConfig(awsCfg, func(o *s3.Options) { o.UsePathStyle = cfg.ForcePathStyle + if cfg.HTTPClient != nil { + o.HTTPClient = cfg.HTTPClient + } if stringsTrim(cfg.Endpoint) != "" { o.BaseEndpoint = aws.String(stringsTrim(cfg.Endpoint)) } diff --git a/internal/snapshotoffload/s3_store_test.go b/internal/snapshotoffload/s3_store_test.go index 010f65981..0571bd693 100644 --- a/internal/snapshotoffload/s3_store_test.go +++ b/internal/snapshotoffload/s3_store_test.go @@ -5,6 +5,7 @@ import ( "context" "fmt" "io" + "net/http" "strings" "sync" "testing" @@ -46,11 +47,34 @@ 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", + HTTPClient: client, + } + 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 TestS3StorePutHeadGetPreservesIntegrityMetadata(t *testing.T) { ctx := context.Background() fake := newFakeS3Client() diff --git a/main.go b/main.go index 6efe3f584..a847091ca 100644 --- a/main.go +++ b/main.go @@ -795,6 +795,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..7326b87c1 --- /dev/null +++ b/main_snapshot_offload.go @@ -0,0 +1,381 @@ +package main + +import ( + "context" + "flag" + "log/slog" + "net/http" + "net/url" + "os" + "path" + "strings" + "time" + + 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 +) + +// 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") + 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 == "" || *snapshotOffloadAllowInsecureEndpoint { + 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. + if !strings.EqualFold(parsed.Scheme, "https") || parsed.Host == "" || 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() *http.Client { + if *snapshotOffloadAllowInsecureEndpoint { + return nil + } + return &http.Client{ + Transport: snapshotOffloadHTTPSTransport{ + next: awshttp.NewBuildableClient().GetTransport(), + }, + CheckRedirect: checkSnapshotOffloadRedirect, + } +} + +type snapshotOffloadHTTPSTransport struct { + next http.RoundTripper +} + +func (t snapshotOffloadHTTPSTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if req == nil || req.URL == nil || !strings.EqualFold(req.URL.Scheme, "https") { + 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 checkSnapshotOffloadRedirect(req *http.Request, via []*http.Request) error { + if req == nil || req.URL == nil || !strings.EqualFold(req.URL.Scheme, "https") { + 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, 0o755); 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) (snapshotoffload.ObjectStore, 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), + HTTPClient: snapshotOffloadHTTPClient(), + }) + 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, +) error { + 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 + } + + opts := []snapshotoffload.SchedulerOption{ + snapshotoffload.WithSchedulerInterval(*snapshotOffloadInterval), + snapshotoffload.WithSchedulerConcurrency(*snapshotOffloadConcurrency), + 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)) + } + + scheduler, err := snapshotoffload.NewScheduler( + store, + snapshotOffloadGroups(runtimes, raftDir, raftID, multi), + strings.TrimSpace(*snapshotOffloadPrefix), + strings.TrimSpace(*snapshotOffloadSourceCluster), + buildVersion(), + opts..., + ) + if err != nil { + return errors.Wrap(err, "snapshot offload: scheduler") + } + + // 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. + if err := scheduler.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + return errors.Wrap(err, "snapshot offload scheduler") + } + return nil + }) + return nil +} diff --git a/main_snapshot_offload_test.go b/main_snapshot_offload_test.go new file mode 100644 index 000000000..6c53e59f3 --- /dev/null +++ b/main_snapshot_offload_test.go @@ -0,0 +1,407 @@ +package main + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "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) + 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: "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})) + + plaintextRedirect := redirect("http://s3.example.com/next", http.StatusTemporaryRedirect) + require.ErrorIs(t, checkSnapshotOffloadRedirect(plaintextRedirect, []*http.Request{origin}), + snapshotoffload.ErrInvalidOptions) + + unsupportedRedirect := redirect("https://s3.example.com/next", http.StatusFound) + require.ErrorIs(t, checkSnapshotOffloadRedirect(unsupportedRedirect, []*http.Request{origin}), + http.ErrUseLastResponse) + + crossHost := redirect("https://other.example.com/next", http.StatusPermanentRedirect) + require.ErrorIs(t, checkSnapshotOffloadRedirect(crossHost, []*http.Request{origin}), + snapshotoffload.ErrInvalidOptions) +} + +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 159bbe822..29dd21486 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 } @@ -65,6 +66,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) return r @@ -297,6 +299,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..6670a2bc4 --- /dev/null +++ b/monitoring/snapshot_offload.go @@ -0,0 +1,157 @@ +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 +) + +// 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{"group_id"}, + ), + 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{"group_id", "reason"}, + ), + 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{"group_id"}, + ), + 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{"group_id"}, + ), + 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{"group_id"}, + ), + 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{"group_id"}, + ), + } + 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" + 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: + 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..b2e68a822 --- /dev/null +++ b/monitoring/snapshot_offload_test.go @@ -0,0 +1,96 @@ +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, "") + + require.Equal(t, 1, testutil.CollectAndCount(m.skipped)) + 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()) +} From a0f0af4d9648492b880015934c5e2bcef2de6e62 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 22:23:25 +0900 Subject: [PATCH 2/5] Name the snapshot offload spool directory permissions --- main_snapshot_offload.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/main_snapshot_offload.go b/main_snapshot_offload.go index 7326b87c1..ca2b6343b 100644 --- a/main_snapshot_offload.go +++ b/main_snapshot_offload.go @@ -21,6 +21,7 @@ import ( const ( snapshotOffloadMaxRedirects = 10 snapshotOffloadStaleSpoolAge = 24 * time.Hour + snapshotOffloadSpoolDirPerm = 0o755 ) // Physical snapshot object offload (design doc §4 / §7). Opt-in: the @@ -166,7 +167,7 @@ func validateSnapshotOffloadSpoolDir() error { if dir == "" { return nil } - if err := os.MkdirAll(dir, 0o755); err != nil { + if err := os.MkdirAll(dir, snapshotOffloadSpoolDirPerm); err != nil { return errors.Wrap(err, "snapshot offload: create spool dir") } info, err := os.Stat(dir) From 76adca145a2dc850e463c52b7baa06ad9e9968a4 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 22:39:06 +0900 Subject: [PATCH 3/5] Preserve snapshot offload trust and bound publications Keep redirect guards active for development HTTP endpoints, decorate the resolved SDK client to preserve CA bundles, and give each publication a configurable deadline with retry metrics. --- ...artial_physical_snapshot_object_offload.md | 6 +- docs/snapshot_offload_operations.md | 5 +- internal/snapshotoffload/s3_store.go | 35 ++++++---- internal/snapshotoffload/s3_store_test.go | 4 +- internal/snapshotoffload/scheduler.go | 23 +++++-- internal/snapshotoffload/scheduler_test.go | 57 ++++++++++++++++ main_snapshot_offload.go | 63 ++++++++++++----- main_snapshot_offload_test.go | 68 +++++++++++++++++-- 8 files changed, 221 insertions(+), 40 deletions(-) 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 2bd15978a..5004535b2 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,6 +1,6 @@ # Physical Snapshot Object Offload -Status: Partial — M0/M1/M2 implemented; M3 runtime wiring and the operations runbook implemented; retention/GC, restore drills, multi-node acceptance, and the versioned-bucket decision pending +Status: Partial — M0/M1/M2 implemented; M3 runtime wiring and the operations runbook implemented; retention/GC integration (#1222 implemented, unmerged), restore drills, multi-node acceptance, and the versioned-bucket decision pending Author: bootjp Date: 2026-07-19 Updated: 2026-09-25 @@ -117,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 diff --git a/docs/snapshot_offload_operations.md b/docs/snapshot_offload_operations.md index 3355ce06d..085800073 100644 --- a/docs/snapshot_offload_operations.md +++ b/docs/snapshot_offload_operations.md @@ -64,6 +64,7 @@ elastickv \ | `--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 @@ -247,7 +248,9 @@ whole node. `--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. + `--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 diff --git a/internal/snapshotoffload/s3_store.go b/internal/snapshotoffload/s3_store.go index 3cf82be09..1928b6aee 100644 --- a/internal/snapshotoffload/s3_store.go +++ b/internal/snapshotoffload/s3_store.go @@ -6,7 +6,6 @@ import ( "encoding/base64" "encoding/hex" "io" - "net/http" "strings" "time" @@ -55,7 +54,7 @@ type S3StoreConfig struct { ServerSideEncryption string SSEKMSKeyID string DisableChecksumHeaders bool - HTTPClient *http.Client + HTTPClientDecorator func(aws.HTTPClient) (aws.HTTPClient, error) } type S3Store struct { @@ -77,19 +76,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 cfg.HTTPClient != nil { - o.HTTPClient = cfg.HTTPClient - } - if stringsTrim(cfg.Endpoint) != "" { - o.BaseEndpoint = aws.String(stringsTrim(cfg.Endpoint)) - } - }) } return &S3Store{ client: client, @@ -102,6 +93,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 +} + func loadS3AWSConfig(ctx context.Context, cfg S3StoreConfig) (aws.Config, error) { optFns := make([]func(*config.LoadOptions) error, 0, s3LoadConfigOptionCapHint) if stringsTrim(cfg.Region) != "" { diff --git a/internal/snapshotoffload/s3_store_test.go b/internal/snapshotoffload/s3_store_test.go index 0571bd693..08ab336f9 100644 --- a/internal/snapshotoffload/s3_store_test.go +++ b/internal/snapshotoffload/s3_store_test.go @@ -62,7 +62,9 @@ func TestS3StoreUsesConfiguredHTTPClientOnlyForS3(t *testing.T) { ServerSideEncryption: "AES256", AccessKeyID: "test-access-key", SecretAccessKey: "test-secret-key", - HTTPClient: client, + HTTPClientDecorator: func(aws.HTTPClient) (aws.HTTPClient, error) { + return client, nil + }, } awsCfg, err := loadS3AWSConfig(context.Background(), cfg) require.NoError(t, err) diff --git a/internal/snapshotoffload/scheduler.go b/internal/snapshotoffload/scheduler.go index 2db05eee1..f584a4221 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 ObjectStore, groups []OffloadGroup, prefix, sourceCluste 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 b788a65ef..2103850a2 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 { + ObjectStore + 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.ObjectStore.HeadObject(ctx, key) +} + +func TestSchedulerPublicationTimeoutReleasesScanAndAllowsRetry(t *testing.T) { + t.Parallel() + root := t.TempDir() + dataDir := seedSchedulerGroup(t, root, "group") + store := &stalledHeadStore{ObjectStore: 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_snapshot_offload.go b/main_snapshot_offload.go index ca2b6343b..2157bb3ee 100644 --- a/main_snapshot_offload.go +++ b/main_snapshot_offload.go @@ -11,6 +11,7 @@ import ( "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" @@ -56,6 +57,8 @@ var ( "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", "", @@ -79,7 +82,7 @@ var snapshotOffloadAllowInsecureEndpoint = flag.Bool("snapshotOffloadAllowInsecu // it fails startup instead. func rejectPlaintextOffloadEndpoint() error { endpoint := strings.TrimSpace(*snapshotOffloadEndpoint) - if endpoint == "" || *snapshotOffloadAllowInsecureEndpoint { + if endpoint == "" { return nil } parsed, err := url.Parse(endpoint) @@ -88,7 +91,9 @@ func rejectPlaintextOffloadEndpoint() error { } // A scheme-less host:port is ambiguous rather than known-plaintext, but // the AWS SDK resolves it as http, so it is refused too. - if !strings.EqualFold(parsed.Scheme, "https") || parsed.Host == "" || parsed.Opaque != "" { + 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 "+ @@ -98,24 +103,39 @@ func rejectPlaintextOffloadEndpoint() error { return nil } -func snapshotOffloadHTTPClient() *http.Client { - if *snapshotOffloadAllowInsecureEndpoint { - return nil - } - return &http.Client{ - Transport: snapshotOffloadHTTPSTransport{ - next: awshttp.NewBuildableClient().GetTransport(), - }, - CheckRedirect: checkSnapshotOffloadRedirect, - } +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 + next http.RoundTripper + allowInsecure bool } func (t snapshotOffloadHTTPSTransport) RoundTrip(req *http.Request) (*http.Response, error) { - if req == nil || req.URL == nil || !strings.EqualFold(req.URL.Scheme, "https") { + if !snapshotOffloadRequestAllowed(req, t.allowInsecure) { return nil, errors.Wrap(snapshotoffload.ErrInvalidOptions, "snapshot offload request must use https") } @@ -127,8 +147,16 @@ func (t snapshotOffloadHTTPSTransport) RoundTrip(req *http.Request) (*http.Respo return response, errors.Wrap(err, "snapshot offload request") } -func checkSnapshotOffloadRedirect(req *http.Request, via []*http.Request) error { - if req == nil || req.URL == nil || !strings.EqualFold(req.URL.Scheme, "https") { +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") } @@ -220,7 +248,7 @@ func buildSnapshotOffloadStore(ctx context.Context) (snapshotoffload.ObjectStore ForcePathStyle: *snapshotOffloadForcePathStyle, ServerSideEncryption: strings.TrimSpace(*snapshotOffloadSSE), SSEKMSKeyID: strings.TrimSpace(*snapshotOffloadSSEKMSKeyID), - HTTPClient: snapshotOffloadHTTPClient(), + HTTPClientDecorator: snapshotOffloadHTTPClient, }) if err != nil { return nil, errors.Wrap(err, "snapshot offload: s3 store") @@ -339,6 +367,7 @@ func startSnapshotOffload( opts := []snapshotoffload.SchedulerOption{ snapshotoffload.WithSchedulerInterval(*snapshotOffloadInterval), snapshotoffload.WithSchedulerConcurrency(*snapshotOffloadConcurrency), + snapshotoffload.WithSchedulerPublishTimeout(*snapshotOffloadPublishTimeout), snapshotoffload.WithSchedulerObserver(observer), snapshotoffload.WithSchedulerLogger(logger), } diff --git a/main_snapshot_offload_test.go b/main_snapshot_offload_test.go index 6c53e59f3..926cd883f 100644 --- a/main_snapshot_offload_test.go +++ b/main_snapshot_offload_test.go @@ -2,9 +2,11 @@ package main import ( "context" + "encoding/pem" "io" "log/slog" "net/http" + "net/http/httptest" "net/url" "os" "path/filepath" @@ -12,6 +14,7 @@ import ( "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" @@ -273,6 +276,7 @@ func TestRejectPlaintextOffloadEndpoint(t *testing.T) { {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 @@ -317,21 +321,77 @@ func TestSnapshotOffloadRedirectPolicy(t *testing.T) { } httpsRedirect := redirect("https://s3.example.com/next", http.StatusTemporaryRedirect) - require.NoError(t, checkSnapshotOffloadRedirect(httpsRedirect, []*http.Request{origin})) + 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}), + 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}), + 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}), + 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 } From 1a925c900fe60c0deecd2f6216fef9d72992f656 Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 00:30:26 +0900 Subject: [PATCH 4/5] test: verify three-node offload and isolated recovery --- ...artial_physical_snapshot_object_offload.md | 19 +- docs/snapshot_offload_operations.md | 45 +++ main_snapshot_offload_acceptance_test.go | 262 ++++++++++++++++++ 3 files changed, 322 insertions(+), 4 deletions(-) create mode 100644 main_snapshot_offload_acceptance_test.go 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 4cdd6205b..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,6 +1,6 @@ # Physical Snapshot Object Offload -Status: Partial — M0/M1/M2 implemented; M3 publication and retention runtime, corruption drills and operations runbook implemented; multi-node acceptance 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-10-22 @@ -51,8 +51,9 @@ The M1 object-store-neutral substrate now adds: 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. Multi-node acceptance -remains pending; the operator runbook is at +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 @@ -253,7 +254,7 @@ 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. 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. Multi-node acceptance remains pending. | +| 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 @@ -280,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 index af6ba8845..1db8b5a7c 100644 --- a/docs/snapshot_offload_operations.md +++ b/docs/snapshot_offload_operations.md @@ -331,3 +331,48 @@ logs a warning and still requires `--expect-source-cluster`. 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/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)) +} From e8166751524f321fcdf2ed1d6e24f0e263dbfdb7 Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 04:57:24 +0900 Subject: [PATCH 5/5] monitoring: reuse snapshot offload label names --- monitoring/snapshot_offload.go | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/monitoring/snapshot_offload.go b/monitoring/snapshot_offload.go index bfbfa8474..a73cbb486 100644 --- a/monitoring/snapshot_offload.go +++ b/monitoring/snapshot_offload.go @@ -14,6 +14,8 @@ 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 @@ -39,28 +41,28 @@ func newSnapshotOffloadMetrics(registerer prometheus.Registerer) *SnapshotOffloa Name: "elastickv_snapshot_offload_published_total", Help: "Total physical snapshots published to the object store, by Raft group.", }, - []string{"group_id"}, + []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{"group_id", "reason"}, + []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{"group_id"}, + []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{"group_id"}, + []string{snapshotOffloadGroupIDLabel}, ), publishSeconds: prometheus.NewHistogramVec( prometheus.HistogramOpts{ @@ -68,7 +70,7 @@ func newSnapshotOffloadMetrics(registerer prometheus.Registerer) *SnapshotOffloa Help: "Wall time to spool, upload and commit one snapshot.", Buckets: []float64{0.5, 1, 5, 15, 30, 60, 300, 900, 1800, 3600}, }, - []string{"group_id"}, + []string{snapshotOffloadGroupIDLabel}, ), payloadBytes: prometheus.NewHistogramVec( prometheus.HistogramOpts{ @@ -80,7 +82,7 @@ func newSnapshotOffloadMetrics(registerer prometheus.Registerer) *SnapshotOffloa snapshotPayloadBucketCount, ), }, - []string{"group_id"}, + []string{snapshotOffloadGroupIDLabel}, ), } registerer.MustRegister(