Race condition / Data integrity

// Proof-of-concept (Go) inspired by Grafana's tests for this fix // The following demonstrates a cancellation-after-data-write scenario where // the data write is committed but the client cancels before the event is emitted. // It uses a wrapper KV that cancels after the data-section write completes. type cancelOnDataSaveKV struct { KV kvStore // hypothetical KV interface from Grafana codebase cancel func() once sync.Once } func (w *cancelOnDataSaveKV) Save(ctx context.Context, section, key string) (io.WriteCloser, error) { wc, err := w.KV.Save(ctx, section, key) if err != nil || section != kv.DataSection { return wc, err } return &cancelOnCloseWriteCloser{WriteCloser: wc, w: w}, nil } type cancelOnCloseWriteCloser struct { io.WriteCloser w *cancelOnDataSaveKV } func (c *cancelOnCloseWriteCloser) Close() error { err := c.WriteCloser.Close() if err == nil { c.w.once.Do(c.w.cancel) } return err } // PoC usage (pseudo; assumes Grafana's test helpers available): func PoC_Main() { wrapper := &cancelOnDataSaveKV{KV: setupBadgerKV(nil)} backend := setupTestStorageBackend(nil, withKV(wrapper)) ctx, cancel := context.WithCancel(context.Background()) wrapper.cancel = cancel defer cancel() testObj, _ := createTestObject() metaAccessor, _ := utils.MetaAccessor(testObj) addEvent := WriteEvent{ Type: resourcepb.WatchEvent_ADDED, Key: &resourcepb.ResourceKey{Namespace: "default", Group: "apps", Resource: "resources", Name: "poc"}, Value: objectToJSONBytes(testObj), Object: metaAccessor, } rv, err := backend.WriteEvent(ctx, addEvent) if err != nil || rv <= 0 { panic("PoC failed: event not written or unexpected error: " + err.Error()) } // Verify the event persisted even though client canceled after data write _ = verifyEventPersisted(backend.eventStore, addEvent.Key) }

