From 3b10815988098667c7266fb4547f65975e452d27 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 8 Sep 2026 16:22:48 +0900 Subject: [PATCH 1/3] snapshotoffload: wire the scheduler into the server runtime MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes M2: the scheduler now actually runs. Opt-in via --snapshotOffloadBucket (S3) or --snapshotOffloadLocalDir, with the §7 configuration surface for region, endpoint, profile, path style, server-side encryption, schedule, jitter, concurrency, spool dir and source-cluster identity. Each local Raft group contributes its own data dir plus both leadership callbacks. Both read the engine through snapshotEngine(): the scheduler outlives startup and races Close(), so a direct field read would be a data race, and a runtime whose engine has been cleared reports "not leader" rather than publishing. A configured-but-unbuildable offload fails startup instead of logging and continuing. An operator who set a backup destination and silently received no backups is worse off than one whose node refused to start. Bucket and local dir are mutually exclusive for the same reason: ambiguity about which destination holds the artifacts is only discovered when a restore is attempted. Adds the scheduler's Prometheus metrics — published/skipped/failed counters, last-published-index gauge, publish-duration and payload-size histograms. The failure counter deliberately carries no error label: messages are unbounded and one recurring failure would explode the metric's cardinality. Skip reasons are normalized into the scheduler's closed set. Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE --- ...artial_physical_snapshot_object_offload.md | 6 +- main.go | 10 + main_snapshot_offload.go | 210 ++++++++++++++++++ main_snapshot_offload_test.go | 125 +++++++++++ monitoring/registry.go | 12 + monitoring/snapshot_offload.go | 157 +++++++++++++ monitoring/snapshot_offload_test.go | 96 ++++++++ 7 files changed, 615 insertions(+), 1 deletion(-) create mode 100644 main_snapshot_offload.go create mode 100644 main_snapshot_offload_test.go create mode 100644 monitoring/snapshot_offload.go create mode 100644 monitoring/snapshot_offload_test.go diff --git a/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..f338b0cca 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,11 @@ 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, operator documentation, and the §7 +versioned-bucket decision remain pending. ## 2. Safety boundary diff --git a/main.go b/main.go index 87aae6b0c..f284e6082 100644 --- a/main.go +++ b/main.go @@ -805,6 +805,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..aae595609 --- /dev/null +++ b/main_snapshot_offload_test.go @@ -0,0 +1,125 @@ +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)) +} diff --git a/monitoring/registry.go b/monitoring/registry.go index 568bfeeb6..c25257c52 100644 --- a/monitoring/registry.go +++ b/monitoring/registry.go @@ -31,6 +31,7 @@ type Registry struct { coldStartObs *ColdStartObserver tso *TSOMetrics tsoObserver *TSOObserver + snapOffload *SnapshotOffloadMetrics } // NewRegistry builds a registry with constant labels that identify the local node. @@ -63,6 +64,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) return r } @@ -292,3 +294,13 @@ 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 +} diff --git a/monitoring/snapshot_offload.go b/monitoring/snapshot_offload.go new file mode 100644 index 000000000..6670a2bc4 --- /dev/null +++ b/monitoring/snapshot_offload.go @@ -0,0 +1,157 @@ +package monitoring + +import ( + "strconv" + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +// snapshotPayloadBucketBase is the smallest payload-size histogram +// bucket (1 MiB). Snapshots below it are rounding error next to the +// multi-gigabyte cases the histogram exists to show. +const ( + snapshotPayloadBucketBase = 1 << 20 // 1 MiB + snapshotPayloadBucketFactor = 4 + snapshotPayloadBucketCount = 8 // 1 MiB through ~16 GiB +) + +// SnapshotOffloadMetrics exposes the physical snapshot offload +// scheduler's outcomes (design doc §4). +// +// group_id is a label on every series: its cardinality is the number +// of Raft groups this process hosts, which is bounded by deployment +// topology rather than by traffic. skip reason is a closed set owned +// by the scheduler. +type SnapshotOffloadMetrics struct { + published *prometheus.CounterVec + skipped *prometheus.CounterVec + failed *prometheus.CounterVec + lastPublishIndex *prometheus.GaugeVec + publishSeconds *prometheus.HistogramVec + payloadBytes *prometheus.HistogramVec +} + +func newSnapshotOffloadMetrics(registerer prometheus.Registerer) *SnapshotOffloadMetrics { + m := &SnapshotOffloadMetrics{ + published: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "elastickv_snapshot_offload_published_total", + Help: "Total physical snapshots published to the object store, by Raft group.", + }, + []string{"group_id"}, + ), + skipped: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "elastickv_snapshot_offload_skipped_total", + Help: "Total offload scans that published nothing, by Raft group and reason. Routine on a follower or an unchanged snapshot.", + }, + []string{"group_id", "reason"}, + ), + failed: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "elastickv_snapshot_offload_failed_total", + Help: "Total offload attempts that failed, by Raft group. A sustained rate means backups are not being taken.", + }, + []string{"group_id"}, + ), + lastPublishIndex: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "elastickv_snapshot_offload_last_published_index", + Help: "Raft index of the most recent snapshot this process published, by group. Staleness here is the backup-freshness signal.", + }, + []string{"group_id"}, + ), + publishSeconds: prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "elastickv_snapshot_offload_publish_seconds", + Help: "Wall time to spool, upload and commit one snapshot.", + Buckets: []float64{0.5, 1, 5, 15, 30, 60, 300, 900, 1800, 3600}, + }, + []string{"group_id"}, + ), + payloadBytes: prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "elastickv_snapshot_offload_payload_bytes", + Help: "Size of each published snapshot payload.", + Buckets: prometheus.ExponentialBuckets( + snapshotPayloadBucketBase, + snapshotPayloadBucketFactor, + snapshotPayloadBucketCount, + ), + }, + []string{"group_id"}, + ), + } + registerer.MustRegister( + m.published, + m.skipped, + m.failed, + m.lastPublishIndex, + m.publishSeconds, + m.payloadBytes, + ) + return m +} + +// ObserveSnapshotOffloadPublished records one successful publication. +func (m *SnapshotOffloadMetrics) ObserveSnapshotOffloadPublished( + groupID, index uint64, payloadBytes int64, elapsed time.Duration, +) { + if m == nil { + return + } + label := snapshotOffloadGroupLabel(groupID) + m.published.WithLabelValues(label).Inc() + m.lastPublishIndex.WithLabelValues(label).Set(float64(index)) + m.publishSeconds.WithLabelValues(label).Observe(max(0, elapsed).Seconds()) + m.payloadBytes.WithLabelValues(label).Observe(float64(max(int64(0), payloadBytes))) +} + +// ObserveSnapshotOffloadSkipped records a scan that published nothing. +func (m *SnapshotOffloadMetrics) ObserveSnapshotOffloadSkipped(groupID uint64, reason string) { + if m == nil { + return + } + m.skipped.WithLabelValues(snapshotOffloadGroupLabel(groupID), normalizeSnapshotOffloadSkip(reason)).Inc() +} + +// ObserveSnapshotOffloadFailed records a failed attempt. The error is +// deliberately not a label: its text is unbounded, and a per-message +// series would let one recurring failure explode the metric's +// cardinality. Diagnosis comes from the scheduler's log line. +func (m *SnapshotOffloadMetrics) ObserveSnapshotOffloadFailed(groupID uint64, _ error) { + if m == nil { + return + } + m.failed.WithLabelValues(snapshotOffloadGroupLabel(groupID)).Inc() +} + +func snapshotOffloadGroupLabel(groupID uint64) string { + return strconv.FormatUint(groupID, 10) +} + +// Skip reasons emitted by the scheduler. +const ( + snapshotOffloadSkipNotLeader = "not_leader" + snapshotOffloadSkipAlreadyPublished = "already_published" + snapshotOffloadSkipNoSnapshot = "no_persisted_snapshot" + snapshotOffloadSkipInFlight = "already_in_flight" + snapshotOffloadSkipUnknownLeader = "leadership_unknown" + snapshotOffloadSkipUnknown = "unknown" +) + +// normalizeSnapshotOffloadSkip keeps the reason label inside the +// scheduler's closed set. +func normalizeSnapshotOffloadSkip(reason string) string { + switch reason { + case snapshotOffloadSkipNotLeader, + snapshotOffloadSkipAlreadyPublished, + snapshotOffloadSkipNoSnapshot, + snapshotOffloadSkipInFlight, + snapshotOffloadSkipUnknownLeader: + return reason + default: + return snapshotOffloadSkipUnknown + } +} diff --git a/monitoring/snapshot_offload_test.go b/monitoring/snapshot_offload_test.go new file mode 100644 index 000000000..b2e68a822 --- /dev/null +++ b/monitoring/snapshot_offload_test.go @@ -0,0 +1,96 @@ +package monitoring + +import ( + "errors" + "strings" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" +) + +func TestSnapshotOffloadMetricsRecordOutcomes(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + m := newSnapshotOffloadMetrics(reg) + + m.ObserveSnapshotOffloadPublished(7, 4211, 5<<20, 12*time.Second) + m.ObserveSnapshotOffloadSkipped(7, snapshotOffloadSkipNotLeader) + m.ObserveSnapshotOffloadSkipped(7, snapshotOffloadSkipAlreadyPublished) + m.ObserveSnapshotOffloadFailed(9, errors.New("object store unavailable")) + + require.NoError(t, testutil.GatherAndCompare( + reg, + strings.NewReader(` +# HELP elastickv_snapshot_offload_published_total Total physical snapshots published to the object store, by Raft group. +# TYPE elastickv_snapshot_offload_published_total counter +elastickv_snapshot_offload_published_total{group_id="7"} 1 +# HELP elastickv_snapshot_offload_last_published_index Raft index of the most recent snapshot this process published, by group. Staleness here is the backup-freshness signal. +# TYPE elastickv_snapshot_offload_last_published_index gauge +elastickv_snapshot_offload_last_published_index{group_id="7"} 4211 +# HELP elastickv_snapshot_offload_failed_total Total offload attempts that failed, by Raft group. A sustained rate means backups are not being taken. +# TYPE elastickv_snapshot_offload_failed_total counter +elastickv_snapshot_offload_failed_total{group_id="9"} 1 +# HELP elastickv_snapshot_offload_skipped_total Total offload scans that published nothing, by Raft group and reason. Routine on a follower or an unchanged snapshot. +# TYPE elastickv_snapshot_offload_skipped_total counter +elastickv_snapshot_offload_skipped_total{group_id="7",reason="already_published"} 1 +elastickv_snapshot_offload_skipped_total{group_id="7",reason="not_leader"} 1 +`), + "elastickv_snapshot_offload_published_total", + "elastickv_snapshot_offload_last_published_index", + "elastickv_snapshot_offload_failed_total", + "elastickv_snapshot_offload_skipped_total", + )) +} + +// TestSnapshotOffloadMetricsBoundTheSkipReasonLabel is the cardinality +// guard: an unrecognised reason must collapse rather than mint a +// series per distinct string. +func TestSnapshotOffloadMetricsBoundTheSkipReasonLabel(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + m := newSnapshotOffloadMetrics(reg) + + m.ObserveSnapshotOffloadSkipped(1, "something-new") + m.ObserveSnapshotOffloadSkipped(1, "something-else") + m.ObserveSnapshotOffloadSkipped(1, "") + + require.Equal(t, 1, testutil.CollectAndCount(m.skipped)) + require.InDelta(t, 3.0, + testutil.ToFloat64(m.skipped.WithLabelValues("1", snapshotOffloadSkipUnknown)), 0.0001) +} + +// TestSnapshotOffloadMetricsDoNotLabelByError pins that the failure +// counter carries no error text: messages are unbounded, and one +// recurring failure would otherwise explode the metric's cardinality. +func TestSnapshotOffloadMetricsDoNotLabelByError(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + m := newSnapshotOffloadMetrics(reg) + + for i := range 20 { + m.ObserveSnapshotOffloadFailed(1, errors.New(strings.Repeat("x", i+1))) + } + require.Equal(t, 1, testutil.CollectAndCount(m.failed), + "distinct error texts must not create distinct series") +} + +func TestSnapshotOffloadMetricsNilReceiverIsInert(t *testing.T) { + t.Parallel() + + var m *SnapshotOffloadMetrics + require.NotPanics(t, func() { + m.ObserveSnapshotOffloadPublished(1, 2, 3, time.Second) + m.ObserveSnapshotOffloadSkipped(1, "x") + m.ObserveSnapshotOffloadFailed(1, errors.New("boom")) + }) + require.NotNil(t, NewRegistry("n1", "127.0.0.1:1").SnapshotOffloadObserver()) + + var nilRegistry *Registry + require.Nil(t, nilRegistry.SnapshotOffloadObserver()) +} From 79ec2ed8bef36fe00f743f556e67f1846776b22a Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 8 Sep 2026 16:30:10 +0900 Subject: [PATCH 2/3] docs: add the snapshot offload operations runbook Closes the M3 operator-documentation item: enabling offload, verifying that backups are actually being produced, retention semantics, restore, and failure modes. Two things the runbook makes explicit because they are the ways an operator gets silently burned: - a group whose last_published_index never advances has no backups even though nothing is failing, so staleness needs its own alert; - a versioned bucket without a noncurrent-version lifecycle rule grows without bound while retention reports success, because a keyed delete only writes a delete marker. Every flag, metric name and skip reason in the runbook was cross-checked against the source rather than written from memory. Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE --- ...artial_physical_snapshot_object_offload.md | 5 +- docs/snapshot_offload_operations.md | 203 ++++++++++++++++++ 2 files changed, 206 insertions(+), 2 deletions(-) create mode 100644 docs/snapshot_offload_operations.md 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 f338b0cca..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 @@ -51,8 +51,9 @@ The M1 object-store-neutral substrate now adds: The runtime scheduler is implemented and wired into main.go, opt-in via `--snapshotOffloadBucket` (or `--snapshotOffloadLocalDir`). Retention/GC is implemented per §5. Restore drills and corruption tests are in place; -multi-node acceptance, operator documentation, and the §7 -versioned-bucket decision remain pending. +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..592dbf86c --- /dev/null +++ b/docs/snapshot_offload_operations.md @@ -0,0 +1,203 @@ +# 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 into a fresh directory. +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 \ + --peers='n1=10.0.0.1:50051,n2=10.0.0.2:50051,n3=10.0.0.3:50051' + +# 3. Start the node normally against the restored directory. +``` + +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. From e349dee1bd2203c5a2056fe9ab51d4b9223992d2 Mon Sep 17 00:00:00 2001 From: bootjp Date: Wed, 9 Sep 2026 13:16:32 +0900 Subject: [PATCH 3/3] docs: correct the multi-group restore path in the offload runbook MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The restore example pointed --data-dir at the node's raft dir, but the server derives a per-group directory: a multi-group node opens //group-. An operator following the runbook during disaster recovery would restore into a path the server never opens, find the per-group directories empty at startup, and have the restore silently ignored — the worst possible moment for a documentation bug. Adds the full derivation table (multi-group, single group, single-node group 0) and states that a multi-group recovery needs one restore invocation per group. A table-driven test pins every documented path against groupDataDir so the runbook cannot drift from the function the server actually uses. Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE --- docs/snapshot_offload_operations.md | 24 +++++++++++++++++--- main_snapshot_offload_test.go | 34 +++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 3 deletions(-) diff --git a/docs/snapshot_offload_operations.md b/docs/snapshot_offload_operations.md index 592dbf86c..e6af6d476 100644 --- a/docs/snapshot_offload_operations.md +++ b/docs/snapshot_offload_operations.md @@ -162,16 +162,34 @@ mistakenly points a restore at a live node. # 1. Find the generation to restore. elastickv-snapshot-offload publish --help # same store flags as below -# 2. Restore into a fresh directory. +# 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 \ + --data-dir=/var/lib/elastickv/n1/group-1 \ --peers='n1=10.0.0.1:50051,n2=10.0.0.2:50051,n3=10.0.0.3:50051' -# 3. Start the node normally against the restored directory. +# 3. Repeat for every group the node hosts, then start it normally. ``` +### 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: + +| Deployment | Group | Directory | +|---|---|---| +| multi-group (`--raftRedisMap` etc.) | any group *G* | `//group-` | +| single group | the default group | `/` | +| single-node, group 0 | 0 | `//group-0` | + +Restoring a multi-group node into `/` puts the data +where the server never looks: startup finds the per-group directories +empty and the restore is silently ignored. **A multi-group recovery +must restore every group's manifest into its own `group-` +directory** — one `restore` invocation per group — or the node comes +back with only the groups you happened to place correctly. + 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. diff --git a/main_snapshot_offload_test.go b/main_snapshot_offload_test.go index aae595609..f879f172a 100644 --- a/main_snapshot_offload_test.go +++ b/main_snapshot_offload_test.go @@ -123,3 +123,37 @@ 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") + }) + } +}