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

move bulkLock and bulkRv logic from sql into resource package

Browse Source
This commit is contained in:
Will Assis
2025-11-11 17:14:40 -03:00
parent a9b52589ff
commit 821c553132
6 changed files with 109 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/resource/bulk.go
+84
View File
@@ -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
}
pkg/storage/unified/sql/backend.go
+2 -2
View File
@@ -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
pkg/storage/unified/sql/bulk.go
+2 -81
View File
@@ -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))
}
pkg/storage/unified/sql/bulk_test.go
+5 -4
View File
@@ -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)
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 *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)
pkg/storage/unified/sql/notifier_sql_test.go
+14 -13
View File
@@ -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