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

revert sql/bulk.go changes

Browse Source
This commit is contained in:
Will Assis
2025-12-02 13:11:33 -03:00
parent c96c4d1e09
commit 6d7ecc500c
4 changed files with 241 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/sql/bulk.go
+222 -107
View File
@@ -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 {
pkg/storage/unified/sql/bulk_test.go
+4 -5
View File
@@ -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)
pkg/storage/unified/sql/notifier_sql.go
+2 -2
View File
@@ -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)
pkg/storage/unified/sql/notifier_sql_test.go
+13 -14
View File
@@ -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