From 6d7ecc500c6c170d176f67a5e07be7133a7db720 Mon Sep 17 00:00:00 2001 From: Will Assis Date: Tue, 2 Dec 2025 13:11:02 -0300 Subject: [PATCH] revert sql/bulk.go changes --- pkg/storage/unified/sql/bulk.go | 329 +++++++++++++------ 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 +- 4 files changed, 241 insertions(+), 128 deletions(-) diff --git a/pkg/storage/unified/sql/bulk.go b/pkg/storage/unified/sql/bulk.go index 105ddd3dfe6..6580975a764 100644 --- a/pkg/storage/unified/sql/bulk.go +++ b/pkg/storage/unified/sql/bulk.go @@ -3,18 +3,25 @@ package sql import ( "context" "fmt" + "net/http" "os" + "sync" "time" + "github.com/fullstorydev/grpchan/inprocgrpc" "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" "github.com/grafana/grafana/pkg/storage/unified/sql/db" + "github.com/grafana/grafana/pkg/storage/unified/sql/db/dbimpl" "github.com/grafana/grafana/pkg/storage/unified/sql/dbutil" "github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate" ) @@ -23,6 +30,80 @@ 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 { @@ -32,6 +113,19 @@ func (b *backend) ProcessBulk(ctx context.Context, setting resource.BulkSettings } defer b.bulkLock.Finish(setting.Collection) + // If provided, reuse the inproc transaction for SQLite + if clientCtx := inprocgrpc.ClientContext(ctx); clientCtx != nil && b.dialect.DialectName() == "sqlite" { + if externalTx := resource.TransactionFromContext(clientCtx); externalTx != nil { + b.log.Info("Using SQLite transaction from client context") + rsp := &resourcepb.BulkResponse{} + err := b.processBulkWithTx(ctx, dbimpl.NewTx(externalTx), setting, iter, rsp) + if err != nil { + rsp.Error = resource.AsErrorResult(err) + } + return rsp + } + } + // We may want to first write parquet, then read parquet if b.dialect.DialectName() == "sqlite" { file, err := os.CreateTemp("", "grafana-bulk-export-*.parquet") @@ -72,109 +166,134 @@ func (b *backend) ProcessBulk(ctx context.Context, setting resource.BulkSettings func (b *backend) processBulk(ctx context.Context, setting resource.BulkSettings, iter resource.BulkRequestIterator) *resourcepb.BulkResponse { rsp := &resourcepb.BulkResponse{} err := b.db.WithTx(ctx, ReadCommitted, func(ctx context.Context, tx db.Tx) error { - rollbackWithError := func(err error) error { - txerr := tx.Rollback() - if txerr != nil { - b.log.Warn("rollback", "error", txerr) - } else { - b.log.Info("rollback") + return b.processBulkWithTx(ctx, tx, setting, iter, rsp) + }) + if err != nil { + rsp.Error = resource.AsErrorResult(err) + } + return rsp +} + +// processBulkWithTx performs the bulk operation using the provided transaction. +// This is used both when creating our own transaction and when reusing an external one. +func (b *backend) processBulkWithTx(ctx context.Context, tx db.Tx, setting resource.BulkSettings, iter resource.BulkRequestIterator, rsp *resourcepb.BulkResponse) error { + rollbackWithError := func(err error) error { + txerr := tx.Rollback() + if txerr != nil { + b.log.Warn("rollback", "error", txerr) + } else { + b.log.Info("rollback") + } + return err + } + bulk := &bulkWroker{ + ctx: ctx, + tx: tx, + dialect: b.dialect, + logger: logging.FromContext(ctx), + } + + // Calculate the RV based on incoming request timestamps + rv := newBulkRV() + + summaries := make(map[string]*resourcepb.BulkResponse_Summary, len(setting.Collection)) + + // First clear everything in the transaction + if setting.RebuildCollection { + for _, key := range setting.Collection { + summary, err := bulk.deleteCollection(key) + if err != nil { + return rollbackWithError(err) } + summaries[resource.NSGR(key)] = summary + rsp.Summary = append(rsp.Summary, summary) + } + } else { + for _, key := range setting.Collection { + summaries[resource.NSGR(key)] = &resourcepb.BulkResponse_Summary{ + Namespace: key.Namespace, + Group: key.Group, + Resource: key.Resource, + } + } + } + + obj := &unstructured.Unstructured{} + + // Write each event into the history + for iter.Next() { + if iter.RollbackRequested() { + return rollbackWithError(nil) + } + req := iter.Request() + if req == nil { + return rollbackWithError(fmt.Errorf("missing request")) + } + rsp.Processed++ + + if req.Action == resourcepb.BulkRequest_UNKNOWN { + rsp.Rejected = append(rsp.Rejected, &resourcepb.BulkResponse_Rejected{ + Key: req.Key, + Action: req.Action, + Error: "unknown action", + }) + continue + } + + err := obj.UnmarshalJSON(req.Value) + if err != nil { + rsp.Rejected = append(rsp.Rejected, &resourcepb.BulkResponse_Rejected{ + Key: req.Key, + Action: req.Action, + Error: "unable to unmarshal json", + }) + continue + } + + // Write the event to history + if _, err := dbutil.Exec(ctx, tx, sqlResourceHistoryInsert, sqlResourceRequest{ + SQLTemplate: sqltemplate.New(b.dialect), + WriteEvent: resource.WriteEvent{ + Key: req.Key, + Type: resourcepb.WatchEvent_Type(req.Action), + Value: req.Value, + PreviousRV: -1, // Used for WATCH, but we want to skip watch events + }, + Folder: req.Folder, + GUID: uuid.New().String(), + ResourceVersion: rv.next(obj), + }); err != nil { + return rollbackWithError(fmt.Errorf("insert into resource history: %w", err)) + } + } + + // Now update the resource table from history + for _, key := range setting.Collection { + k := fmt.Sprintf("%s/%s/%s", key.Namespace, key.Group, key.Resource) + summary := summaries[k] + if summary == nil { + return rollbackWithError(fmt.Errorf("missing summary key for: %s", k)) + } + + err := bulk.syncCollection(key, summary) + if err != nil { return err } - bulk := &bulkWroker{ - ctx: ctx, - tx: tx, - dialect: b.dialect, - logger: logging.FromContext(ctx), - } - // Calculate the RV based on incoming request timestamps - rv := resource.NewBulkRV() - - summaries := make(map[string]*resourcepb.BulkResponse_Summary, len(setting.Collection)) - - // First clear everything in the transaction - if setting.RebuildCollection { - for _, key := range setting.Collection { - summary, err := bulk.deleteCollection(key) - if err != nil { - return rollbackWithError(err) + if b.dialect.DialectName() == "sqlite" { + nextRV, err := b.rvManager.lock(ctx, tx, key.Group, key.Resource) + if err != nil { + b.log.Error("error locking RV", "error", err, "key", resource.NSGR(key)) + } else { + b.log.Info("successfully locked RV", "nextRV", nextRV, "key", resource.NSGR(key)) + // Save the incremented RV + if err := b.rvManager.saveRV(ctx, tx, key.Group, key.Resource, nextRV); err != nil { + b.log.Error("error saving RV", "error", err, "key", resource.NSGR(key)) + } else { + b.log.Info("successfully saved RV", "rv", nextRV, "key", resource.NSGR(key)) } - summaries[resource.NSGR(key)] = summary - rsp.Summary = append(rsp.Summary, summary) } } else { - for _, key := range setting.Collection { - summaries[resource.NSGR(key)] = &resourcepb.BulkResponse_Summary{ - Namespace: key.Namespace, - Group: key.Group, - Resource: key.Resource, - } - } - } - - obj := &unstructured.Unstructured{} - - // Write each event into the history - for iter.Next() { - if iter.RollbackRequested() { - return rollbackWithError(nil) - } - req := iter.Request() - if req == nil { - return rollbackWithError(fmt.Errorf("missing request")) - } - rsp.Processed++ - - if req.Action == resourcepb.BulkRequest_UNKNOWN { - rsp.Rejected = append(rsp.Rejected, &resourcepb.BulkResponse_Rejected{ - Key: req.Key, - Action: req.Action, - Error: "unknown action", - }) - continue - } - - err := obj.UnmarshalJSON(req.Value) - if err != nil { - rsp.Rejected = append(rsp.Rejected, &resourcepb.BulkResponse_Rejected{ - Key: req.Key, - Action: req.Action, - Error: "unable to unmarshal json", - }) - continue - } - - // Write the event to history - if _, err := dbutil.Exec(ctx, tx, sqlResourceHistoryInsert, sqlResourceRequest{ - SQLTemplate: sqltemplate.New(b.dialect), - WriteEvent: resource.WriteEvent{ - Key: req.Key, - Type: resourcepb.WatchEvent_Type(req.Action), - Value: req.Value, - PreviousRV: -1, // Used for WATCH, but we want to skip watch events - }, - Folder: req.Folder, - GUID: uuid.New().String(), - ResourceVersion: rv.Next(obj), - }); err != nil { - return rollbackWithError(fmt.Errorf("insert into resource history: %w", err)) - } - } - - // Now update the resource table from history - for _, key := range setting.Collection { - k := fmt.Sprintf("%s/%s/%s", key.Namespace, key.Group, key.Resource) - summary := summaries[k] - if summary == nil { - return rollbackWithError(fmt.Errorf("missing summary key for: %s", k)) - } - - err := bulk.syncCollection(key, summary) - if err != nil { - return err - } - // Make sure the collection RV is above our last written event _, err = b.rvManager.ExecWithRV(ctx, key, func(tx db.Tx) (string, error) { return "", nil @@ -182,19 +301,15 @@ func (b *backend) processBulk(ctx context.Context, setting resource.BulkSettings if err != nil { b.log.Warn("error increasing RV", "error", err) } - - // Update the last import time. This is important to trigger reindexing - // of the resource for a given namespace. - if err := b.updateLastImportTime(ctx, tx, key, time.Now()); err != nil { - return rollbackWithError(err) - } } - return nil - }) - if err != nil { - rsp.Error = resource.AsErrorResult(err) + + // Update the last import time. This is important to trigger reindexing + // of the resource for a given namespace. + if err := b.updateLastImportTime(ctx, tx, key, time.Now()); err != nil { + return rollbackWithError(err) + } } - return rsp + return nil } func (b *backend) updateLastImportTime(ctx context.Context, tx db.Tx, key *resourcepb.ResourceKey, now time.Time) error { diff --git a/pkg/storage/unified/sql/bulk_test.go b/pkg/storage/unified/sql/bulk_test.go index 4d8738d479f..e1e00fb7839 100644 --- a/pkg/storage/unified/sql/bulk_test.go +++ b/pkg/storage/unified/sql/bulk_test.go @@ -3,7 +3,6 @@ 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" ) @@ -14,10 +13,10 @@ func TestBatch(t *testing.T) { t.Run("rv iterator", func(t *testing.T) { t.Parallel() - rv := resource.NewBulkRV() - v0 := rv.Next(&unstructured.Unstructured{}) - v1 := rv.Next(&unstructured.Unstructured{}) - v2 := rv.Next(&unstructured.Unstructured{}) + rv := 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 4937ce83610..da06c5b6555 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 *resource.BulkLock + bulkLock *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 *resource.BulkLock + bulkLock *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 c69dee8b282..d0998d71d86 100644 --- a/pkg/storage/unified/sql/notifier_sql_test.go +++ b/pkg/storage/unified/sql/notifier_sql_test.go @@ -10,7 +10,6 @@ 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" ) @@ -30,7 +29,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -44,7 +43,7 @@ func TestPollingNotifierConfig(t *testing.T) { name: "missing historyPoll", config: &pollingNotifierConfig{ listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -60,7 +59,7 @@ func TestPollingNotifierConfig(t *testing.T) { historyPoll: func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) { return nil, nil }, - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -93,7 +92,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, log: &logging.NoOpLogger{}, watchBufferSize: 10, pollingInterval: time.Second, @@ -109,7 +108,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), watchBufferSize: 10, pollingInterval: time.Second, @@ -125,7 +124,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 0, @@ -142,7 +141,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -159,7 +158,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -175,7 +174,7 @@ func TestPollingNotifierConfig(t *testing.T) { return nil, nil }, listLatestRVs: func(ctx context.Context) (groupResourceRV, error) { return nil, nil }, - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, tracer: noop.NewTracerProvider().Tracer("test"), log: &logging.NoOpLogger{}, watchBufferSize: 10, @@ -257,7 +256,7 @@ func TestPollingNotifier(t *testing.T) { watchBufferSize: 10, log: &logging.NoOpLogger{}, tracer: noop.NewTracerProvider().Tracer("test"), - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, listLatestRVs: listLatestRVs, historyPoll: historyPoll, done: done, @@ -311,7 +310,7 @@ func TestPollingNotifier(t *testing.T) { watchBufferSize: 10, log: &logging.NoOpLogger{}, tracer: noop.NewTracerProvider().Tracer("test"), - bulkLock: &resource.BulkLock{}, + bulkLock: &bulkLock{}, listLatestRVs: listLatestRVs, historyPoll: historyPoll, done: done, @@ -345,7 +344,7 @@ func TestPollingNotifier(t *testing.T) { watchBufferSize: 10, log: &logging.NoOpLogger{}, tracer: noop.NewTracerProvider().Tracer("test"), - bulkLock: &resource.BulkLock{}, + bulkLock: &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 @@ -382,7 +381,7 @@ func TestPollingNotifier(t *testing.T) { watchBufferSize: 10, log: &logging.NoOpLogger{}, tracer: noop.NewTracerProvider().Tracer("test"), - bulkLock: &resource.BulkLock{}, + bulkLock: &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