diff --git a/cmd/elastickv-snapshot-offload/main.go b/cmd/elastickv-snapshot-offload/main.go index efdd1640d..b874138cb 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,12 @@ 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 } func main() { @@ -151,6 +158,14 @@ 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.") if err := fs.Parse(argv); err != nil { return nil, errors.WithStack(err) } @@ -166,6 +181,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 } @@ -259,10 +285,12 @@ func runRestore(ctx context.Context, cfg *restoreConfig, logger *slog.Logger) er return err } 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, }) 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..6238c5039 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) @@ -168,6 +172,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 +189,187 @@ 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") +} 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..7581ea7ce 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 @@ -48,7 +48,12 @@ 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 per §5. Restore drills and corruption tests are in place; +multi-node acceptance and the §7 versioned-bucket decision remain +pending; the operator runbook is at +[`../snapshot_offload_operations.md`](../snapshot_offload_operations.md). ## 2. Safety boundary diff --git a/docs/snapshot_offload_operations.md b/docs/snapshot_offload_operations.md new file mode 100644 index 000000000..99651b11d --- /dev/null +++ b/docs/snapshot_offload_operations.md @@ -0,0 +1,287 @@ +# 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, which lands +> in a separate change. 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 + +- **`--snapshotOffloadEndpoint` must be `https://`.** The node refuses a + plaintext endpoint at startup: this path carries whole snapshot payloads + and the credentials used to write them, and a silent downgrade is + invisible until someone captures the traffic. 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 before scheduling, so repeated crashes during large snapshots + no longer accumulate full payload copies. 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. + +## 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..f1ff9887f 100644 --- a/internal/snapshotoffload/offload_test.go +++ b/internal/snapshotoffload/offload_test.go @@ -58,6 +58,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 +97,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) @@ -376,10 +380,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 +486,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 +519,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 +544,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 +736,125 @@ 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", + }) + 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") +} diff --git a/internal/snapshotoffload/publish.go b/internal/snapshotoffload/publish.go index d53914c0f..340003fa6 100644 --- a/internal/snapshotoffload/publish.go +++ b/internal/snapshotoffload/publish.go @@ -315,6 +315,85 @@ func validatePublishOptions(opts PublishOptions) error { } } +// 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 can pass any positive age, since no publish of this +// process has begun yet; the guard matters for a second node sharing the +// directory. +// +// 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") + } + matches, err := filepath.Glob(filepath.Join(filepath.Clean(spoolDir), spoolFilePattern)) + if err != nil { + return nil, errors.Wrapf(err, "glob spool dir %s", spoolDir) + } + var ( + removed []string + failures []error + ) + for _, match := range matches { + gone, err := removeStaleSpoolFile(match, olderThan) + if err != nil { + failures = append(failures, err) + continue + } + if gone { + removed = append(removed, match) + } + } + if len(failures) > 0 { + return removed, errors.Wrapf(errors.Join(failures...), + "clean spool dir %s: %d of %d files failed", spoolDir, len(failures), len(matches)) + } + 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 +405,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..d10d4268d 100644 --- a/internal/snapshotoffload/restore.go +++ b/internal/snapshotoffload/restore.go @@ -20,6 +20,34 @@ 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 } const ( @@ -89,6 +117,14 @@ 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); err != nil { + return Manifest{}, "", nil, err + } if err := checkRestorePreflight(ctx, opts.DataDir); err != nil { return Manifest{}, "", nil, err } @@ -210,11 +246,44 @@ 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) error { + expect = stringsTrim(expect) + if expect == "" { + return errors.Wrap(ErrInvalidOptions, "expected source cluster is required") + } + if stringsTrim(manifest.SourceCluster) != 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_test.go b/internal/snapshotoffload/s3_store_test.go index 010f65981..325da797c 100644 --- a/internal/snapshotoffload/s3_store_test.go +++ b/internal/snapshotoffload/s3_store_test.go @@ -46,6 +46,8 @@ 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) diff --git a/main.go b/main.go index bbb963d7e..7a7af8340 100644 --- a/main.go +++ b/main.go @@ -806,6 +806,16 @@ func startDistributionStartup(in distributionStartupInput) (distributionStartup, } startMonitoringCollectors(in.ctx, in.metricsRegistry, in.runtimes, in.clock) 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 { + 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..fdba2ecd9 --- /dev/null +++ b/main_snapshot_offload.go @@ -0,0 +1,291 @@ +package main + +import ( + "context" + "flag" + "log/slog" + "net/url" + "strings" + "time" + + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/bootjp/elastickv/internal/snapshotoffload" + "github.com/cockroachdb/errors" + "golang.org/x/sync/errgroup" +) + +// 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") { + return errors.Wrapf(snapshotoffload.ErrInvalidOptions, + "--snapshotOffloadEndpoint %q is not https; snapshot payloads and credentials "+ + "would cross the network unencrypted. Pass an https:// endpoint, or "+ + "--snapshotOffloadAllowInsecureEndpoint for local development only", + endpoint) + } + 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), + }) + if err != nil { + return nil, errors.Wrap(err, "snapshot offload: s3 store") + } + return store, nil +} + +// snapshotOffloadGroups builds one OffloadGroup per local Raft group. +// +// Both leadership callbacks read the engine through snapshotEngine(): +// the scheduler outlives startup and races Close(), so a direct field +// read would be a data race. A runtime whose engine has been cleared +// reports "not leader", which fails closed. +// cleanStaleOffloadSpool removes leftover spool files from every directory +// this process would publish through. +// +// Failures are logged, not fatal: an undeletable spool file is a disk-space +// problem, and refusing to serve because of one would be a worse outcome than +// the leak it guards against. +func cleanStaleOffloadSpool( + runtimes []*raftGroupRuntime, + raftDir, raftID string, + multi bool, + logger *slog.Logger, +) { + configured := strings.TrimSpace(*snapshotOffloadSpoolDir) + seen := make(map[string]struct{}) + now := time.Now() + 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, now) + 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))) + } + } +} + +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 + } + 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") + } + + // Before the scheduler starts: spool files left by a killed process are + // full payload copies, and nothing else removed them, so repeated crashes + // during large snapshots filled the spool volume. Safe to do here + // because no publish of THIS process has begun, so every existing file + // belongs to a previous one. + 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..1c7fc73a6 --- /dev/null +++ b/main_snapshot_offload_test.go @@ -0,0 +1,291 @@ +package main + +import ( + "context" + "io" + "log/slog" + "path/filepath" + "testing" + + "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: "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) + }) + } +} diff --git a/monitoring/registry.go b/monitoring/registry.go index 0a7631415..2ecc0ddc2 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 } @@ -64,6 +65,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) return r } @@ -295,6 +297,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()) +}