From 1263a3d364010e848a9cf4b4cda993f31c9072b6 Mon Sep 17 00:00:00 2001 From: Will Assis <35489495+gassiss@users.noreply.github.com> Date: Mon, 12 Jan 2026 12:17:41 -0500 Subject: [PATCH] unified-storage: HappyPath and notifier tests + couple of bugfixes (#116087) * unified-storage: couple of bugfixes and enable HappyPath and notifier sqlkv tests --- pkg/storage/unified/resource/notifier.go | 8 +-- pkg/storage/unified/resource/notifier_test.go | 22 +++----- .../unified/resource/storage_backend.go | 3 +- .../storage_backend_sql_compatibility.go | 54 +++++++++++++------ .../unified/testing/storage_backend_test.go | 32 +++++------ 5 files changed, 67 insertions(+), 52 deletions(-) diff --git a/pkg/storage/unified/resource/notifier.go b/pkg/storage/unified/resource/notifier.go index 5dd6a17ad29..3d3b2024d7e 100644 --- a/pkg/storage/unified/resource/notifier.go +++ b/pkg/storage/unified/resource/notifier.go @@ -78,13 +78,13 @@ func (n *notifier) Watch(ctx context.Context, opts watchOptions) <-chan Event { cache := gocache.New(cacheTTL, cacheCleanupInterval) events := make(chan Event, opts.BufferSize) - initialRV, err := n.lastEventResourceVersion(ctx) + lastRV, err := n.lastEventResourceVersion(ctx) if errors.Is(err, ErrNotFound) { - initialRV = snowflakeFromTime(time.Now()) // No events yet, start from the beginning + lastRV = 0 // No events yet, start from the beginning } else if err != nil { n.log.Error("Failed to get last event resource version", "error", err) } - lastRV := initialRV + 1 // We want to start watching from the next event + lastRV = lastRV + 1 // We want to start watching from the next event go func() { defer close(events) @@ -110,7 +110,7 @@ func (n *notifier) Watch(ctx context.Context, opts watchOptions) <-chan Event { } // Skip old events lower than the requested resource version - if evt.ResourceVersion <= initialRV { + if evt.ResourceVersion < lastRV { continue } diff --git a/pkg/storage/unified/resource/notifier_test.go b/pkg/storage/unified/resource/notifier_test.go index 060f8eecfbe..f78629ebeb7 100644 --- a/pkg/storage/unified/resource/notifier_test.go +++ b/pkg/storage/unified/resource/notifier_test.go @@ -25,7 +25,6 @@ func setupTestNotifier(t *testing.T) (*notifier, *eventStore) { return notifier, eventStore } -// nolint:unused func setupTestNotifierSqlKv(t *testing.T) (*notifier, *eventStore) { dbstore := db.InitTestDB(t) eDB, err := dbimpl.ProvideResourceDB(dbstore, setting.NewCfg(), nil) @@ -60,8 +59,7 @@ func runNotifierTestWith(t *testing.T, storeName string, newStoreFn func(*testin func TestNotifier_lastEventResourceVersion(t *testing.T) { runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierLastEventResourceVersion) - // enable this when sqlkv is ready - // runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierLastEventResourceVersion) + runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierLastEventResourceVersion) } func testNotifierLastEventResourceVersion(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) { @@ -112,8 +110,7 @@ func testNotifierLastEventResourceVersion(t *testing.T, ctx context.Context, not func TestNotifier_cachekey(t *testing.T) { runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierCachekey) - // enable this when sqlkv is ready - // runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierCachekey) + runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierCachekey) } func testNotifierCachekey(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) { @@ -167,8 +164,7 @@ func testNotifierCachekey(t *testing.T, ctx context.Context, notifier *notifier, func TestNotifier_Watch_NoEvents(t *testing.T) { runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierWatchNoEvents) - // enable this when sqlkv is ready - // runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchNoEvents) + runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchNoEvents) } func testNotifierWatchNoEvents(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) { @@ -209,8 +205,7 @@ func testNotifierWatchNoEvents(t *testing.T, ctx context.Context, notifier *noti func TestNotifier_Watch_WithExistingEvents(t *testing.T) { runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierWatchWithExistingEvents) - // enable this when sqlkv is ready - // runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchWithExistingEvents) + runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchWithExistingEvents) } func testNotifierWatchWithExistingEvents(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) { @@ -284,8 +279,7 @@ func testNotifierWatchWithExistingEvents(t *testing.T, ctx context.Context, noti func TestNotifier_Watch_EventDeduplication(t *testing.T) { runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierWatchEventDeduplication) - // enable this when sqlkv is ready - // runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchEventDeduplication) + runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchEventDeduplication) } func testNotifierWatchEventDeduplication(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) { @@ -351,8 +345,7 @@ func testNotifierWatchEventDeduplication(t *testing.T, ctx context.Context, noti func TestNotifier_Watch_ContextCancellation(t *testing.T) { runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierWatchContextCancellation) - // enable this when sqlkv is ready - // runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchContextCancellation) + runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchContextCancellation) } func testNotifierWatchContextCancellation(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) { @@ -398,8 +391,7 @@ func testNotifierWatchContextCancellation(t *testing.T, ctx context.Context, not func TestNotifier_Watch_MultipleEvents(t *testing.T) { runNotifierTestWith(t, "badger", setupTestNotifier, testNotifierWatchMultipleEvents) - // enable this when sqlkv is ready - // runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchMultipleEvents) + runNotifierTestWith(t, "sqlkv", setupTestNotifierSqlKv, testNotifierWatchMultipleEvents) } func testNotifierWatchMultipleEvents(t *testing.T, ctx context.Context, notifier *notifier, eventStore *eventStore) { diff --git a/pkg/storage/unified/resource/storage_backend.go b/pkg/storage/unified/resource/storage_backend.go index 55843905c72..4db6da89d9a 100644 --- a/pkg/storage/unified/resource/storage_backend.go +++ b/pkg/storage/unified/resource/storage_backend.go @@ -346,7 +346,8 @@ func (k *kvStorageBackend) WriteEvent(ctx context.Context, event WriteEvent) (in return 0, fmt.Errorf("failed to write data: %w", err) } - dataKey.ResourceVersion = rvmanager.SnowflakeFromRv(rv) + rv = rvmanager.SnowflakeFromRv(rv) + dataKey.ResourceVersion = rv } else { err := k.dataStore.Save(ctx, dataKey, bytes.NewReader(event.Value)) if err != nil { diff --git a/pkg/storage/unified/testing/storage_backend_sql_compatibility.go b/pkg/storage/unified/testing/storage_backend_sql_compatibility.go index 6584992f3cd..9066a39221c 100644 --- a/pkg/storage/unified/testing/storage_backend_sql_compatibility.go +++ b/pkg/storage/unified/testing/storage_backend_sql_compatibility.go @@ -9,7 +9,6 @@ import ( "testing" "time" - "github.com/bwmarrin/snowflake" "github.com/stretchr/testify/require" claims "github.com/grafana/authlib/types" @@ -187,13 +186,30 @@ func runKeyPathTest(t *testing.T, backend resource.StorageBackend, nsPrefix stri // verifyKeyPath is a helper function to verify key_path generation func verifyKeyPath(t *testing.T, db sqldb.DB, ctx context.Context, key *resourcepb.ResourceKey, action string, resourceVersion int64, expectedFolder string) { + // For SQL backend (namespace contains "-sql"), resourceVersion is in microsecond format + // but key_path stores snowflake RV, so convert to snowflake + // For KV backend (namespace contains "-kv"), resourceVersion is already in snowflake format + isSqlBackend := strings.Contains(key.Namespace, "-sql") + + var keyPathRV int64 + if isSqlBackend { + // Convert microsecond RV to snowflake for key_path construction + keyPathRV = rvmanager.SnowflakeFromRv(resourceVersion) + } else { + // KV backend already provides snowflake RV + keyPathRV = resourceVersion + } + + // Build the expected key_path using DataKey format: unified/data/group/resource/namespace/name/resourceVersion~action~folder + expectedKeyPath := fmt.Sprintf("unified/data/%s/%s/%s/%s/%d~%s~%s", key.Group, key.Resource, key.Namespace, key.Name, keyPathRV, action, expectedFolder) + var query string if db.DriverName() == "postgres" { - query = "SELECT key_path, resource_version, action, folder FROM resource_history WHERE namespace = $1 AND name = $2 AND resource_version = $3" + query = "SELECT key_path, resource_version, action, folder FROM resource_history WHERE key_path = $1" } else { - query = "SELECT key_path, resource_version, action, folder FROM resource_history WHERE namespace = ? AND name = ? AND resource_version = ?" + query = "SELECT key_path, resource_version, action, folder FROM resource_history WHERE key_path = ?" } - rows, err := db.QueryContext(ctx, query, key.Namespace, key.Name, resourceVersion) + rows, err := db.QueryContext(ctx, query, expectedKeyPath) require.NoError(t, err) require.True(t, rows.Next(), "Resource not found in resource_history table - both SQL and KV backends should write to this table") @@ -220,10 +236,6 @@ func verifyKeyPath(t *testing.T, db sqldb.DB, ctx context.Context, key *resource // Verify action suffix require.Contains(t, keyPath, fmt.Sprintf("~%s~", action)) - // Verify snowflake calculation - expectedSnowflake := (((resourceVersion / 1000) - snowflake.Epoch) << (snowflake.NodeBits + snowflake.StepBits)) + (resourceVersion % 1000) - require.Contains(t, keyPath, fmt.Sprintf("/%d~", expectedSnowflake), "actual RV: %d", actualRV) - // Verify folder if specified if expectedFolder != "" { require.Equal(t, expectedFolder, actualFolder) @@ -492,10 +504,10 @@ func verifyResourceHistoryRecord(t *testing.T, record ResourceHistoryRecord, exp } // Validate previous_resource_version - // For KV backend operations, resource versions are stored as snowflake format - // but expectedPrevRV is in microsecond format, so we need to use IsRvEqual for comparison + // For KV backend operations, expectedPrevRV is now in snowflake format (returned by KV backend) + // but resource_history table stores microsecond RV, so we need to use IsRvEqual for comparison if strings.Contains(record.Namespace, "-kv") { - require.True(t, rvmanager.IsRvEqual(record.PreviousResourceVersion, expectedPrevRV), + require.True(t, rvmanager.IsRvEqual(expectedPrevRV, record.PreviousResourceVersion), "Previous resource version should match (KV backend snowflake format)") } else { require.Equal(t, expectedPrevRV, record.PreviousResourceVersion) @@ -505,9 +517,10 @@ func verifyResourceHistoryRecord(t *testing.T, record ResourceHistoryRecord, exp require.Equal(t, expectedGeneration, record.Generation) // Validate resource_version - // For KV backend operations, resource versions are stored as snowflake format + // For KV backend operations, expectedRV is now in snowflake format (returned by KV backend) + // but resource_history table stores microsecond RV, so we need to use IsRvEqual for comparison if strings.Contains(record.Namespace, "-kv") { - require.True(t, rvmanager.IsRvEqual(record.ResourceVersion, expectedRV), + require.True(t, rvmanager.IsRvEqual(expectedRV, record.ResourceVersion), "Resource version should match (KV backend snowflake format)") } else { require.Equal(t, expectedRV, record.ResourceVersion) @@ -574,7 +587,7 @@ func verifyResourceTable(t *testing.T, db sqldb.DB, namespace string, resources // Resource version should match the expected version for test-resource-3 (updated version) expectedRV := resourceVersions[2][1] // test-resource-3's update version if strings.Contains(namespace, "-kv") { - require.True(t, rvmanager.IsRvEqual(record.ResourceVersion, expectedRV), + require.True(t, rvmanager.IsRvEqual(expectedRV, record.ResourceVersion), "Resource version should match (KV backend snowflake format)") } else { require.Equal(t, expectedRV, record.ResourceVersion) @@ -625,9 +638,16 @@ func verifyResourceVersionTable(t *testing.T, db sqldb.DB, namespace string, res // The resource_version table should contain the latest RV for the group+resource // It might be slightly higher due to RV manager operations, so check it's at least our max - require.GreaterOrEqual(t, record.ResourceVersion, maxRV, "resource_version should be at least the latest RV we tracked") - // But it shouldn't be too much higher (within a reasonable range) - require.LessOrEqual(t, record.ResourceVersion, maxRV+100, "resource_version shouldn't be much higher than expected") + // For KV backend, maxRV is in snowflake format but record.ResourceVersion is in microsecond format + // Use IsRvEqual for proper comparison between different RV formats + isKvBackend := strings.Contains(namespace, "-kv") + recordResourceVersion := record.ResourceVersion + if isKvBackend { + recordResourceVersion = rvmanager.SnowflakeFromRv(record.ResourceVersion) + } + + require.Less(t, recordResourceVersion, int64(9223372036854775807), "resource_version should be reasonable") + require.Greater(t, recordResourceVersion, maxRV, "resource_version should be at least the latest RV we tracked") } // runTestCrossBackendConsistency tests basic consistency between SQL and KV backends (lightweight) diff --git a/pkg/storage/unified/testing/storage_backend_test.go b/pkg/storage/unified/testing/storage_backend_test.go index 092cd476b52..3046967adee 100644 --- a/pkg/storage/unified/testing/storage_backend_test.go +++ b/pkg/storage/unified/testing/storage_backend_test.go @@ -38,7 +38,6 @@ func TestBadgerKVStorageBackend(t *testing.T) { func TestSQLKVStorageBackend(t *testing.T) { skipTests := map[string]bool{ - TestHappyPath: true, TestWatchWriteEvents: true, TestList: true, TestBlobSupport: true, @@ -51,21 +50,24 @@ func TestSQLKVStorageBackend(t *testing.T) { TestGetResourceLastImportTime: true, TestOptimisticLocking: true, } - // without RvManager - RunStorageBackendTest(t, func(ctx context.Context) resource.StorageBackend { - backend, _ := NewTestSqlKvBackend(t, ctx, false) - return backend - }, &TestOptions{ - NSPrefix: "sqlkvstorage-test", - SkipTests: skipTests, + + t.Run("Without RvManager", func(t *testing.T) { + RunStorageBackendTest(t, func(ctx context.Context) resource.StorageBackend { + backend, _ := NewTestSqlKvBackend(t, ctx, false) + return backend + }, &TestOptions{ + NSPrefix: "sqlkvstorage-test", + SkipTests: skipTests, + }) }) - // with RvManager - RunStorageBackendTest(t, func(ctx context.Context) resource.StorageBackend { - backend, _ := NewTestSqlKvBackend(t, ctx, true) - return backend - }, &TestOptions{ - NSPrefix: "sqlkvstorage-withrvmanager-test", - SkipTests: skipTests, + t.Run("With RvManager", func(t *testing.T) { + RunStorageBackendTest(t, func(ctx context.Context) resource.StorageBackend { + backend, _ := NewTestSqlKvBackend(t, ctx, true) + return backend + }, &TestOptions{ + NSPrefix: "sqlkvstorage-withrvmanager-test", + SkipTests: skipTests, + }) }) }