From 16592c464f140eee691be3dce031cbdd9e8b0137 Mon Sep 17 00:00:00 2001 From: bootjp Date: Mon, 7 Sep 2026 21:44:15 +0900 Subject: [PATCH 1/2] =?UTF-8?q?encryption:=20add=20sidecar=20and=20KEK=20m?= =?UTF-8?q?etrics=20(Stage=209C-2,=20=C2=A79.2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the three remaining §9.2 metrics that have live sources today: elastickv_encryption_active_dek_id{purpose} elastickv_encryption_sidecar_raft_index elastickv_encryption_kek_unwrap_seconds StateCache gains observability-only mirrors of sidecar.Active.Raft and sidecar.RaftAppliedIndex, refreshed by the same RefreshFromSidecar call that maintains the decision mirrors so they cannot drift from them. A polling observer publishes them, sampling immediately at startup rather than making an operator wait a full interval after a restart. Both purpose series are published at 0 on construction: without that, an alert on active_dek_id == 0 cannot distinguish "not bootstrapped" from "this node never reported", which are very different incidents. KEK latency is captured by decorating the loaded KEK source once in main.go, so every unwrap path (startup hydration, bootstrap apply, rotation apply) is timed without per-path instrumentation. Failed unwraps are timed too — a KMS outage surfaces as slow errors, and excluding them would hide the signal the histogram exists to expose. last_proposed_index_per_raft_dek still needs the §5.4 raft-DEK Wrap path and stays unregistered rather than always-zero. Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE --- ...6_04_29_partial_data_at_rest_encryption.md | 15 +- internal/encryption/applier.go | 34 +++ .../state_cache_observability_test.go | 59 +++++ main.go | 20 +- monitoring/encryption.go | 172 +++++++++++++ monitoring/encryption_test.go | 226 ++++++++++++++++++ monitoring/registry.go | 22 ++ 7 files changed, 539 insertions(+), 9 deletions(-) create mode 100644 internal/encryption/state_cache_observability_test.go 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..5d56e3c9a 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 diff --git a/internal/encryption/state_cache_observability_test.go b/internal/encryption/state_cache_observability_test.go new file mode 100644 index 000000000..8d1201719 --- /dev/null +++ b/internal/encryption/state_cache_observability_test.go @@ -0,0 +1,59 @@ +package encryption_test + +import ( + "testing" + + "github.com/bootjp/elastickv/internal/encryption" + "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()) +} diff --git a/main.go b/main.go index bbb963d7e..2136a490e 100644 --- a/main.go +++ b/main.go @@ -477,6 +477,11 @@ func run() error { if err != nil { return err } + // §9.2 elastickv_encryption_kek_unwrap_seconds. Decorating here — + // the single place the KEK source is loaded — means every unwrap + // path (startup hydration, bootstrap apply, rotation apply) is + // timed without each one needing its own instrumentation. + kekUnwrapper := monitoring.NewTimedKEKUnwrapper(kekWrapper, metricsRegistry.KEKUnwrapObserver()) keystore := encryption.NewKeystore() redisApplyObserver := adapter.NewRedisApplyObserver() readTracker := kv.NewActiveTimestampTracker(kv.WithActiveTimestampTrackerMaxBackupPins(*backupMaxActivePins)) @@ -503,7 +508,7 @@ func run() error { return metricsRegistry.RaftProposalObserver(groupID) }, clock, - kekWrapper, + kekUnwrapper, keystore, *encryptionSidecarPath, *encryptionEnabled, @@ -804,7 +809,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, @@ -3304,7 +3309,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 +3329,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/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 +} From b978c3dcd51455a1d481631ff233c0454d0e0a9a Mon Sep 17 00:00:00 2001 From: bootjp Date: Mon, 7 Sep 2026 22:19:29 +0900 Subject: [PATCH 2/2] encryption: keep the sidecar mirror and KEK timing complete MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two P2s from review. The no-op apply branches (stale DEKID, already-active cutover) advance and persist sc.RaftAppliedIndex but returned without refreshing the StateCache, so the mirror — and therefore elastickv_encryption_sidecar_raft_index — sat behind the durable sidecar until the next fresh mutation or a restart, which reads as a false sidecar-divergence signal. Fixed structurally rather than at the two named branches: all eight sidecar-write sites now go through Applier.persistSidecar, which pairs the write with the refresh. Patching only the reported branches would have left the same latent defect in every future one. The KEK decorator was constructed after loadKEKAfterPreNonceStartupGuards had already run CheckStartupGuards and kek.VerifyWrapper against the raw wrapper. Both perform real unwrap round trips, and on a fresh node the preflight unwrap can be the only one that ever happens, so elastickv_encryption_kek_unwrap_seconds could stay empty despite completed KMS calls. The source is now decorated inside the loader, before those guards run. Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE --- internal/encryption/applier.go | 54 ++++++----- .../state_cache_observability_test.go | 90 +++++++++++++++++++ main.go | 29 +++--- main_encryption_metrics_wiring_test.go | 55 ++++++++++++ 4 files changed, 193 insertions(+), 35 deletions(-) diff --git a/internal/encryption/applier.go b/internal/encryption/applier.go index 5d56e3c9a..474710505 100644 --- a/internal/encryption/applier.go +++ b/internal/encryption/applier.go @@ -896,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 @@ -1123,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 @@ -1143,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. // @@ -1182,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 } @@ -1311,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 @@ -1329,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 @@ -1348,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 @@ -1433,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 index 8d1201719..01e18bff0 100644 --- a/internal/encryption/state_cache_observability_test.go +++ b/internal/encryption/state_cache_observability_test.go @@ -1,9 +1,12 @@ 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" ) @@ -57,3 +60,90 @@ func TestStateCacheObservabilityMirrorsAreNilSafe(t *testing.T) { 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 2136a490e..bcd8874e0 100644 --- a/main.go +++ b/main.go @@ -473,15 +473,13 @@ 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 } - // §9.2 elastickv_encryption_kek_unwrap_seconds. Decorating here — - // the single place the KEK source is loaded — means every unwrap - // path (startup hydration, bootstrap apply, rotation apply) is - // timed without each one needing its own instrumentation. - kekUnwrapper := monitoring.NewTimedKEKUnwrapper(kekWrapper, metricsRegistry.KEKUnwrapObserver()) keystore := encryption.NewKeystore() redisApplyObserver := adapter.NewRedisApplyObserver() readTracker := kv.NewActiveTimestampTracker(kv.WithActiveTimestampTrackerMaxBackupPins(*backupMaxActivePins)) @@ -631,7 +629,7 @@ func run() error { defaultRuntime, postCutoverProposerForRuntime(defaultRuntime, shardGroups), *encryptionSidecarPath, - kekWrapper, + kekUnwrapper, encWiring.raftEnvelope, etcdraftengine.DeriveNodeID(*raftId), encWiring.epoch, @@ -649,7 +647,7 @@ func run() error { cleanup: &cleanup, s3BlobBackfiller: s3BlobBackfiller, encWiring: encWiring, - kekConfigured: kekWrapper != nil, + kekConfigured: kekUnwrapper != nil, keyvizSampler: sampler, autoSplitRuntime: autoSplitRuntime, encryptionConfChangeInterceptor: encryptionConfChangeInterceptor, @@ -1806,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, @@ -1818,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 @@ -1843,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, 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 +}