From 20fe9b455da1cf6a6f40bf24f85675a81c3fe1bf Mon Sep 17 00:00:00 2001 From: Will Assis Date: Tue, 2 Dec 2025 14:37:30 -0300 Subject: [PATCH] convert RVs to snowflake --- pkg/storage/unified/resource/bulk.go | 53 ++++++++++++++----- .../unified/resource/storage_backend.go | 4 +- 2 files changed, 41 insertions(+), 16 deletions(-) diff --git a/pkg/storage/unified/resource/bulk.go b/pkg/storage/unified/resource/bulk.go index 98a049e5fd1..889a265b7be 100644 --- a/pkg/storage/unified/resource/bulk.go +++ b/pkg/storage/unified/resource/bulk.go @@ -15,6 +15,7 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/bwmarrin/snowflake" authlib "github.com/grafana/authlib/types" "github.com/grafana/grafana/pkg/apimachinery/utils" @@ -333,35 +334,59 @@ func (b *batchRunner) RollbackRequested() bool { return false } -type bulkRV struct { - max int64 - counter int64 +// getDummySnowflake returns a snowflake with machine and counter bits set to 0 +func getDummySnowflake(ts time.Time) int64 { + return (ts.UnixMilli() - snowflake.Epoch) << 22 } -// Used when executing a bulk import where we can fake the RV values -func NewBulkRV() *bulkRV { - t := time.Now().Truncate(time.Second * 10) +type bulkRV struct { + max int64 + counters map[int64]int64 +} + +// Used when executing a bulk import so that we can generate snowflake RVs in the past +func newBulkRV() *bulkRV { + t := getDummySnowflake(time.Now()) return &bulkRV{ - max: (t.UnixMicro() / 10000000) * 10000000, - counter: 0, + max: t, + counters: make(map[int64]int64), } } -func (x *bulkRV) Next(obj metav1.Object) int64 { - ts := obj.GetCreationTimestamp().UnixMicro() +func (x *bulkRV) next(obj metav1.Object) int64 { + ts := getDummySnowflake(obj.GetCreationTimestamp().Time) anno := obj.GetAnnotations() if anno != nil { v := anno[utils.AnnoKeyUpdatedTimestamp] t, err := time.Parse(time.RFC3339, v) if err == nil { - ts = t.UnixMicro() + ts = getDummySnowflake(t) } } - if ts > x.max || ts < 10000000 { + if ts > x.max || ts < 0 { ts = x.max } - x.counter++ - return (ts/10000000)*10000000 + x.counter + + counter := x.counters[ts] + counter++ + + if counter > 65535 { + for { + ts += (1 << 22) // Add 1ms in snowflake format + if x.counters[ts] < 65535 { + break + } + } + + counter = x.counters[ts] + 1 + + if ts > x.max { + x.max = ts + } + } + + x.counters[ts] = counter + return ts + counter } type BulkLock struct { diff --git a/pkg/storage/unified/resource/storage_backend.go b/pkg/storage/unified/resource/storage_backend.go index ead0c2245ed..244afa0d789 100644 --- a/pkg/storage/unified/resource/storage_backend.go +++ b/pkg/storage/unified/resource/storage_backend.go @@ -1161,7 +1161,7 @@ func (b *kvStorageBackend) ProcessBulk(ctx context.Context, setting BulkSettings } defer b.bulkLock.Finish(setting.Collection) - bulkRvGenerator := NewBulkRV() + bulkRvGenerator := newBulkRV() summaries := make(map[string]*resourcepb.BulkResponse_Summary, len(setting.Collection)) rsp := &resourcepb.BulkResponse{} @@ -1310,7 +1310,7 @@ func (b *kvStorageBackend) ProcessBulk(ctx context.Context, setting BulkSettings Resource: req.Key.Resource, Namespace: req.Key.Namespace, Name: req.Key.Name, - ResourceVersion: bulkRvGenerator.Next(obj), + ResourceVersion: bulkRvGenerator.next(obj), Action: action, Folder: req.Folder, }