diff --git a/cmd/elastickv-snapshot-offload/main.go b/cmd/elastickv-snapshot-offload/main.go index efdd1640d..0bf1bba33 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,11 @@ 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 } func main() { @@ -151,6 +157,10 @@ 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.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 +176,14 @@ 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") + } + 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 +277,11 @@ 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, }) 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..e2fc4839c 100644 --- a/cmd/elastickv-snapshot-offload/main_test.go +++ b/cmd/elastickv-snapshot-offload/main_test.go @@ -50,6 +50,9 @@ 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", }, io.Discard, logger) require.NoError(t, err) require.Equal(t, exitSuccess, code) @@ -168,6 +171,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 +188,112 @@ 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", + }, 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", + }) + 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", + }) + require.Error(t, err) +} 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..f28daba98 --- /dev/null +++ b/docs/snapshot_offload_operations.md @@ -0,0 +1,262 @@ +# 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 \ + --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 says "group" must say the SAME group + +`--manifest-key`, `--data-dir`, `--expect-group` and `--peers` are four +independent statements about which group you are restoring. Three of them +used to be uncheckable against each other: + +- **`--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. + +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..a3e6a09df 100644 --- a/internal/snapshotoffload/manifest.go +++ b/internal/snapshotoffload/manifest.go @@ -19,9 +19,15 @@ 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") + 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..8a54bcf7a 100644 --- a/internal/snapshotoffload/offload_test.go +++ b/internal/snapshotoffload/offload_test.go @@ -58,6 +58,7 @@ func TestPublishAndRestorePhysicalSnapshotRoundTrip(t *testing.T) { Peers: []etcdraftengine.Peer{ {NodeID: 9, ID: "n9", Address: "127.0.0.1:19009"}, }, + ExpectGroupID: expectGroup(manifest.GroupID), }) require.NoError(t, err) require.Equal(t, int64(len(payload)), result.PayloadBytes) @@ -95,10 +96,11 @@ 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), }) require.ErrorIs(t, err, ErrIntegrity) _, statErr := os.Stat(restoreDataDir) @@ -376,10 +378,11 @@ 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), }) require.ErrorIs(t, err, ErrIntegrity) require.Zero(t, tracked.getObjectCalls) @@ -480,10 +483,11 @@ 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), }) require.ErrorIs(t, err, etcdraftengine.ErrExternalSnapshotRestoreExists) } @@ -511,6 +515,7 @@ func TestRestoreRejectsInvalidPeersBeforePayloadDownload(t *testing.T) { Peers: []etcdraftengine.Peer{ {NodeID: 0, ID: "n0", Address: "127.0.0.1:12000"}, }, + ExpectGroupID: expectGroup(manifest.GroupID), }) require.ErrorIs(t, err, ErrInvalidOptions) require.Zero(t, tracked.getObjectCalls) @@ -534,10 +539,11 @@ 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), }) require.ErrorIs(t, err, context.Canceled) require.Zero(t, tracked.getObjectCalls) @@ -724,3 +730,10 @@ 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 +} diff --git a/internal/snapshotoffload/restore.go b/internal/snapshotoffload/restore.go index 0a45e819b..5d7008a57 100644 --- a/internal/snapshotoffload/restore.go +++ b/internal/snapshotoffload/restore.go @@ -20,6 +20,23 @@ 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 } const ( @@ -89,6 +106,11 @@ 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 := checkRestorePreflight(ctx, opts.DataDir); err != nil { return Manifest{}, "", nil, err } @@ -210,11 +232,27 @@ 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") 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 +} + 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..f795abf7c 100644 --- a/internal/snapshotoffload/s3_store_test.go +++ b/internal/snapshotoffload/s3_store_test.go @@ -46,6 +46,7 @@ func TestPublishAndRestorePhysicalSnapshotRoundTripWithS3Store(t *testing.T) { Peers: []etcdraftengine.Peer{ {NodeID: 2, ID: "n2", Address: "127.0.0.1:12002"}, }, + ExpectGroupID: expectGroup(manifest.GroupID), }) 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..be9ef13e3 --- /dev/null +++ b/main_snapshot_offload.go @@ -0,0 +1,210 @@ +package main + +import ( + "context" + "flag" + "log/slog" + "strings" + + "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") +) + +// 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 + } + 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. +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") + } + + 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..ad2e28bda --- /dev/null +++ b/main_snapshot_offload_test.go @@ -0,0 +1,240 @@ +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") + }) + } +} 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()) +}