diff --git a/docs/design/2026_04_29_partial_data_at_rest_encryption.md b/docs/design/2026_04_29_partial_data_at_rest_encryption.md index 99f306e4c..a842d6391 100644 --- a/docs/design/2026_04_29_partial_data_at_rest_encryption.md +++ b/docs/design/2026_04_29_partial_data_at_rest_encryption.md @@ -1,6 +1,6 @@ # Data-at-rest encryption for elastickv -Status: Partial — Stages 0–8, 9A–9B, and 9C-1 shipped (5E deferred); remaining Stage 9 work open +Status: Partial — Stages 0–8, 9A–9B, 9C-1, and 9C-2 shipped (5E deferred); remaining Stage 9 work open Author: bootjp Date: 2026-04-29 @@ -35,7 +35,8 @@ Date: 2026-04-29 | 9A | Compress-then-encrypt, authenticated compression flag, encrypted-store Pebble compression policy, storage benchmark (§6.4, §8.3) | shipped | `2026_07_18_implemented_9a_encryption_compression.md` | | 9B | AWS KMS, GCP KMS, Vault Transit, and test/CI env KEK providers; mutually-exclusive source loader and loaded-provider mutator gate (§5.1, §6.1, §6.5) | shipped | `2026_07_18_implemented_9b_kek_providers.md` | | 9C-1 | Storage-envelope observability: `decrypt_failures_total`, `writes_per_dek`, `value_overhead_bytes`, wired from the storage envelope path through `monitoring.Registry` (§9.2) | shipped | — | -| 9C+ | Rotation budget/rewrap/retire/rewrite, the remaining §9.2 metrics (`active_dek_id`, `last_proposed_index_per_raft_dek`, `kek_unwrap_seconds`, `sidecar_raft_index`), remaining benchmarks and encrypted Jepsen (§5.2, §5.4, §6.5, §8) | open | — | +| 9C-2 | Sidecar/KEK observability: `active_dek_id{purpose}`, `sidecar_raft_index`, `kek_unwrap_seconds` (§9.2) | shipped | — | +| 9C+ | Rotation budget/rewrap/retire/rewrite, `last_proposed_index_per_raft_dek` (needs the §5.4 raft-DEK Wrap path), remaining benchmarks and encrypted Jepsen (§5.2, §5.4, §6.5, §8) | open | — | Stages 0–4 ship the entire byte-tag pipeline (storage envelope, raft envelope, FSM dispatch, halt-on-error) but leave it **production @@ -2527,10 +2528,12 @@ relevant flag and runbook section. New metrics (Stage 9C-1 shipped the three emitted by the storage envelope path — `decrypt_failures_total`, `writes_per_dek`, and -`value_overhead_bytes`. The remaining four are owned by the rotation / -sidecar / KEK subsystems and land with their milestones; they are -deliberately NOT registered yet, so an operator cannot mistake an -always-zero series for a healthy signal): +`value_overhead_bytes`. Stage 9C-2 shipped the three with live +sidecar/KEK sources — `active_dek_id`, `sidecar_raft_index`, and +`kek_unwrap_seconds`. Only `last_proposed_index_per_raft_dek` remains: +it needs the §5.4 raft-DEK Wrap path, so it is deliberately NOT +registered yet — an always-zero series is worse than an absent one, +because an operator could mistake it for a healthy signal): - `elastickv_encryption_active_dek_id{purpose}` — gauge, label is storage/raft. diff --git a/internal/encryption/applier.go b/internal/encryption/applier.go index 8e0137edc..474710505 100644 --- a/internal/encryption/applier.go +++ b/internal/encryption/applier.go @@ -228,6 +228,15 @@ type StateCache struct { // activeStorageDEKID and re-registers, which Registered()'s // equality check handles without a reset. registeredStorageDEKID atomic.Uint32 + // activeRaftDEKID mirrors sidecar.Active.Raft, and + // sidecarRaftAppliedIndex mirrors sidecar.RaftAppliedIndex. + // Neither participates in a decision: they exist so the §9.2 + // observability collector can read current sidecar state without + // doing file I/O on a metrics tick. They are refreshed by the + // same RefreshFromSidecar call that maintains the decision + // mirrors above, so they cannot drift from them. + activeRaftDEKID atomic.Uint32 + sidecarRaftAppliedIndex atomic.Uint64 } // NewStateCache returns a zero-initialised StateCache. The @@ -250,6 +259,31 @@ func (c *StateCache) RefreshFromSidecar(sc *Sidecar) { } c.activeStorageDEKID.Store(sc.Active.Storage) c.storageEnvelopeActive.Store(sc.StorageEnvelopeActive) + c.activeRaftDEKID.Store(sc.Active.Raft) + c.sidecarRaftAppliedIndex.Store(sc.RaftAppliedIndex) +} + +// ActiveRaftKeyID returns the current sidecar.Active.Raft DEK id. +// Observability only — the raft envelope path resolves its own key id +// through the raft envelope runtime, not through this mirror. +func (c *StateCache) ActiveRaftKeyID() (uint32, bool) { + if c == nil { + return 0, false + } + id := c.activeRaftDEKID.Load() + return id, id != 0 +} + +// SidecarRaftAppliedIndex returns the sidecar's last persisted +// raft_applied_index. §5.5 uses a persistent gap between this and the +// FSM applied index as the sidecar-divergence signal, and §9.2 +// exposes it as elastickv_encryption_sidecar_raft_index so an +// operator can alert on that gap. +func (c *StateCache) SidecarRaftAppliedIndex() uint64 { + if c == nil { + return 0 + } + return c.sidecarRaftAppliedIndex.Load() } // ActiveStorageKeyID returns the current sidecar.Active.Storage DEK @@ -862,8 +896,26 @@ func (a *Applier) writeBootstrapSidecar(raftIdx uint64, p fsmwire.BootstrapPaylo Created: createdAt, LocalEpoch: 0, } + if err := a.persistSidecar(sc, "bootstrap"); err != nil { + return err + } + return nil +} + +// persistSidecar writes the sidecar and, on success, refreshes the +// StateCache mirrors from the same value. +// +// Every apply path that persists the sidecar must go through this. +// Pairing the two here rather than at each call site is deliberate: +// the no-op branches (stale DEKID, already-active cutover) advance and +// persist RaftAppliedIndex but have no other work to do, and each one +// that forgot to refresh left the cache — and therefore the §9.2 +// elastickv_encryption_sidecar_raft_index gauge — behind the durable +// sidecar until the next fresh mutation or restart, which reads as a +// false sidecar-divergence signal. +func (a *Applier) persistSidecar(sc *Sidecar, context string) error { if err := WriteSidecar(a.sidecarPath, sc); err != nil { - return errors.Wrap(err, "applier: write sidecar for bootstrap") + return errors.Wrapf(err, "applier: write sidecar for %s", context) } a.stateCache.RefreshFromSidecar(sc) return nil @@ -1089,10 +1141,7 @@ func (a *Applier) applyEnableStorageEnvelope(raftIdx uint64, p fsmwire.RotationP // flipping the cutover fields. if p.DEKID != sc.Active.Storage { advanceRaftAppliedIndex(sc, raftIdx) - if err := WriteSidecar(a.sidecarPath, sc); err != nil { - return errors.Wrap(err, "applier: write sidecar for stale-dekid cutover no-op") - } - return nil + return a.persistSidecar(sc, "stale-dekid cutover no-op") } // §2.1 constraint #4 — idempotency. Preserve the original // StorageEnvelopeCutoverIndex; only advance the generic @@ -1109,10 +1158,7 @@ func (a *Applier) applyEnableStorageEnvelope(raftIdx uint64, p fsmwire.RotationP // registration-before-sidecar reordering landed in 74a504c8. if sc.StorageEnvelopeActive { advanceRaftAppliedIndex(sc, raftIdx) - if err := WriteSidecar(a.sidecarPath, sc); err != nil { - return errors.Wrap(err, "applier: write sidecar for already-active cutover no-op") - } - return nil + return a.persistSidecar(sc, "already-active cutover no-op") } // Fresh successful apply. // @@ -1148,10 +1194,9 @@ func (a *Applier) applyEnableStorageEnvelope(raftIdx uint64, p fsmwire.RotationP sc.StorageEnvelopeActive = true sc.StorageEnvelopeCutoverIndex = raftIdx advanceRaftAppliedIndex(sc, raftIdx) - if err := WriteSidecar(a.sidecarPath, sc); err != nil { - return errors.Wrap(err, "applier: write sidecar for cutover") + if err := a.persistSidecar(sc, "cutover"); err != nil { + return err } - a.stateCache.RefreshFromSidecar(sc) return nil } @@ -1277,8 +1322,8 @@ func (a *Applier) applyEnableRaftEnvelope(raftIdx uint64, p fsmwire.RotationPayl // separate bool flag). if sc.RaftEnvelopeCutoverIndex != 0 { advanceRaftAppliedIndex(sc, raftIdx) - if err := WriteSidecar(a.sidecarPath, sc); err != nil { - return errors.Wrap(err, "applier: write sidecar for already-active raft-cutover no-op") + if err := a.persistSidecar(sc, "already-active raft-cutover no-op"); err != nil { + return err } // Installer takes the CURRENT sc.Active.Raft, NOT the // replayed p.DEKID — the wrap closure must key to the @@ -1295,10 +1340,7 @@ func (a *Applier) applyEnableRaftEnvelope(raftIdx uint64, p fsmwire.RotationPayl // flipping the cutover field. if p.DEKID != sc.Active.Raft { advanceRaftAppliedIndex(sc, raftIdx) - if err := WriteSidecar(a.sidecarPath, sc); err != nil { - return errors.Wrap(err, "applier: write sidecar for stale-dekid raft-cutover no-op") - } - return nil + return a.persistSidecar(sc, "stale-dekid raft-cutover no-op") } // Fresh successful apply. Crash-recovery ordering follows // the storage variant: ApplyRegistration runs BEFORE @@ -1314,10 +1356,9 @@ func (a *Applier) applyEnableRaftEnvelope(raftIdx uint64, p fsmwire.RotationPayl } sc.RaftEnvelopeCutoverIndex = raftIdx advanceRaftAppliedIndex(sc, raftIdx) - if err := WriteSidecar(a.sidecarPath, sc); err != nil { - return errors.Wrap(err, "applier: write sidecar for raft cutover") + if err := a.persistSidecar(sc, "raft cutover"); err != nil { + return err } - a.stateCache.RefreshFromSidecar(sc) // Stage 6E-2e-1 BLOCKER (b) — publish the wrap closure on every // replica's local FSM apply so a follower that becomes leader // post-cutover already has wrap active without needing the @@ -1399,10 +1440,9 @@ func (a *Applier) writeRotationSidecar(raftIdx uint64, p fsmwire.RotationPayload Created: a.now().UTC().Format(time.RFC3339), LocalEpoch: keyLocalEpoch, } - if err := WriteSidecar(a.sidecarPath, sc); err != nil { - return nil, errors.Wrap(err, "applier: write sidecar for rotation") + if err := a.persistSidecar(sc, "rotation"); err != nil { + return nil, err } - a.stateCache.RefreshFromSidecar(sc) return sc, nil } diff --git a/internal/encryption/state_cache_observability_test.go b/internal/encryption/state_cache_observability_test.go new file mode 100644 index 000000000..01e18bff0 --- /dev/null +++ b/internal/encryption/state_cache_observability_test.go @@ -0,0 +1,149 @@ +package encryption_test + +import ( + "path/filepath" + "testing" + + "github.com/bootjp/elastickv/internal/encryption" + "github.com/bootjp/elastickv/internal/encryption/fsmwire" + etcdraftengine "github.com/bootjp/elastickv/internal/raftengine/etcd" + "github.com/stretchr/testify/require" +) + +// TestStateCacheMirrorsRaftSlotAndSidecarIndex pins the §9.2 +// observability mirrors. They are refreshed by the same +// RefreshFromSidecar call that maintains the decision mirrors, so a +// refresh that updated one and not the other would let the metrics +// report stale sidecar state indefinitely. +func TestStateCacheMirrorsRaftSlotAndSidecarIndex(t *testing.T) { + t.Parallel() + + cache := encryption.NewStateCache() + + // Pre-bootstrap posture. + id, ok := cache.ActiveRaftKeyID() + require.Zero(t, id) + require.False(t, ok) + require.Zero(t, cache.SidecarRaftAppliedIndex()) + + cache.RefreshFromSidecar(&encryption.Sidecar{ + Version: 1, + RaftAppliedIndex: 9182, + Active: encryption.ActiveKeys{Storage: 7, Raft: 8}, + }) + + id, ok = cache.ActiveRaftKeyID() + require.Equal(t, uint32(8), id) + require.True(t, ok) + require.Equal(t, uint64(9182), cache.SidecarRaftAppliedIndex()) + + // The storage mirror must still track its own slot. + storageID, ok := cache.ActiveStorageKeyID() + require.Equal(t, uint32(7), storageID) + require.True(t, ok) + + // A later refresh must advance the index, not latch it. + cache.RefreshFromSidecar(&encryption.Sidecar{ + Version: 1, + RaftAppliedIndex: 9200, + Active: encryption.ActiveKeys{Storage: 7, Raft: 8}, + }) + require.Equal(t, uint64(9200), cache.SidecarRaftAppliedIndex()) +} + +func TestStateCacheObservabilityMirrorsAreNilSafe(t *testing.T) { + t.Parallel() + + var cache *encryption.StateCache + id, ok := cache.ActiveRaftKeyID() + require.Zero(t, id) + require.False(t, ok) + require.Zero(t, cache.SidecarRaftAppliedIndex()) +} + +// TestApplierRefreshesTheCacheOnNoOpSidecarWrites is the regression for +// the metric-lag defect: the idempotency and stale-DEKID branches +// advance and persist sc.RaftAppliedIndex but do no other work, so a +// branch that skipped RefreshFromSidecar left the cache — and the §9.2 +// elastickv_encryption_sidecar_raft_index gauge — behind the durable +// sidecar until the next fresh mutation or a restart. That reads as a +// false sidecar-divergence signal. +func TestApplierRefreshesTheCacheOnNoOpSidecarWrites(t *testing.T) { + t.Parallel() + + const storageDEK = uint32(7) + const raftDEK = uint32(8) + nodeID := etcdraftengine.DeriveNodeID("n1") + + dir := t.TempDir() + sidecarPath := filepath.Join(dir, "keys.json") + cache := encryption.NewStateCache() + applier := newObservabilityApplier(t, sidecarPath, cache) + + require.NoError(t, applier.ApplyBootstrap(1, fsmwire.BootstrapPayload{ + StorageDEKID: storageDEK, + WrappedStorage: []byte("wrapped-storage-dek"), + RaftDEKID: raftDEK, + WrappedRaft: []byte("wrapped-raft-dek-distinct"), + BatchRegistry: []fsmwire.RegistrationPayload{ + {DEKID: storageDEK, FullNodeID: nodeID, LocalEpoch: 0}, + }, + })) + require.Equal(t, uint64(1), cache.SidecarRaftAppliedIndex()) + + // Fresh cutover. + require.NoError(t, applier.ApplyRotation(2, fsmwire.RotationPayload{ + SubTag: fsmwire.RotateSubEnableStorageEnvelope, + DEKID: storageDEK, + Purpose: fsmwire.PurposeStorage, + Wrapped: []byte{}, + ProposerRegistration: fsmwire.RegistrationPayload{DEKID: storageDEK, FullNodeID: nodeID, LocalEpoch: 1}, + })) + require.Equal(t, uint64(2), cache.SidecarRaftAppliedIndex()) + + // A DUPLICATE cutover entry: the already-active no-op branch. It + // persists the new applied index, so the cache must follow. + require.NoError(t, applier.ApplyRotation(3, fsmwire.RotationPayload{ + SubTag: fsmwire.RotateSubEnableStorageEnvelope, + DEKID: storageDEK, + Purpose: fsmwire.PurposeStorage, + Wrapped: []byte{}, + ProposerRegistration: fsmwire.RegistrationPayload{DEKID: storageDEK, FullNodeID: nodeID, LocalEpoch: 2}, + })) + onDisk, err := encryption.ReadSidecar(sidecarPath) + require.NoError(t, err) + require.Equal(t, uint64(3), onDisk.RaftAppliedIndex, "the no-op branch must persist the index") + require.Equal(t, onDisk.RaftAppliedIndex, cache.SidecarRaftAppliedIndex(), + "the cache must never lag the durable sidecar after a no-op write") + + // A STALE-DEKID cutover entry: the other no-op branch. + require.NoError(t, applier.ApplyRotation(4, fsmwire.RotationPayload{ + SubTag: fsmwire.RotateSubEnableStorageEnvelope, + DEKID: storageDEK + 100, // no longer the active DEK + Purpose: fsmwire.PurposeStorage, + Wrapped: []byte{}, + ProposerRegistration: fsmwire.RegistrationPayload{DEKID: storageDEK + 100, FullNodeID: nodeID, LocalEpoch: 3}, + })) + onDisk, err = encryption.ReadSidecar(sidecarPath) + require.NoError(t, err) + require.Equal(t, uint64(4), onDisk.RaftAppliedIndex) + require.Equal(t, onDisk.RaftAppliedIndex, cache.SidecarRaftAppliedIndex(), + "the stale-DEKID no-op branch must refresh the cache too") +} + +// newObservabilityApplier builds an applier wired to a real sidecar +// path and the supplied cache, matching the production topology in +// main_encryption_write_wiring.go. +func newObservabilityApplier( + t *testing.T, sidecarPath string, cache *encryption.StateCache, +) *encryption.Applier { + t.Helper() + app, err := encryption.NewApplier(newMapRegistryStore(), + encryption.WithKEK(&fakeKEK{}), + encryption.WithKeystore(encryption.NewKeystore()), + encryption.WithSidecarPath(sidecarPath), + encryption.WithStateCache(cache), + ) + require.NoError(t, err) + return app +} diff --git a/main.go b/main.go index bbb963d7e..bcd8874e0 100644 --- a/main.go +++ b/main.go @@ -473,7 +473,10 @@ func run() error { // are only attached to the applier when a KEK source is loaded // (else the applier stays in the Stage 6A posture where // ApplyBootstrap / ApplyRotation return ErrKEKNotConfigured). - kekWrapper, err := loadKEKAfterPreNonceStartupGuards(cfg) + // The KEK source is decorated for §9.2 + // elastickv_encryption_kek_unwrap_seconds inside this call, before + // the startup guards run their own unwraps. + kekUnwrapper, err := loadKEKAfterPreNonceStartupGuards(cfg, metricsRegistry.KEKUnwrapObserver()) if err != nil { return err } @@ -503,7 +506,7 @@ func run() error { return metricsRegistry.RaftProposalObserver(groupID) }, clock, - kekWrapper, + kekUnwrapper, keystore, *encryptionSidecarPath, *encryptionEnabled, @@ -626,7 +629,7 @@ func run() error { defaultRuntime, postCutoverProposerForRuntime(defaultRuntime, shardGroups), *encryptionSidecarPath, - kekWrapper, + kekUnwrapper, encWiring.raftEnvelope, etcdraftengine.DeriveNodeID(*raftId), encWiring.epoch, @@ -644,7 +647,7 @@ func run() error { cleanup: &cleanup, s3BlobBackfiller: s3BlobBackfiller, encWiring: encWiring, - kekConfigured: kekWrapper != nil, + kekConfigured: kekUnwrapper != nil, keyvizSampler: sampler, autoSplitRuntime: autoSplitRuntime, encryptionConfChangeInterceptor: encryptionConfChangeInterceptor, @@ -804,7 +807,7 @@ func startDistributionStartup(in distributionStartupInput) (distributionStartup, if err != nil { return distributionStartup{}, err } - startMonitoringCollectors(in.ctx, in.metricsRegistry, in.runtimes, in.clock) + startMonitoringCollectors(in.ctx, in.metricsRegistry, in.runtimes, in.clock, in.encWiring.cache) startFSMCompactorIfEnabled(in.ctx, in.eg, in.runtimes, in.readTracker) return distributionStartup{ defaultRuntime: defaultRuntime, @@ -1801,7 +1804,9 @@ func appliedIndexForEngine(engine raftengine.Engine) func() uint64 { return applied.AppliedIndex } -func loadKEKAfterPreNonceStartupGuards(cfg runtimeConfig) (kek.Wrapper, error) { +func loadKEKAfterPreNonceStartupGuards( + cfg runtimeConfig, unwrapObserver monitoring.KEKUnwrapObserver, +) (kek.Wrapper, error) { if err := checkEnvelopeCutoverDivergenceBeforeNonceBump( *raftId, *raftDir, @@ -1813,7 +1818,7 @@ func loadKEKAfterPreNonceStartupGuards(cfg runtimeConfig) (kek.Wrapper, error) { ); err != nil { return nil, err } - return loadKEKAndRunStartupGuards() + return loadKEKAndRunStartupGuards(unwrapObserver) } // loadKEKAndRunStartupGuards loads the configured KEK wrapper and @@ -1838,11 +1843,18 @@ func loadKEKAfterPreNonceStartupGuards(cfg runtimeConfig) (kek.Wrapper, error) { // buildShardGroupsWithEncryptionWiring, still before Raft engine // startup; the sidecar-behind-raft-log gap guard remains later // because it needs an opened engine's applied index and scanner. -func loadKEKAndRunStartupGuards() (kek.Wrapper, error) { +func loadKEKAndRunStartupGuards(unwrapObserver monitoring.KEKUnwrapObserver) (kek.Wrapper, error) { kekWrapper, err := loadKEKWrapperFromFlag() if err != nil { return nil, err } + // Decorate BEFORE the guards run. CheckStartupGuards and + // kek.VerifyWrapper both perform real unwrap round trips — on a + // fresh node the preflight unwrap can be the only one that ever + // happens — so decorating after them would leave + // elastickv_encryption_kek_unwrap_seconds empty despite completed + // KMS calls. + kekWrapper = monitoring.NewTimedKEKUnwrapper(kekWrapper, unwrapObserver) if err := encryption.CheckStartupGuards(encryption.StartupConfig{ EncryptionEnabled: *encryptionEnabled, KEKConfigured: kekWrapper != nil, @@ -3304,7 +3316,13 @@ func startMemoryWatchdog(ctx context.Context, eg *errgroup.Group, cancel context // on top of the running raft runtimes. Kept separate from run() so // the latter stays under the cyclop complexity budget and so new // collectors can be added without widening run() further. -func startMonitoringCollectors(ctx context.Context, reg *monitoring.Registry, runtimes []*raftGroupRuntime, clock *kv.HLC) { +func startMonitoringCollectors( + ctx context.Context, + reg *monitoring.Registry, + runtimes []*raftGroupRuntime, + clock *kv.HLC, + encryptionState monitoring.EncryptionStateSource, +) { reg.RaftObserver().Start(ctx, raftMonitorRuntimes(runtimes), raftMetricsObserveInterval) if collector := reg.DispatchCollector(); collector != nil { collector.Start(ctx, dispatchMonitorSources(runtimes), raftMetricsObserveInterval) @@ -3318,6 +3336,9 @@ func startMonitoringCollectors(ctx context.Context, reg *monitoring.Registry, ru if obs := reg.HLCObserver(); obs != nil && clock != nil { obs.Start(ctx, clock, raftMetricsObserveInterval) } + if obs := reg.EncryptionStateObserver(); obs != nil { + obs.Start(ctx, encryptionState, raftMetricsObserveInterval) + } } // startSQSDepthObserver wires the SQS adapter (when enabled on this diff --git a/main_encryption_metrics_wiring_test.go b/main_encryption_metrics_wiring_test.go index e4849e5c4..7ef6127aa 100644 --- a/main_encryption_metrics_wiring_test.go +++ b/main_encryption_metrics_wiring_test.go @@ -1,13 +1,17 @@ package main import ( + "bytes" "context" "sync" "testing" + "time" "github.com/bootjp/elastickv/internal/encryption" + "github.com/bootjp/elastickv/internal/encryption/kek" "github.com/bootjp/elastickv/monitoring" "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" ) // wiringEncryptionObserver records the §9.2 observations the production @@ -120,3 +124,54 @@ func TestMetricsRegistryEncryptionObserverSatisfiesStoreInterface(t *testing.T) obs.ObserveEncryptionWrite(1, 10, 42) obs.ObserveEncryptionDecryptFailure(encryption.DecryptFailureReasonTagMismatch) } + +// countingKEKObserver records how many unwraps were timed. +type countingKEKObserver struct { + mu sync.Mutex + count int +} + +func (o *countingKEKObserver) ObserveEncryptionKEKUnwrap(time.Duration) { + o.mu.Lock() + defer o.mu.Unlock() + o.count++ +} + +func (o *countingKEKObserver) total() int { + o.mu.Lock() + defer o.mu.Unlock() + return o.count +} + +// TestTimedKEKUnwrapperCoversStartupPreflightUnwraps pins the +// decoration ORDER. loadKEKAndRunStartupGuards runs CheckStartupGuards +// and kek.VerifyWrapper, both of which perform real unwrap round trips +// — on a fresh node the preflight unwrap can be the only one that ever +// happens. Decorating after those guards would leave +// elastickv_encryption_kek_unwrap_seconds empty despite completed KMS +// calls, so the decorator must wrap the source before they run. +func TestTimedKEKUnwrapperCoversStartupPreflightUnwraps(t *testing.T) { + t.Parallel() + + obs := &countingKEKObserver{} + timed := monitoring.NewTimedKEKUnwrapper(preflightKEK{}, obs) + require.NotNil(t, timed) + + // Both preflight paths unwrap through the decorated source. + require.NoError(t, kek.VerifyWrapper(timed)) + require.Positive(t, obs.total(), + "the startup preflight unwrap must be timed, not bypass the decorator") +} + +// preflightKEK is a minimal kek.Wrapper for the decoration-order test. +type preflightKEK struct{} + +func (preflightKEK) Name() string { return "preflight-fake" } + +func (preflightKEK) Wrap(dek []byte) ([]byte, error) { + return append([]byte("w:"), dek...), nil +} + +func (preflightKEK) Unwrap(wrapped []byte) ([]byte, error) { + return bytes.TrimPrefix(wrapped, []byte("w:")), nil +} diff --git a/monitoring/encryption.go b/monitoring/encryption.go index cd56e5e66..e6fa1363b 100644 --- a/monitoring/encryption.go +++ b/monitoring/encryption.go @@ -1,13 +1,23 @@ package monitoring import ( + "context" "strconv" "sync" + "time" "github.com/bootjp/elastickv/internal/encryption" + "github.com/bootjp/elastickv/internal/encryption/kek" + "github.com/cockroachdb/errors" "github.com/prometheus/client_golang/prometheus" ) +// Purpose label values for elastickv_encryption_active_dek_id. +const ( + encryptionPurposeStorage = "storage" + encryptionPurposeRaft = "raft" +) + // EncryptionObserver is the surface the storage envelope path uses to // report §9.2 encryption telemetry. The store holds it as an interface // so a node with metrics disabled carries a nil observer and pays only @@ -27,6 +37,18 @@ type EncryptionObserver interface { ObserveEncryptionWrite(keyID uint32, plaintextBytes, payloadBytes int) } +// EncryptionStateSource is the sidecar-state surface the §9.2 +// collector polls. *encryption.StateCache implements it. +// +// Declared here rather than importing the concrete cache so a test can +// drive the collector without building an applier, matching the +// HLCSource precedent in this package. +type EncryptionStateSource interface { + ActiveStorageKeyID() (uint32, bool) + ActiveRaftKeyID() (uint32, bool) + SidecarRaftAppliedIndex() uint64 +} + // EncryptionMetrics implements EncryptionObserver over Prometheus. // // writesPerDEK is labelled by key_id, whose cardinality is bounded by @@ -40,6 +62,10 @@ type EncryptionMetrics struct { writesPerDEK *prometheus.CounterVec valueOverhead prometheus.Histogram + activeDEKID *prometheus.GaugeVec + sidecarRaftIdx prometheus.Gauge + kekUnwrapSecond prometheus.Histogram + mu sync.RWMutex writeCtr map[uint32]prometheus.Counter } @@ -71,16 +97,74 @@ func newEncryptionMetrics(registerer prometheus.Registerer) *EncryptionMetrics { Buckets: []float64{-1048576, -65536, -4096, -256, 0, 16, 32, 48, 64, 96, 128, 256, 1024, 4096, 65536}, }, ), + activeDEKID: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "elastickv_encryption_active_dek_id", + Help: "Currently active DEK id per purpose; 0 means the cluster has not bootstrapped that purpose.", + }, + []string{"purpose"}, + ), + sidecarRaftIdx: prometheus.NewGauge( + prometheus.GaugeOpts{ + Name: "elastickv_encryption_sidecar_raft_index", + Help: "The encryption sidecar's persisted raft_applied_index. A persistent gap below the FSM applied index is the sidecar-divergence signal.", + }, + ), + kekUnwrapSecond: prometheus.NewHistogram( + prometheus.HistogramOpts{ + Name: "elastickv_encryption_kek_unwrap_seconds", + Help: "KEK unwrap round-trip latency. For a remote KMS this is a network call; sustained growth indicates a KMS outage.", + // Spans a local file unwrap (microseconds) through a + // remote KMS call and into timeout territory. + Buckets: []float64{0.0001, 0.0005, 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30}, + }, + ), writeCtr: make(map[uint32]prometheus.Counter), } registerer.MustRegister( m.decryptFailures, m.writesPerDEK, m.valueOverhead, + m.activeDEKID, + m.sidecarRaftIdx, + m.kekUnwrapSecond, ) + // Publish the pre-bootstrap posture immediately so both purposes + // exist as series from process start. Without this an alert on + // active_dek_id == 0 could not distinguish "not bootstrapped" + // from "this node never reported", which are very different. + m.activeDEKID.WithLabelValues(encryptionPurposeStorage).Set(0) + m.activeDEKID.WithLabelValues(encryptionPurposeRaft).Set(0) return m } +// ObserveEncryptionKEKUnwrap records one KEK unwrap round-trip. +func (m *EncryptionMetrics) ObserveEncryptionKEKUnwrap(d time.Duration) { + if m == nil { + return + } + if d < 0 { + d = 0 + } + m.kekUnwrapSecond.Observe(d.Seconds()) +} + +// observeState publishes one sample of the sidecar-derived gauges. +func (m *EncryptionMetrics) observeState(source EncryptionStateSource) { + if m == nil || source == nil { + return + } + storageID, _ := source.ActiveStorageKeyID() + raftID, _ := source.ActiveRaftKeyID() + // The `ok` half is deliberately dropped: it is exactly + // `id != 0`, and 0 is the design's "not bootstrapped" sentinel, + // so reporting the raw id keeps the metric and the sentinel + // carrying the same meaning. + m.activeDEKID.WithLabelValues(encryptionPurposeStorage).Set(float64(storageID)) + m.activeDEKID.WithLabelValues(encryptionPurposeRaft).Set(float64(raftID)) + m.sidecarRaftIdx.Set(float64(source.SidecarRaftAppliedIndex())) +} + // ObserveEncryptionDecryptFailure counts one decrypt-path failure. // An unrecognised reason is folded into the doc's `unknown` bucket // rather than registering a new label value, so the series stays @@ -140,3 +224,91 @@ func normalizeDecryptFailureReason(reason string) string { return encryption.DecryptFailureReasonUnknown } } + +// EncryptionStateObserver polls an EncryptionStateSource and mirrors +// it into the §9.2 gauges. +type EncryptionStateObserver struct { + metrics *EncryptionMetrics +} + +func newEncryptionStateObserver(metrics *EncryptionMetrics) *EncryptionStateObserver { + return &EncryptionStateObserver{metrics: metrics} +} + +// Start samples source immediately and then on every tick until ctx is +// cancelled. A nil receiver or nil source silently no-ops so a node +// without encryption wired needs no conditional at the call site. +func (o *EncryptionStateObserver) Start(ctx context.Context, source EncryptionStateSource, interval time.Duration) { + if o == nil || source == nil { + return + } + if interval <= 0 { + interval = defaultObserveInterval + } + o.metrics.observeState(source) + ticker := time.NewTicker(interval) + go func() { + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + o.metrics.observeState(source) + } + } + }() +} + +// TimedKEKUnwrapper decorates a KEK source so every unwrap feeds +// elastickv_encryption_kek_unwrap_seconds. +// +// It wraps kek.Wrapper (rather than the narrower +// encryption.KEKUnwrapper) because that is the type main.go threads +// through the shard-group wiring; Wrap and Name delegate untouched. +// Only Unwrap is timed, per §9.2: unwrap is the call on the startup +// and apply paths, so a KMS outage surfaces there first. +type TimedKEKUnwrapper struct { + inner kek.Wrapper + observer KEKUnwrapObserver + now func() time.Time +} + +// KEKUnwrapObserver receives KEK unwrap durations. +type KEKUnwrapObserver interface { + ObserveEncryptionKEKUnwrap(d time.Duration) +} + +// NewTimedKEKUnwrapper returns inner unchanged when either inner or +// observer is nil, so wiring stays a single unconditional call and a +// node without a KEK source keeps passing the same nil it had before. +func NewTimedKEKUnwrapper(inner kek.Wrapper, observer KEKUnwrapObserver) kek.Wrapper { + if inner == nil || observer == nil { + return inner + } + return &TimedKEKUnwrapper{inner: inner, observer: observer, now: time.Now} +} + +// Unwrap times the inner call. Failures are timed too: a KMS outage +// usually shows up as slow errors, and excluding them would hide the +// signal this histogram exists to expose. +func (u *TimedKEKUnwrapper) Unwrap(wrapped []byte) ([]byte, error) { + start := u.now() + out, err := u.inner.Unwrap(wrapped) + u.observer.ObserveEncryptionKEKUnwrap(u.now().Sub(start)) + if err != nil { + // errors.Wrap preserves Is/As, so the startup guards that + // match ErrKEKMismatch and friends still see through this. + return nil, errors.Wrapf(err, "kek %s: unwrap", u.inner.Name()) + } + return out, nil +} + +// Wrap delegates untouched. +func (u *TimedKEKUnwrapper) Wrap(dek []byte) ([]byte, error) { + return u.inner.Wrap(dek) +} + +// Name delegates so the decorated source still reports its real +// provider ("file", "aws-kms", ...) in logs and the status RPC. +func (u *TimedKEKUnwrapper) Name() string { return u.inner.Name() } diff --git a/monitoring/encryption_test.go b/monitoring/encryption_test.go index c69e932eb..594c6149b 100644 --- a/monitoring/encryption_test.go +++ b/monitoring/encryption_test.go @@ -1,11 +1,14 @@ package monitoring import ( + "context" "strings" "sync" "testing" + "time" "github.com/bootjp/elastickv/internal/encryption" + "github.com/cockroachdb/errors" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" @@ -166,3 +169,226 @@ func TestRegistryExposesEncryptionObserver(t *testing.T) { var nilRegistry *Registry require.Nil(t, nilRegistry.EncryptionObserver()) } + +// fakeEncryptionState drives the §9.2 sidecar gauges. +type fakeEncryptionState struct { + storageID uint32 + raftID uint32 + raftIndex uint64 +} + +func (f fakeEncryptionState) ActiveStorageKeyID() (uint32, bool) { + return f.storageID, f.storageID != 0 +} + +func (f fakeEncryptionState) ActiveRaftKeyID() (uint32, bool) { + return f.raftID, f.raftID != 0 +} + +func (f fakeEncryptionState) SidecarRaftAppliedIndex() uint64 { return f.raftIndex } + +// TestEncryptionMetricsPublishPreBootstrapPostureAtConstruction pins +// that both purpose series exist from process start. Without it an +// alert on active_dek_id == 0 could not tell "not bootstrapped" from +// "this node never reported", which are very different incidents. +func TestEncryptionMetricsPublishPreBootstrapPostureAtConstruction(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + metrics := newEncryptionMetrics(reg) + + require.NoError(t, testutil.GatherAndCompare( + reg, + strings.NewReader(` +# HELP elastickv_encryption_active_dek_id Currently active DEK id per purpose; 0 means the cluster has not bootstrapped that purpose. +# TYPE elastickv_encryption_active_dek_id gauge +elastickv_encryption_active_dek_id{purpose="raft"} 0 +elastickv_encryption_active_dek_id{purpose="storage"} 0 +`), + "elastickv_encryption_active_dek_id", + )) + require.NotNil(t, metrics) +} + +func TestEncryptionMetricsObserveSidecarState(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + metrics := newEncryptionMetrics(reg) + + metrics.observeState(fakeEncryptionState{storageID: 7, raftID: 8, raftIndex: 4211}) + + require.NoError(t, testutil.GatherAndCompare( + reg, + strings.NewReader(` +# HELP elastickv_encryption_active_dek_id Currently active DEK id per purpose; 0 means the cluster has not bootstrapped that purpose. +# TYPE elastickv_encryption_active_dek_id gauge +elastickv_encryption_active_dek_id{purpose="raft"} 8 +elastickv_encryption_active_dek_id{purpose="storage"} 7 +# HELP elastickv_encryption_sidecar_raft_index The encryption sidecar's persisted raft_applied_index. A persistent gap below the FSM applied index is the sidecar-divergence signal. +# TYPE elastickv_encryption_sidecar_raft_index gauge +elastickv_encryption_sidecar_raft_index 4211 +`), + "elastickv_encryption_active_dek_id", + "elastickv_encryption_sidecar_raft_index", + )) +} + +// TestEncryptionStateObserverSamplesImmediatelyAndOnTick covers the +// startup case: an operator restarting a node must not wait a full +// interval before the gauges reflect reality. +func TestEncryptionStateObserverSamplesImmediatelyAndOnTick(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + metrics := newEncryptionMetrics(reg) + observer := newEncryptionStateObserver(metrics) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + observer.Start(ctx, fakeEncryptionState{storageID: 3, raftID: 4, raftIndex: 99}, time.Hour) + + // No tick has fired; the immediate sample must already be visible. + require.InDelta(t, 3.0, testutil.ToFloat64( + metrics.activeDEKID.WithLabelValues(encryptionPurposeStorage)), 0.0001) + require.InDelta(t, 4.0, testutil.ToFloat64( + metrics.activeDEKID.WithLabelValues(encryptionPurposeRaft)), 0.0001) + require.InDelta(t, 99.0, testutil.ToFloat64(metrics.sidecarRaftIdx), 0.0001) +} + +func TestEncryptionStateObserverIsInertWithoutSource(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + observer := newEncryptionStateObserver(newEncryptionMetrics(reg)) + + var nilObserver *EncryptionStateObserver + require.NotPanics(t, func() { + observer.Start(context.Background(), nil, time.Second) + nilObserver.Start(context.Background(), fakeEncryptionState{}, time.Second) + }) +} + +// fakeKEK is a kek.Wrapper whose Unwrap can be made slow and failing. +type fakeKEK struct { + unwrapErr error + calls int +} + +func (k *fakeKEK) Wrap(dek []byte) ([]byte, error) { return append([]byte("w:"), dek...), nil } +func (k *fakeKEK) Name() string { return "fake" } +func (k *fakeKEK) Unwrap(wrapped []byte) ([]byte, error) { + k.calls++ + if k.unwrapErr != nil { + return nil, k.unwrapErr + } + return append([]byte("u:"), wrapped...), nil +} + +func TestTimedKEKUnwrapperRecordsLatencyAndDelegates(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + metrics := newEncryptionMetrics(reg) + inner := &fakeKEK{} + + timed := NewTimedKEKUnwrapper(inner, metrics) + require.NotNil(t, timed) + + // Drive a deterministic 250ms round trip. + decorator, ok := timed.(*TimedKEKUnwrapper) + require.True(t, ok) + base := time.Unix(1_700_000_000, 0) + step := 0 + decorator.now = func() time.Time { + step++ + if step == 1 { + return base + } + return base.Add(250 * time.Millisecond) + } + + out, err := timed.Unwrap([]byte("dek")) + require.NoError(t, err) + require.Equal(t, []byte("u:dek"), out) + require.Equal(t, 1, inner.calls) + + // Name and Wrap must pass through untouched. + require.Equal(t, "fake", timed.Name()) + wrapped, err := timed.Wrap([]byte("dek")) + require.NoError(t, err) + require.Equal(t, []byte("w:dek"), wrapped) + + families, err := reg.Gather() + require.NoError(t, err) + var sum float64 + for _, family := range families { + if family.GetName() == "elastickv_encryption_kek_unwrap_seconds" { + sum = family.GetMetric()[0].GetHistogram().GetSampleSum() + } + } + require.InDelta(t, 0.25, sum, 0.0001) +} + +// TestTimedKEKUnwrapperTimesAndPreservesFailures pins two things at +// once: a failing unwrap is still timed (a KMS outage shows up as slow +// errors, and dropping them would hide the signal), and the decorator +// stays transparent to errors.Is so the startup guards that match +// ErrKEKMismatch keep working through it. +func TestTimedKEKUnwrapperTimesAndPreservesFailures(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + metrics := newEncryptionMetrics(reg) + sentinel := encryption.ErrKEKMismatch + timed := NewTimedKEKUnwrapper(&fakeKEK{unwrapErr: sentinel}, metrics) + + _, err := timed.Unwrap([]byte("dek")) + require.Error(t, err) + require.True(t, errors.Is(err, sentinel), + "the decorator must not hide the inner typed error from the startup guards") + + require.Equal(t, uint64(1), gatheredHistogramCount(t, reg, "elastickv_encryption_kek_unwrap_seconds"), + "a failed unwrap must still be timed") +} + +// gatheredHistogramCount returns a histogram's observation count. +// CollectAndCount is the wrong tool here: it counts SERIES, and a +// histogram is one series whether or not anything was observed. +func gatheredHistogramCount(t *testing.T, reg *prometheus.Registry, name string) uint64 { + t.Helper() + families, err := reg.Gather() + require.NoError(t, err) + for _, family := range families { + if family.GetName() != name { + continue + } + require.Len(t, family.GetMetric(), 1) + return family.GetMetric()[0].GetHistogram().GetSampleCount() + } + t.Fatalf("histogram %s not registered", name) + return 0 +} + +func TestNewTimedKEKUnwrapperReturnsInnerWhenNotObservable(t *testing.T) { + t.Parallel() + + inner := &fakeKEK{} + require.Same(t, inner, NewTimedKEKUnwrapper(inner, nil), + "a node without metrics must keep the undecorated source") + + var nilRegistry *Registry + require.Nil(t, NewTimedKEKUnwrapper(nil, nilRegistry.KEKUnwrapObserver())) +} + +func TestRegistryExposesEncryptionStateAndKEKObservers(t *testing.T) { + t.Parallel() + + reg := NewRegistry("n1", "127.0.0.1:1") + require.NotNil(t, reg.EncryptionStateObserver()) + require.NotNil(t, reg.KEKUnwrapObserver()) + + var nilRegistry *Registry + require.Nil(t, nilRegistry.EncryptionStateObserver()) + require.Nil(t, nilRegistry.KEKUnwrapObserver()) +} diff --git a/monitoring/registry.go b/monitoring/registry.go index 0a7631415..159bbe822 100644 --- a/monitoring/registry.go +++ b/monitoring/registry.go @@ -32,6 +32,7 @@ type Registry struct { tso *TSOMetrics tsoObserver *TSOObserver encryption *EncryptionMetrics + encryptionObs *EncryptionStateObserver } // NewRegistry builds a registry with constant labels that identify the local node. @@ -65,6 +66,7 @@ func NewRegistry(nodeID string, nodeAddress string) *Registry { r.tso = newTSOMetrics(registerer) r.tsoObserver = newTSOObserver(r.tso) r.encryption = newEncryptionMetrics(registerer) + r.encryptionObs = newEncryptionStateObserver(r.encryption) return r } @@ -308,3 +310,23 @@ func (r *Registry) EncryptionObserver() EncryptionObserver { } return r.encryption } + +// EncryptionStateObserver returns the collector that mirrors sidecar +// state (active DEK ids, sidecar raft_applied_index) into the §9.2 +// gauges. Start it with the process-shared encryption StateCache. +func (r *Registry) EncryptionStateObserver() *EncryptionStateObserver { + if r == nil { + return nil + } + return r.encryptionObs +} + +// KEKUnwrapObserver returns the observer that records KEK unwrap +// latency. Pass it to monitoring.NewTimedKEKUnwrapper alongside the +// loaded KEK source. +func (r *Registry) KEKUnwrapObserver() KEKUnwrapObserver { + if r == nil || r.encryption == nil { + return nil + } + return r.encryption +}