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

unified-storage: fix event persistence when sqlkv is enabled (#116033)

Browse Source
This commit is contained in:
Renato Costa
2026-01-09 07:58:58 -05:00
committed by GitHub
parent 12abbd5a15
commit ccdafc3fb2
2 changed files with 5 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/resource/sqlkv.go
+1 -1
View File
@@ -437,7 +437,7 @@ func (w *sqlWriteCloser) Close() error {
_, err = dbutil.Exec(w.ctx, tx, sqlKVInsertLegacyResourceHistory, sqlKVSaveRequest{
SQLTemplate: sqltemplate.New(w.kv.dialect),
sqlKVSectionKey: w.sectionKey,
sqlKVSectionKey: w.sectionKey, // unused: key_path is set by rvmanager
Value: value,
GUID: dataKey.GUID,
Group: dataKey.Group,
pkg/storage/unified/resource/storage_backend.go
+4 -20
View File
@@ -346,7 +346,7 @@ func (k *kvStorageBackend) WriteEvent(ctx context.Context, event WriteEvent) (in
return 0, fmt.Errorf("failed to write data: %w", err)
}
dataKey.ResourceVersion = rv
dataKey.ResourceVersion = rvmanager.SnowflakeFromRv(rv)
} else {
err := k.dataStore.Save(ctx, dataKey, bytes.NewReader(event.Value))
if err != nil {
@@ -372,22 +372,14 @@ func (k *kvStorageBackend) WriteEvent(ctx context.Context, event WriteEvent) (in
}
// Check if the RV we just wrote is the latest. If not, a concurrent write with higher RV happened
if !rvmanager.IsRvEqual(latestKey.ResourceVersion, rv) {
if latestKey.ResourceVersion != dataKey.ResourceVersion {
// Delete the data we just wrote since it's not the latest
// if we're running with rvManager, convert the ResourceVersion back to snowflake to delete
if k.rvManager != nil {
dataKey.ResourceVersion = rvmanager.SnowflakeFromRv(dataKey.ResourceVersion)
}
_ = k.dataStore.Delete(ctx, dataKey)
return 0, fmt.Errorf("optimistic locking failed: concurrent modification detected")
}
if !rvmanager.IsRvEqual(prevKey.ResourceVersion, event.PreviousRV) {
// Another concurrent write happened between our read and write
// if we're running with rvManager, convert the ResourceVersion back to snowflake to delete
if k.rvManager != nil {
dataKey.ResourceVersion = rvmanager.SnowflakeFromRv(dataKey.ResourceVersion)
}
_ = k.dataStore.Delete(ctx, dataKey)
return 0, fmt.Errorf("optimistic locking failed: resource was modified concurrently (expected previous RV %d, found %d)", event.PreviousRV, prevKey.ResourceVersion)
}
@@ -406,12 +398,8 @@ func (k *kvStorageBackend) WriteEvent(ctx context.Context, event WriteEvent) (in
}
// Check if the RV we just wrote is the latest. If not, a concurrent create with higher RV happened
if !rvmanager.IsRvEqual(latestKey.ResourceVersion, rv) {
if latestKey.ResourceVersion != dataKey.ResourceVersion {
// Delete the data we just wrote since it's not the latest
// if we're running with rvManager, convert the ResourceVersion back to snowflake to delete
if k.rvManager != nil {
dataKey.ResourceVersion = rvmanager.SnowflakeFromRv(dataKey.ResourceVersion)
}
_ = k.dataStore.Delete(ctx, dataKey)
return 0, fmt.Errorf("optimistic locking failed: concurrent create detected")
}
@@ -419,10 +407,6 @@ func (k *kvStorageBackend) WriteEvent(ctx context.Context, event WriteEvent) (in
// Verify that the immediate predecessor is not a create
if prevKey.Action == DataActionCreated {
// Another concurrent create happened - delete our write and return error
// if we're running with rvManager, convert the ResourceVersion back to snowflake to delete
if k.rvManager != nil {
dataKey.ResourceVersion = rvmanager.SnowflakeFromRv(dataKey.ResourceVersion)
}
_ = k.dataStore.Delete(ctx, dataKey)
return 0, fmt.Errorf("optimistic locking failed: concurrent create detected")
}
@@ -434,7 +418,7 @@ func (k *kvStorageBackend) WriteEvent(ctx context.Context, event WriteEvent) (in
Group: event.Key.Group,
Resource: event.Key.Resource,
Name: event.Key.Name,
ResourceVersion: rv,
ResourceVersion: dataKey.ResourceVersion,
Action: action,
Folder: obj.GetFolder(),
PreviousRV: event.PreviousRV,