diff --git a/pkg/setting/setting.go b/pkg/setting/setting.go index fa58afce72557..a2dd5c3c473dd 100644 --- a/pkg/setting/setting.go +++ b/pkg/setting/setting.go @@ -817,6 +817,8 @@ type Cfg struct { // TODO: remove this when sql/backend backwards compatibility is no longer needed. LogSQLBackendCalls bool EnableKVLeases bool + KVLeaseTTL time.Duration + KVLeaseAutoRenew bool EnableGarbageCollection bool GarbageCollectionDryRun bool GarbageCollectionInterval time.Duration diff --git a/pkg/setting/setting_unified_storage.go b/pkg/setting/setting_unified_storage.go index 3235694796157..9c7aee90adcfd 100644 --- a/pkg/setting/setting_unified_storage.go +++ b/pkg/setting/setting_unified_storage.go @@ -264,9 +264,13 @@ func (cfg *Cfg) setUnifiedStorageConfig() { // (temporary smoke-test instrumentation; default off) // TODO: remove this when sql/backend backwards compatibility is no longer needed. cfg.LogSQLBackendCalls = section.Key("log_sql_backend_calls").MustBool(false) - // enable per-resource leases in the KV backend; only effective when the - // SQL RV manager is not in use. + // enable per-resource leases in the KV backend; cfg.EnableKVLeases = section.Key("enable_kv_leases").MustBool(false) + // TTL for per-resource write leases; 0 uses the backend default (10s). + cfg.KVLeaseTTL = section.Key("kv_lease_ttl").MustDuration(0) + // auto-renew write leases in the background so they are not lost while a + // slow write is still in flight. + cfg.KVLeaseAutoRenew = section.Key("kv_lease_auto_renew").MustBool(false) cfg.MaxFileIndexAge = section.Key("max_file_index_age").MustDuration(0) cfg.MinFileIndexBuildVersion = section.Key("min_file_index_build_version").MustString("") diff --git a/pkg/storage/unified/resource/storage_backend.go b/pkg/storage/unified/resource/storage_backend.go index a7d62a10dea15..82c536edfd943 100644 --- a/pkg/storage/unified/resource/storage_backend.go +++ b/pkg/storage/unified/resource/storage_backend.go @@ -45,6 +45,7 @@ const ( defaultEventPruningInterval = 5 * time.Minute defaultSearchLookback = 1 * time.Second defaultGarbageCollectionBatchWait = 1 * time.Second + persistDeadline = 10 * time.Second ) // IsResourceNameMixedCase reports whether a successful read returned a @@ -104,6 +105,13 @@ type kvStorageBackend struct { // resource via per-resource leases. Ignored when using `rvManager`. leaseManager *lease.Manager + // leaseTTL is the TTL applied when acquiring a write lease. Zero uses the + // lease package default. + leaseTTL time.Duration + + // leaseAutoRenew enables background auto-renewal of write leases. + leaseAutoRenew bool + // dbKeepAlive holds a reference to the database provider/connection owner to prevent it from being GC'd dbKeepAlive any @@ -122,7 +130,8 @@ type kvStorageBackend struct { } type kvBackendMetrics struct { - ConflictErrors *prometheus.CounterVec + ConflictErrors *prometheus.CounterVec + EventEmitFailures *prometheus.CounterVec } func newKVBackendMetrics(reg prometheus.Registerer) *kvBackendMetrics { @@ -132,6 +141,11 @@ func newKVBackendMetrics(reg prometheus.Registerer) *kvBackendMetrics { Name: "optimistic_lock_conflicts_total", Help: "Total number of optimistic lock conflict errors in the KV storage backend", }, []string{"resource", "action"}), + EventEmitFailures: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Namespace: "storage_server", + Name: "event_emit_after_commit_failures_total", + Help: "Total number of writes whose data was committed but whose event failed to be emitted", + }, []string{"resource", "action"}), } } @@ -142,6 +156,13 @@ func (m *kvBackendMetrics) recordConflict(event WriteEvent) { m.ConflictErrors.WithLabelValues(event.Key.Resource, event.Type.String()).Inc() } +func (m *kvBackendMetrics) recordEventEmitFailure(event WriteEvent) { + if m == nil { + return + } + m.EventEmitFailures.WithLabelValues(event.Key.Resource, event.Type.String()).Inc() +} + var _ KVBackend = &kvStorageBackend{} type KVBackend interface { @@ -214,6 +235,15 @@ type KVBackendOptions struct { // Holder identifies this process for lease ownership. Required when // EnableKVLeases is true. Holder string + + // LeaseTTL overrides the per-resource write lease TTL. Zero uses the lease + // package default (10s). Only effective when EnableKVLeases is true. + LeaseTTL time.Duration + + // LeaseAutoRenew enables background auto-renewal of the write lease so it + // is not lost while a slow write is still in flight. Only effective when + // EnableKVLeases is true. + LeaseAutoRenew bool } var ( @@ -290,6 +320,8 @@ func NewKVStorageBackend(opts KVBackendOptions) (KVBackend, error) { eventPruningInterval: eventPruningInterval, rvManager: opts.RvManager, leaseManager: leaseManager, + leaseTTL: opts.LeaseTTL, + leaseAutoRenew: opts.LeaseAutoRenew, dbKeepAlive: opts.DBKeepAlive, lastImportStore: newLastImportStore(kv), lastImportTimeMaxAge: opts.LastImportTimeMaxAge, @@ -855,7 +887,15 @@ func (k *kvStorageBackend) maybeAcquireWriteLease(ctx context.Context, event Wri name = event.Key.Group + "/" + event.Key.Resource + "/" + event.Key.Name } - l, err := k.leaseManager.Acquire(ctx, name) + var acquireOpts []lease.AcquireOption + if k.leaseTTL > 0 { + acquireOpts = append(acquireOpts, lease.WithTTL(k.leaseTTL)) + } + if k.leaseAutoRenew { + acquireOpts = append(acquireOpts, lease.WithAutoRenew()) + } + + l, err := k.leaseManager.Acquire(ctx, name, acquireOpts...) if err != nil { if errors.Is(err, lease.ErrLeaseAlreadyHeld) { k.metrics.recordConflict(event) @@ -1079,7 +1119,11 @@ func (k *kvStorageBackend) WriteEvent(ctx context.Context, event WriteEvent) (rv if k.rvManager != nil { dataKey.GUID = uuid.New().String() var err error - rv, err = k.rvManager.ExecWithRV(ctx, event.Key, func(txnCtx context.Context, tx db.Tx) (string, error) { + // ExecWithRV commits the data on its own context regardless of client cancellation. + // Passing a detached context makes ExecWithRV wait for that guaranteed commit + // instead of bailing out on cancellation and losing the assigned resource version (which would + // leave the data committed but the event unwritten). + rv, err = k.rvManager.ExecWithRV(context.WithoutCancel(ctx), event.Key, func(txnCtx context.Context, tx db.Tx) (string, error) { if err := k.dataStore.Save(kv.ContextWithTx(txnCtx, tx), dataKey, bytes.NewReader(event.Value)); err != nil { return "", fmt.Errorf("failed to write data: %w", err) } @@ -1115,6 +1159,14 @@ func (k *kvStorageBackend) WriteEvent(ctx context.Context, event WriteEvent) (rv } } + // The data is now durably committed. From here on the data store and event + // store must not diverge: detach from client/lease cancellation so the event + // is still emitted, but keep a bounded deadline so a stuck datastore cannot + // hold a worker indefinitely. + persistCtx, cancelPersist := context.WithTimeout(context.WithoutCancel(ctx), persistDeadline) + defer cancelPersist() + ctx = persistCtx + // Optimistic concurrency control to verify our write is the latest version // and that the resource still had the expected PreviousRV when we wrote it. if !leasesActive { @@ -1191,6 +1243,15 @@ func (k *kvStorageBackend) WriteEvent(ctx context.Context, event WriteEvent) (rv PreviousRV: event.PreviousRV, } if err := k.eventStore.Save(ctx, eventData); err != nil { + k.metrics.recordEventEmitFailure(event) + k.log.Error("data committed but failed to emit event; data and event store diverged", + "namespace", event.Key.Namespace, + "group", event.Key.Group, + "resource", event.Key.Resource, + "name", event.Key.Name, + "action", action, + "error", err, + ) return 0, fmt.Errorf("failed to save event: %w", err) } diff --git a/pkg/storage/unified/resource/storage_backend_test.go b/pkg/storage/unified/resource/storage_backend_test.go index a52930be8c9d3..83b7a3f3d7934 100644 --- a/pkg/storage/unified/resource/storage_backend_test.go +++ b/pkg/storage/unified/resource/storage_backend_test.go @@ -264,6 +264,84 @@ func TestKvStorageBackend_WriteEvent_Success(t *testing.T) { require.NoError(t, err) } +// cancelOnDataSaveKV wraps a KV and cancels a context the first time a value is +// durably written to the data section. This simulates a client cancelling the +// request in the window between the data store commit and the event store +// write, which used to leave the two stores split: the data was persisted but +// the event never was, so watch consumers and search never saw the write. +type cancelOnDataSaveKV struct { + KV + cancel func() + once sync.Once +} + +func (w *cancelOnDataSaveKV) Save(ctx context.Context, section, key string) (io.WriteCloser, error) { + wc, err := w.KV.Save(ctx, section, key) + if err != nil || section != kv.DataSection { + return wc, err + } + return &cancelOnCloseWriteCloser{WriteCloser: wc, w: w}, nil +} + +type cancelOnCloseWriteCloser struct { + io.WriteCloser + w *cancelOnDataSaveKV +} + +func (c *cancelOnCloseWriteCloser) Close() error { + err := c.WriteCloser.Close() + if err == nil { + c.w.once.Do(c.w.cancel) + } + return err +} + +// TestKvStorageBackend_WriteEvent_ClientCancelAfterDataSave_PersistsEvent +// verifies that once the data has been durably written, cancelling the client's +// context does not abort the event write. Otherwise the data store and event +// store diverge and consumers miss the event through watch and search. +func TestKvStorageBackend_WriteEvent_ClientCancelAfterDataSave_PersistsEvent(t *testing.T) { + wrapper := &cancelOnDataSaveKV{KV: setupBadgerKV(t)} + backend := setupTestStorageBackend(t, withKV(wrapper)) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + wrapper.cancel = cancel + + testObj, err := createTestObject() + require.NoError(t, err) + metaAccessor, err := utils.MetaAccessor(testObj) + require.NoError(t, err) + + const resourceName = "cancel-after-data-save" + addEvent := WriteEvent{ + Type: resourcepb.WatchEvent_ADDED, + Key: &resourcepb.ResourceKey{ + Namespace: "default", + Group: "apps", + Resource: "resources", + Name: resourceName, + }, + Value: objectToJSONBytes(t, testObj), + Object: metaAccessor, + } + + rv, err := backend.WriteEvent(ctx, addEvent) + require.NoError(t, err) + require.Greater(t, rv, int64(0)) + + // The event must be persisted even though the client cancelled right after + // the data was committed. + found := false + for ev, err := range backend.eventStore.ListSince(context.Background(), 0, SortOrderAsc) { + require.NoError(t, err) + if ev.Name == resourceName { + found = true + } + } + require.True(t, found, "event should be persisted after data save even if the client cancels") +} + func TestKvStorageBackend_WatchWriteEvents(t *testing.T) { for _, useChannel := range []bool{true, false} { backend := setupTestStorageBackend(t) diff --git a/pkg/storage/unified/sql/backend.go b/pkg/storage/unified/sql/backend.go index b913385bb5fab..561dc7f42688b 100644 --- a/pkg/storage/unified/sql/backend.go +++ b/pkg/storage/unified/sql/backend.go @@ -240,6 +240,8 @@ func NewStorageBackend( if cfg.EnableKVLeases { kvBackendOpts.EnableKVLeases = true kvBackendOpts.Holder = ResolveLeaseHolder(cfg) + kvBackendOpts.LeaseTTL = cfg.KVLeaseTTL + kvBackendOpts.LeaseAutoRenew = cfg.KVLeaseAutoRenew } return resource.NewKVStorageBackend(kvBackendOpts)