diff --git a/pkg/storage/unified/sql/data/resource_history_update_rv.sql b/pkg/storage/unified/sql/data/resource_history_update_rv.sql index 767338ddb34..266227b5317 100644 --- a/pkg/storage/unified/sql/data/resource_history_update_rv.sql +++ b/pkg/storage/unified/sql/data/resource_history_update_rv.sql @@ -5,6 +5,25 @@ SET {{ .Ident "resource_version" }} = ( WHEN {{ $.Ident "guid" }} = {{ $.Arg $guid }} THEN CAST({{ $.Arg $rv }} AS {{ if eq $.DialectName "postgres" }}BIGINT{{ else }}SIGNED{{ end }}) {{ end }} END +), {{ .Ident "key_path" }} = ( + CASE + {{ range $guid, $snowflakeRv := .GUIDToSnowflakeRV }} + WHEN {{ $.Ident "guid" }} = {{ $.Arg $guid }} THEN CONCAT( + 'unified', {{ $.SlashFunc }}, 'data', {{ $.SlashFunc }}, + {{ $.Ident "group" }}, {{ $.SlashFunc }}, + {{ $.Ident "resource" }}, {{ $.SlashFunc }}, + {{ $.Ident "namespace" }}, {{ $.SlashFunc }}, + {{ $.Ident "name" }}, {{ $.SlashFunc }}, + CAST({{ $.Arg $snowflakeRv }} AS {{ if eq $.DialectName "postgres" }}BIGINT{{ else }}SIGNED{{ end }}), + {{ $.TildeFunc }}, + CASE {{ $.Ident "action" }} + WHEN 1 THEN 'created' + WHEN 2 THEN 'updated' + WHEN 3 THEN 'deleted' + END, {{ $.TildeFunc }}, + COALESCE({{ $.Ident "folder" }}, '')) + {{ end }} + END ) WHERE {{ .Ident "guid" }} IN ( {{$first := true}} diff --git a/pkg/storage/unified/sql/queries.go b/pkg/storage/unified/sql/queries.go index e51cf943041..5b9a177e17c 100644 --- a/pkg/storage/unified/sql/queries.go +++ b/pkg/storage/unified/sql/queries.go @@ -369,13 +369,30 @@ func (r sqlResourceBlobQueryRequest) Validate() error { type sqlResourceUpdateRVRequest struct { sqltemplate.SQLTemplate - GUIDToRV map[string]int64 + GUIDToRV map[string]int64 + GUIDToSnowflakeRV map[string]int64 } func (r sqlResourceUpdateRVRequest) Validate() error { return nil // TODO } +func (r sqlResourceUpdateRVRequest) SlashFunc() string { + if r.DialectName() == "postgres" { + return "CHR(47)" + } + + return "CHAR(47)" +} + +func (r sqlResourceUpdateRVRequest) TildeFunc() string { + if r.DialectName() == "postgres" { + return "CHR(126)" + } + + return "CHAR(126)" +} + // resource_version table requests. type resourceVersionResponse struct { ResourceVersion int64 diff --git a/pkg/storage/unified/sql/rv_manager.go b/pkg/storage/unified/sql/rv_manager.go index 1232aa9c700..858345b1fc2 100644 --- a/pkg/storage/unified/sql/rv_manager.go +++ b/pkg/storage/unified/sql/rv_manager.go @@ -8,6 +8,7 @@ import ( "sync" "time" + "github.com/bwmarrin/snowflake" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" "go.opentelemetry.io/otel/attribute" @@ -240,6 +241,7 @@ func (m *resourceVersionManager) execBatch(ctx context.Context, group, resource defer cancel() guidToRV := make(map[string]int64, len(batch)) + guidToSnowflakeRV := make(map[string]int64, len(batch)) guids := make([]string, len(batch)) // The GUIDs of the created resources in the same order as the batch rvs := make([]int64, len(batch)) // The RVs of the created resources in the same order as the batch @@ -285,6 +287,7 @@ func (m *resourceVersionManager) execBatch(ctx context.Context, group, resource // Allocate the RVs for i, guid := range guids { guidToRV[guid] = rv + guidToSnowflakeRV[guid] = snowflakeFromRv(rv) rvs[i] = rv rv++ } @@ -301,8 +304,9 @@ func (m *resourceVersionManager) execBatch(ctx context.Context, group, resource span.AddEvent("resource_versions_updated") if _, err := dbutil.Exec(ctx, tx, sqlResourceHistoryUpdateRV, sqlResourceUpdateRVRequest{ - SQLTemplate: sqltemplate.New(m.dialect), - GUIDToRV: guidToRV, + SQLTemplate: sqltemplate.New(m.dialect), + GUIDToRV: guidToRV, + GUIDToSnowflakeRV: guidToSnowflakeRV, }); err != nil { span.AddEvent("resource_history_update_rv_failed", trace.WithAttributes( attribute.String("error", err.Error()), @@ -340,6 +344,12 @@ func (m *resourceVersionManager) execBatch(ctx context.Context, group, resource } } +// takes a unix microsecond rv and transforms into a snowflake format. The timestamp is converted from microsecond to +// millisecond (the integer division) and the remainder is saved in the stepbits section. machine id is always 0 +func snowflakeFromRv(rv int64) int64 { + return (((rv / 1000) - snowflake.Epoch) << (snowflake.NodeBits + snowflake.StepBits)) + (rv % 1000) +} + // lock locks the resource version for the given key func (m *resourceVersionManager) lock(ctx context.Context, x db.ContextExecer, group, resource string) (nextRV int64, err error) { // 1. Lock the row and prevent concurrent updates until the transaction is committed diff --git a/pkg/storage/unified/sql/test/benchmark_test.go b/pkg/storage/unified/sql/test/benchmark_test.go index f9665c83b7d..8bf65cbd6f7 100644 --- a/pkg/storage/unified/sql/test/benchmark_test.go +++ b/pkg/storage/unified/sql/test/benchmark_test.go @@ -15,5 +15,6 @@ func TestIntegrationBenchmarkSQLStorageBackend(t *testing.T) { if db.IsTestDbSQLite() { opts.Concurrency = 1 // to avoid SQLite database is locked error } - test.BenchmarkStorageBackend(t, newTestBackend(t, true, 2*time.Millisecond), opts) + backend, _ := newTestBackend(t, true, 2*time.Millisecond) + test.BenchmarkStorageBackend(t, backend, opts) } diff --git a/pkg/storage/unified/sql/test/integration_test.go b/pkg/storage/unified/sql/test/integration_test.go index cf45ee64d43..eaf78de0779 100644 --- a/pkg/storage/unified/sql/test/integration_test.go +++ b/pkg/storage/unified/sql/test/integration_test.go @@ -24,6 +24,7 @@ import ( "github.com/grafana/grafana/pkg/storage/unified/resourcepb" "github.com/grafana/grafana/pkg/storage/unified/search" "github.com/grafana/grafana/pkg/storage/unified/sql" + sqldb "github.com/grafana/grafana/pkg/storage/unified/sql/db" "github.com/grafana/grafana/pkg/storage/unified/sql/db/dbimpl" unitest "github.com/grafana/grafana/pkg/storage/unified/testing" "github.com/grafana/grafana/pkg/tests/testsuite" @@ -38,7 +39,7 @@ var initMutex = &sync.Mutex{} // newTestBackend creates a fresh database and backend for a test. // It uses a mutex to ensure the entire initialization and migration // process is atomic and does not race with other parallel tests. -func newTestBackend(t *testing.T, isHA bool, simulatedNetworkLatency time.Duration) resource.StorageBackend { +func newTestBackend(t *testing.T, isHA bool, simulatedNetworkLatency time.Duration) (resource.StorageBackend, sqldb.DB) { // Lock to ensure the entire init block is atomic. initMutex.Lock() // Unlock once the function returns the initialized backend. @@ -61,7 +62,11 @@ func newTestBackend(t *testing.T, isHA bool, simulatedNetworkLatency time.Durati // Use a context with a reasonable timeout for migrations. err = backend.Init(testutil.NewTestContext(t, time.Now().Add(1*time.Minute))) require.NoError(t, err) - return backend + + sqlDB, err := eDB.Init(testutil.NewTestContext(t, time.Now().Add(1*time.Minute))) + require.NoError(t, err) + + return backend, sqlDB } func TestMain(m *testing.M) { @@ -73,7 +78,8 @@ func TestIntegrationStorageServer(t *testing.T) { t.Cleanup(db.CleanupTestDB) unitest.RunStorageServerTest(t, func(ctx context.Context) resource.StorageBackend { - return newTestBackend(t, true, 0) + backend, _ := newTestBackend(t, true, 0) + return backend }) } @@ -84,12 +90,31 @@ func TestIntegrationSQLStorageBackend(t *testing.T) { t.Run("IsHA (polling notifier)", func(t *testing.T) { unitest.RunStorageBackendTest(t, func(ctx context.Context) resource.StorageBackend { - return newTestBackend(t, true, 0) + backend, _ := newTestBackend(t, true, 0) + return backend }, nil) }) t.Run("NotHA (in process notifier)", func(t *testing.T) { unitest.RunStorageBackendTest(t, func(ctx context.Context) resource.StorageBackend { + backend, _ := newTestBackend(t, false, 0) + return backend + }, nil) + }) +} + +func TestIntegrationSQLStorageAndSQLKVCompatibilityTests(t *testing.T) { + testutil.SkipIntegrationTestInShortMode(t) + t.Cleanup(db.CleanupTestDB) + + t.Run("IsHA (polling notifier)", func(t *testing.T) { + unitest.RunSQLStorageBackendCompatibilityTest(t, func(ctx context.Context) (resource.StorageBackend, sqldb.DB) { + return newTestBackend(t, true, 0) + }, nil) + }) + + t.Run("NotHA (in process notifier)", func(t *testing.T) { + unitest.RunSQLStorageBackendCompatibilityTest(t, func(ctx context.Context) (resource.StorageBackend, sqldb.DB) { return newTestBackend(t, false, 0) }, nil) }) @@ -110,7 +135,7 @@ func TestIntegrationSearchAndStorage(t *testing.T) { t.Cleanup(search.Stop) // Create a new resource backend - storage := newTestBackend(t, false, 0) + storage, _ := newTestBackend(t, false, 0) require.NotNil(t, storage) // Run the shared storage and search tests diff --git a/pkg/storage/unified/sql/testdata/mysql--resource_history_update_rv-single path.sql b/pkg/storage/unified/sql/testdata/mysql--resource_history_update_rv-single path.sql index 6b07ff5e55b..62bb64c8300 100755 --- a/pkg/storage/unified/sql/testdata/mysql--resource_history_update_rv-single path.sql +++ b/pkg/storage/unified/sql/testdata/mysql--resource_history_update_rv-single path.sql @@ -4,6 +4,9 @@ SET `resource_version` = ( WHEN `guid` = 'guid1' THEN CAST(123 AS SIGNED) WHEN `guid` = 'guid2' THEN CAST(456 AS SIGNED) END +), `key_path` = ( + CASE + END ) WHERE `guid` IN ( 'guid1', 'guid2' diff --git a/pkg/storage/unified/sql/testdata/postgres--resource_history_update_rv-single path.sql b/pkg/storage/unified/sql/testdata/postgres--resource_history_update_rv-single path.sql index 85352e9ead4..6529fb1e4d5 100755 --- a/pkg/storage/unified/sql/testdata/postgres--resource_history_update_rv-single path.sql +++ b/pkg/storage/unified/sql/testdata/postgres--resource_history_update_rv-single path.sql @@ -4,6 +4,9 @@ SET "resource_version" = ( WHEN "guid" = 'guid1' THEN CAST(123 AS BIGINT) WHEN "guid" = 'guid2' THEN CAST(456 AS BIGINT) END +), "key_path" = ( + CASE + END ) WHERE "guid" IN ( 'guid1', 'guid2' diff --git a/pkg/storage/unified/sql/testdata/sqlite--resource_history_update_rv-single path.sql b/pkg/storage/unified/sql/testdata/sqlite--resource_history_update_rv-single path.sql index 9fb23fb3956..707fbad6cb8 100755 --- a/pkg/storage/unified/sql/testdata/sqlite--resource_history_update_rv-single path.sql +++ b/pkg/storage/unified/sql/testdata/sqlite--resource_history_update_rv-single path.sql @@ -4,6 +4,9 @@ SET "resource_version" = ( WHEN "guid" = 'guid1' THEN CAST(123 AS SIGNED) WHEN "guid" = 'guid2' THEN CAST(456 AS SIGNED) END +), "key_path" = ( + CASE + END ) WHERE "guid" IN ( 'guid1', 'guid2' diff --git a/pkg/storage/unified/testing/storage_backend.go b/pkg/storage/unified/testing/storage_backend.go index ce36d8836bc..730efe418f6 100644 --- a/pkg/storage/unified/testing/storage_backend.go +++ b/pkg/storage/unified/testing/storage_backend.go @@ -11,6 +11,7 @@ import ( "testing" "time" + "github.com/bwmarrin/snowflake" "github.com/go-jose/go-jose/v4/jwt" "github.com/google/uuid" "github.com/stretchr/testify/assert" @@ -25,6 +26,7 @@ import ( "github.com/grafana/grafana/pkg/apimachinery/utils" "github.com/grafana/grafana/pkg/storage/unified/resource" "github.com/grafana/grafana/pkg/storage/unified/resourcepb" + sqldb "github.com/grafana/grafana/pkg/storage/unified/sql/db" "github.com/grafana/grafana/pkg/util/testutil" ) @@ -42,10 +44,14 @@ const ( TestCreateNewResource = "create new resource" TestGetResourceLastImportTime = "get resource last import time" TestOptimisticLocking = "optimistic locking on concurrent writes" + TestKeyPathGeneration = "key_path generation" ) type NewBackendFunc func(ctx context.Context) resource.StorageBackend +// NewBackendWithDBFunc creates a backend with database access for testing +type NewBackendWithDBFunc func(ctx context.Context) (resource.StorageBackend, sqldb.DB) + // TestOptions configures which tests to run type TestOptions struct { SkipTests map[string]bool // tests to skip @@ -100,6 +106,37 @@ func RunStorageBackendTest(t *testing.T, newBackend NewBackendFunc, opts *TestOp } } +func RunSQLStorageBackendCompatibilityTest(t *testing.T, newBackend NewBackendWithDBFunc, opts *TestOptions) { + if opts == nil { + opts = &TestOptions{} + } + + if opts.NSPrefix == "" { + opts.NSPrefix = GenerateRandomNSPrefix() + } + + t.Logf("Running tests with namespace prefix: %s", opts.NSPrefix) + + cases := []struct { + name string + fn func(*testing.T, resource.StorageBackend, string, sqldb.DB) + }{ + {TestKeyPathGeneration, runTestIntegrationBackendKeyPathGeneration}, + } + + for _, tc := range cases { + if shouldSkip := opts.SkipTests[tc.name]; shouldSkip { + t.Logf("Skipping test: %s", tc.name) + continue + } + + t.Run(tc.name, func(t *testing.T) { + backend, db := newBackend(context.Background()) + tc.fn(t, backend, opts.NSPrefix, db) + }) + } +} + func runTestIntegrationBackendHappyPath(t *testing.T, backend resource.StorageBackend, nsPrefix string) { ctx := types.WithAuthInfo(context.Background(), authn.NewAccessTokenAuthInfo(authn.Claims[authn.AccessTokenClaims]{ Claims: jwt.Claims{ @@ -1722,3 +1759,222 @@ func runTestIntegrationBackendOptimisticLocking(t *testing.T, backend resource.S require.LessOrEqual(t, successes, 1, "at most one create should succeed (errors: %v)", errorMessages) }) } + +func runTestIntegrationBackendKeyPathGeneration(t *testing.T, backend resource.StorageBackend, nsPrefix string, db sqldb.DB) { + ctx := testutil.NewDefaultTestContext(t) + + t.Run("Create resource", func(t *testing.T) { + // Create a test resource + key := &resourcepb.ResourceKey{ + Group: "playlist.grafana.app", + Resource: "playlists", + Namespace: nsPrefix + "-default", + Name: "test-playlist-crud", + } + + // Create the K8s unstructured object + testObj := &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": "playlist.grafana.app/v0alpha1", + "kind": "Playlist", + "metadata": map[string]interface{}{ + "name": "test-playlist-crud", + "namespace": nsPrefix + "-default", + "uid": "test-uid-crud-123", + }, + "spec": map[string]interface{}{ + "title": "My Test Playlist", + }, + }, + } + + // Get metadata accessor + metaAccessor, err := utils.MetaAccessor(testObj) + require.NoError(t, err) + + // Serialize to JSON + jsonBytes, err := testObj.MarshalJSON() + require.NoError(t, err) + + // Create WriteEvent + writeEvent := resource.WriteEvent{ + Type: resourcepb.WatchEvent_ADDED, + Key: key, + Value: jsonBytes, + Object: metaAccessor, + PreviousRV: 0, // Always 0 for new resources + GUID: "create-guid-crud-123", + } + + // Create the resource using WriteEvent + createRV, err := backend.WriteEvent(ctx, writeEvent) + require.NoError(t, err) + require.Greater(t, createRV, int64(0)) + + // Verify created resource key_path + verifyKeyPath(t, db, ctx, key, "created", createRV, "") + + t.Run("Update resource", func(t *testing.T) { + // Update the resource + testObj.Object["spec"] = map[string]interface{}{ + "title": "My Updated Playlist", + } + + updatedMetaAccessor, err := utils.MetaAccessor(testObj) + require.NoError(t, err) + + updatedJsonBytes, err := testObj.MarshalJSON() + require.NoError(t, err) + + updateEvent := resource.WriteEvent{ + Type: resourcepb.WatchEvent_MODIFIED, + Key: key, + Value: updatedJsonBytes, + Object: updatedMetaAccessor, + PreviousRV: createRV, + GUID: fmt.Sprintf("update-guid-%d", createRV), + } + + // Update the resource + updateRV, err := backend.WriteEvent(ctx, updateEvent) + require.NoError(t, err) + require.Greater(t, updateRV, createRV) + + // Verify updated resource key_path + verifyKeyPath(t, db, ctx, key, "updated", updateRV, "") + + t.Run("Delete resource", func(t *testing.T) { + deleteEvent := resource.WriteEvent{ + Type: resourcepb.WatchEvent_DELETED, + Key: key, + Value: updatedJsonBytes, // Keep the last known value + Object: updatedMetaAccessor, + PreviousRV: updateRV, + GUID: fmt.Sprintf("delete-guid-%d", updateRV), + } + + // Delete the resource + deleteRV, err := backend.WriteEvent(ctx, deleteEvent) + require.NoError(t, err) + require.Greater(t, deleteRV, updateRV) + + // Verify deleted resource key_path + verifyKeyPath(t, db, ctx, key, "deleted", deleteRV, "") + }) + }) + }) + + t.Run("Resource with folder", func(t *testing.T) { + // Create a resource in a folder + folderKey := &resourcepb.ResourceKey{ + Group: "dashboard.grafana.app", + Resource: "dashboards", + Namespace: nsPrefix + "-default", + Name: "my-dashboard", + } + + // Create dashboard object with folder + dashboardObj := &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": "dashboard.grafana.app/v0alpha1", + "kind": "Dashboard", + "metadata": map[string]interface{}{ + "name": "my-dashboard", + "namespace": nsPrefix + "-default", + "uid": "dash-uid-456", + "annotations": map[string]interface{}{ + "grafana.app/folder": "test-folder", + }, + }, + "spec": map[string]interface{}{ + "title": "My Dashboard", + }, + }, + } + + folderMetaAccessor, err := utils.MetaAccessor(dashboardObj) + require.NoError(t, err) + + folderJsonBytes, err := dashboardObj.MarshalJSON() + require.NoError(t, err) + + folderWriteEvent := resource.WriteEvent{ + Type: resourcepb.WatchEvent_ADDED, + Key: folderKey, + Value: folderJsonBytes, + Object: folderMetaAccessor, + PreviousRV: 0, + GUID: "folder-guid-456", + } + + // Create the dashboard in folder + folderRV, err := backend.WriteEvent(ctx, folderWriteEvent) + require.NoError(t, err) + require.Greater(t, folderRV, int64(0)) + + // Verify folder resource key_path includes folder + verifyKeyPath(t, db, ctx, folderKey, "created", folderRV, "test-folder") + }) +} + +// 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) { + 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" + } else { + query = "SELECT key_path, resource_version, action, folder FROM resource_history WHERE namespace = ? AND name = ? AND resource_version = ?" + } + rows, err := db.QueryContext(ctx, query, key.Namespace, key.Name, resourceVersion) + require.NoError(t, err) + + require.True(t, rows.Next()) + + var keyPath string + var actualRV int64 + var actualAction int + var actualFolder string + + err = rows.Scan(&keyPath, &actualRV, &actualAction, &actualFolder) + require.NoError(t, err) + err = rows.Close() + require.NoError(t, err) + + // Verify basic key_path format + require.Contains(t, keyPath, "unified/data/") + require.Contains(t, keyPath, key.Group) + require.Contains(t, keyPath, key.Resource) + require.Contains(t, keyPath, key.Namespace) + require.Contains(t, keyPath, key.Name) + + // 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), fmt.Sprintf("actual RV: %d", actualRV)) + + // Verify folder if specified + if expectedFolder != "" { + require.Equal(t, expectedFolder, actualFolder) + require.Contains(t, keyPath, expectedFolder) + } + + // Verify action code matches + var expectedActionCode int + switch action { + case "created": + expectedActionCode = 1 + case "updated": + expectedActionCode = 2 + case "deleted": + expectedActionCode = 3 + } + require.Equal(t, expectedActionCode, actualAction) + + t.Logf("Action: %s, RV: %d, Snowflake: %d", action, resourceVersion, expectedSnowflake) + t.Logf("Key_path: %s", keyPath) + if expectedFolder != "" { + t.Logf("Folder: %s", actualFolder) + } +}