public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

unified-storage: make sql backend update key_path for kv store (#114879)

Browse Source 
* unified-storage: update resource_history_update_rv.sql to populate key_path in resource_history
This commit is contained in:
Will Assis
2025-12-10 07:06:06 -05:00
committed by GitHub
parent 532a2e5f4d
commit 755b479be4
9 changed files with 346 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/sql/data/resource_history_update_rv.sql
+19
View File
@@ -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}}
pkg/storage/unified/sql/queries.go
+18 -1
View File
@@ -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
pkg/storage/unified/sql/rv_manager.go
+12 -2
View File
@@ -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
pkg/storage/unified/sql/test/benchmark_test.go
+2 -1
View File
@@ -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)
}
pkg/storage/unified/sql/test/integration_test.go
+30 -5
View File
@@ -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
pkg/storage/unified/sql/testdata/mysql--resource_history_update_rv-single path.sql
Vendored
+3
View File
@@ -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'
pkg/storage/unified/sql/testdata/postgres--resource_history_update_rv-single path.sql
Vendored
+3
View File
@@ -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'
pkg/storage/unified/sql/testdata/sqlite--resource_history_update_rv-single path.sql
Vendored
+3
View File
@@ -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'
pkg/storage/unified/testing/storage_backend.go
+256
View File
@@ -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)
}
}