From 821c5531323e529019952237082115d1969bbdcf Mon Sep 17 00:00:00 2001 From: Will Assis Date: Tue, 11 Nov 2025 17:14:22 -0300 Subject: [PATCH] move bulkLock and bulkRv logic from sql into resource package --- pkg/storage/unified/resource/bulk.go | 84 ++++++++++++++++++++ pkg/storage/unified/sql/backend.go | 4 +- pkg/storage/unified/sql/bulk.go | 83 +------------------ pkg/storage/unified/sql/bulk_test.go | 9 ++- pkg/storage/unified/sql/notifier_sql.go | 4 +- pkg/storage/unified/sql/notifier_sql_test.go | 27 ++++--- 6 files changed, 109 insertions(+), 102 deletions(-) diff --git a/pkg/storage/unified/resource/bulk.go b/pkg/storage/unified/resource/bulk.go index 1ac0ff2032f..023d8d46bdf 100644 --- a/pkg/storage/unified/resource/bulk.go +++ b/pkg/storage/unified/resource/bulk.go @@ -6,10 +6,14 @@ import ( "fmt" "io" "net/http" + "sync" + "time" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" "google.golang.org/grpc/metadata" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" authlib "github.com/grafana/authlib/types" @@ -328,3 +332,83 @@ func (b *batchRunner) RollbackRequested() bool { } return false } + +type bulkRV struct { + max int64 + counter int64 +} + +// When executing a bulk import we can fake the RV values +func NewBulkRV() *bulkRV { + t := time.Now().Truncate(time.Second * 10) + return &bulkRV{ + max: (t.UnixMicro() / 10000000) * 10000000, + counter: 0, + } +} + +func (x *bulkRV) Next(obj metav1.Object) int64 { + ts := obj.GetCreationTimestamp().UnixMicro() + anno := obj.GetAnnotations() + if anno != nil { + v := anno[utils.AnnoKeyUpdatedTimestamp] + t, err := time.Parse(time.RFC3339, v) + if err == nil { + ts = t.UnixMicro() + } + } + if ts > x.max || ts < 10000000 { + ts = x.max + } + x.counter++ + return (ts/10000000)*10000000 + x.counter +} + +type BulkLock struct { + running map[string]bool + mu sync.Mutex +} + +func NewBulkLock() *BulkLock { + return &BulkLock{ + running: make(map[string]bool), + } +} + +func (x *BulkLock) Start(keys []*resourcepb.ResourceKey) error { + x.mu.Lock() + defer x.mu.Unlock() + + // First verify that it is not already running + ids := make([]string, len(keys)) + for i, k := range keys { + id := NSGR(k) + if x.running[id] { + return &apierrors.StatusError{ErrStatus: metav1.Status{ + Code: http.StatusPreconditionFailed, + Message: "bulk export is already running", + }} + } + ids[i] = id + } + + // Then add the keys to the lock + for _, k := range ids { + x.running[k] = true + } + return nil +} + +func (x *BulkLock) Finish(keys []*resourcepb.ResourceKey) { + x.mu.Lock() + defer x.mu.Unlock() + for _, k := range keys { + delete(x.running, NSGR(k)) + } +} + +func (x *BulkLock) Active() bool { + x.mu.Lock() + defer x.mu.Unlock() + return len(x.running) > 0 +} diff --git a/pkg/storage/unified/sql/backend.go b/pkg/storage/unified/sql/backend.go index ac7e12a51d0..c05dada4394 100644 --- a/pkg/storage/unified/sql/backend.go +++ b/pkg/storage/unified/sql/backend.go @@ -99,7 +99,7 @@ func NewBackend(opts BackendOptions) (Backend, error) { pollingInterval: opts.PollingInterval, watchBufferSize: opts.WatchBufferSize, storageMetrics: opts.storageMetrics, - bulkLock: &bulkLock{running: make(map[string]bool)}, + bulkLock: resource.NewBulkLock(), simulatedNetworkLatency: opts.SimulatedNetworkLatency, withPruner: opts.withPruner, lastImportTimeMaxAge: opts.LastImportTimeMaxAge, @@ -126,7 +126,7 @@ type backend struct { dbProvider db.DBProvider db db.DB dialect sqltemplate.Dialect - bulkLock *bulkLock + bulkLock *resource.BulkLock // watch streaming //stream chan *resource.WatchEvent diff --git a/pkg/storage/unified/sql/bulk.go b/pkg/storage/unified/sql/bulk.go index 7d4d7165d50..105ddd3dfe6 100644 --- a/pkg/storage/unified/sql/bulk.go +++ b/pkg/storage/unified/sql/bulk.go @@ -3,19 +3,14 @@ package sql import ( "context" "fmt" - "net/http" "os" - "sync" "time" "github.com/google/uuid" - apierrors "k8s.io/apimachinery/pkg/api/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "github.com/grafana/grafana-app-sdk/logging" - "github.com/grafana/grafana/pkg/apimachinery/utils" "github.com/grafana/grafana/pkg/storage/unified/parquet" "github.com/grafana/grafana/pkg/storage/unified/resource" "github.com/grafana/grafana/pkg/storage/unified/resourcepb" @@ -28,80 +23,6 @@ var ( _ resource.BulkProcessingBackend = (*backend)(nil) ) -type bulkRV struct { - max int64 - counter int64 -} - -// When executing a bulk import we can fake the RV values -func newBulkRV() *bulkRV { - t := time.Now().Truncate(time.Second * 10) - return &bulkRV{ - max: (t.UnixMicro() / 10000000) * 10000000, - counter: 0, - } -} - -func (x *bulkRV) next(obj metav1.Object) int64 { - ts := obj.GetCreationTimestamp().UnixMicro() - anno := obj.GetAnnotations() - if anno != nil { - v := anno[utils.AnnoKeyUpdatedTimestamp] - t, err := time.Parse(time.RFC3339, v) - if err == nil { - ts = t.UnixMicro() - } - } - if ts > x.max || ts < 10000000 { - ts = x.max - } - x.counter++ - return (ts/10000000)*10000000 + x.counter -} - -type bulkLock struct { - running map[string]bool - mu sync.Mutex -} - -func (x *bulkLock) Start(keys []*resourcepb.ResourceKey) error { - x.mu.Lock() - defer x.mu.Unlock() - - // First verify that it is not already running - ids := make([]string, len(keys)) - for i, k := range keys { - id := resource.NSGR(k) - if x.running[id] { - return &apierrors.StatusError{ErrStatus: metav1.Status{ - Code: http.StatusPreconditionFailed, - Message: "bulk export is already running", - }} - } - ids[i] = id - } - - // Then add the keys to the lock - for _, k := range ids { - x.running[k] = true - } - return nil -} - -func (x *bulkLock) Finish(keys []*resourcepb.ResourceKey) { - x.mu.Lock() - defer x.mu.Unlock() - for _, k := range keys { - delete(x.running, resource.NSGR(k)) - } -} - -func (x *bulkLock) Active() bool { - x.mu.Lock() - defer x.mu.Unlock() - return len(x.running) > 0 -} - func (b *backend) ProcessBulk(ctx context.Context, setting resource.BulkSettings, iter resource.BulkRequestIterator) *resourcepb.BulkResponse { err := b.bulkLock.Start(setting.Collection) if err != nil { @@ -168,7 +89,7 @@ func (b *backend) processBulk(ctx context.Context, setting resource.BulkSettings } // Calculate the RV based on incoming request timestamps - rv := newBulkRV() + rv := resource.NewBulkRV() summaries := make(map[string]*resourcepb.BulkResponse_Summary, len(setting.Collection)) @@ -235,7 +156,7 @@ func (b *backend) processBulk(ctx context.Context, setting resource.BulkSettings }, Folder: req.Folder, GUID: uuid.New().String(), - ResourceVersion: rv.next(obj), + ResourceVersion: rv.Next(obj), }); err != nil { return rollbackWithError(fmt.Errorf("insert into resource history: %w", err)) } diff --git a/pkg/storage/unified/sql/bulk_test.go b/pkg/storage/unified/sql/bulk_test.go index e1e00fb7839..4d8738d479f 100644 --- a/pkg/storage/unified/sql/bulk_test.go +++ b/pkg/storage/unified/sql/bulk_test.go @@ -3,6 +3,7 @@ package sql import ( "testing" + "github.com/grafana/grafana/pkg/storage/unified/resource" "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" ) @@ -13,10 +14,10 @@ func TestBatch(t *testing.T) { t.Run("rv iterator", func(t *testing.T) { t.Parallel() - rv := newBulkRV() - v0 := rv.next(&unstructured.Unstructured{}) - v1 := rv.next(&unstructured.Unstructured{}) - v2 := rv.next(&unstructured.Unstructured{}) + rv := resource.NewBulkRV() + v0 := rv.Next(&unstructured.Unstructured{}) + v1 := rv.Next(&unstructured.Unstructured{}) + v2 := rv.Next(&unstructured.Unstructured{}) require.True(t, v0 > 1000) require.Equal(t, int64(1), v1-v0) require.Equal(t, int64(1), v2-v1) diff --git a/pkg/storage/unified/sql/notifier_sql.go b/pkg/storage/unified/sql/notifier_sql.go index da06c5b6555..4937ce83610 100644 --- a/pkg/storage/unified/sql/notifier_sql.go +++ b/pkg/storage/unified/sql/notifier_sql.go @@ -37,7 +37,7 @@ type pollingNotifier struct { tracer trace.Tracer storageMetrics *resource.StorageMetrics - bulkLock *bulkLock + bulkLock *resource.BulkLock listLatestRVs func(ctx context.Context) (groupResourceRV, error) historyPoll func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) @@ -53,7 +53,7 @@ type pollingNotifierConfig struct { tracer trace.Tracer storageMetrics *resource.StorageMetrics - bulkLock *bulkLock + bulkLock *resource.BulkLock listLatestRVs func(ctx context.Context) (groupResourceRV, error) historyPoll func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) diff --git a/pkg/storage/unified/sql/notifier_sql_test.go b/pkg/storage/unified/sql/notifier_sql_test.go index d0998d71d86..c69dee8b282 100644 --- a/pkg/storage/unified/sql/notifier_sql_test.go +++ b/pkg/storage/unified/sql/notifier_sql_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/require" "go.opentelemetry.io/otel/trace/noop" + "github.com/grafana/grafana/pkg/storage/unified/resource" "github.com/grafana/grafana/pkg/storage/unified/resourcepb" "github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate" ) @@ -29,7 +30,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -43,7 +44,7 @@ func TestPollingNotifierConfig(t *testing.T) { name: "missing historyPoll", config: &pollingNotifierConfig{ listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -59,7 +60,7 @@ func TestPollingNotifierConfig(t *testing.T) { historyPoll: func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) { return nil, nil }, - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -92,7 +93,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, log: &logging.NoOpLogger{}, watchBufferSize: 10, pollingInterval: time.Second, @@ -108,7 +109,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), watchBufferSize: 10, pollingInterval: time.Second, @@ -124,7 +125,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 0, @@ -141,7 +142,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -158,7 +159,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -174,7 +175,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -256,7 +257,7 @@ func TestPollingNotifier(t *testing.T) { watchBufferSize: 10, log: &logging.NoOpLogger{}, tracer: noop.NewTracerProvider().Tracer("test"), - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, listLatestRVs: listLatestRVs, historyPoll: historyPoll, done: done, @@ -310,7 +311,7 @@ func TestPollingNotifier(t *testing.T) { watchBufferSize: 10, log: &logging.NoOpLogger{}, tracer: noop.NewTracerProvider().Tracer("test"), - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, listLatestRVs: listLatestRVs, historyPoll: historyPoll, done: done, @@ -344,7 +345,7 @@ func TestPollingNotifier(t *testing.T) { watchBufferSize: 10, log: &logging.NoOpLogger{}, tracer: noop.NewTracerProvider().Tracer("test"), - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, historyPoll: func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) { return nil, nil @@ -381,7 +382,7 @@ func TestPollingNotifier(t *testing.T) { watchBufferSize: 10, log: &logging.NoOpLogger{}, tracer: noop.NewTracerProvider().Tracer("test"), - bulkLock: &bulkLock{}, + bulkLock: &resource.BulkLock{}, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, historyPoll: func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) { return nil, nil