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

Fix kvstore wire (#111827)

Browse Source
This commit is contained in:
Georges Chaudy
2025-10-14 11:30:07 +02:00
committed by GitHub
parent 16502532e1
commit 616de7b566
12 changed files with 65 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/server/module_server.go
+6 -2
View File
@@ -45,8 +45,9 @@ func NewModule(opts Options,
tracer tracing.Tracer, // Ensures tracing is initialized
license licensing.Licensing,
moduleRegisterer ModuleRegisterer,
storageBackend resource.StorageBackend, // Ensures unified storage backend is initialized
) (*ModuleServer, error) {
s, err := newModuleServer(opts, apiOpts, features, cfg, storageMetrics, indexMetrics, reg, promGatherer, license, moduleRegisterer)
s, err := newModuleServer(opts, apiOpts, features, cfg, storageMetrics, indexMetrics, reg, promGatherer, license, moduleRegisterer, storageBackend)
if err != nil {
return nil, err
}
@@ -68,6 +69,7 @@ func newModuleServer(opts Options,
promGatherer prometheus.Gatherer,
license licensing.Licensing,
moduleRegisterer ModuleRegisterer,
storageBackend resource.StorageBackend,
) (*ModuleServer, error) {
rootCtx, shutdownFn := context.WithCancel(context.Background())
@@ -90,6 +92,7 @@ func newModuleServer(opts Options,
registerer: reg,
license: license,
moduleRegisterer: moduleRegisterer,
storageBackend: storageBackend,
}
return s, nil
@@ -111,6 +114,7 @@ type ModuleServer struct {
shutdownFinished chan struct{}
isInitialized bool
mtx sync.Mutex
storageBackend resource.StorageBackend
storageMetrics *resource.StorageMetrics
indexMetrics *resource.BleveIndexMetrics
license licensing.Licensing
@@ -193,7 +197,7 @@ func (s *ModuleServer) Run() error {
if err != nil {
return nil, err
}
return sql.ProvideUnifiedStorageGrpcService(s.cfg, s.features, nil, s.log, s.registerer, docBuilders, s.storageMetrics, s.indexMetrics, s.searchServerRing, s.MemberlistKVConfig, s.httpServerRouter)
return sql.ProvideUnifiedStorageGrpcService(s.cfg, s.features, nil, s.log, s.registerer, docBuilders, s.storageMetrics, s.indexMetrics, s.searchServerRing, s.MemberlistKVConfig, s.httpServerRouter, s.storageBackend)
})
m.RegisterModule(modules.ZanzanaServer, func() (services.Service, error) {
pkg/server/search_server_distributor_test.go
+1 -1
View File
@@ -326,7 +326,7 @@ func initModuleServerForTest(
) testModuleServer {
tracer := tracing.InitializeTracerForTest()
ms, err := NewModule(opts, apiOpts, featuremgmt.WithFeatures(featuremgmt.FlagUnifiedStorageSearch), cfg, nil, nil, prometheus.NewRegistry(), prometheus.DefaultGatherer, tracer, nil, ProvideNoopModuleRegisterer())
ms, err := NewModule(opts, apiOpts, featuremgmt.WithFeatures(featuremgmt.FlagUnifiedStorageSearch), cfg, nil, nil, prometheus.NewRegistry(), prometheus.DefaultGatherer, tracer, nil, ProvideNoopModuleRegisterer(), nil)
require.NoError(t, err)
conn, err := grpc.NewClient(cfg.GRPCServer.Address,
pkg/server/wire_gen.go
+6 -1
View File
@@ -258,6 +258,7 @@ import (
"github.com/grafana/grafana/pkg/storage/unified"
"github.com/grafana/grafana/pkg/storage/unified/resource"
"github.com/grafana/grafana/pkg/storage/unified/search"
"github.com/grafana/grafana/pkg/storage/unified/sql"
"github.com/grafana/grafana/pkg/tsdb/azuremonitor"
"github.com/grafana/grafana/pkg/tsdb/cloud-monitoring"
"github.com/grafana/grafana/pkg/tsdb/cloudwatch"
@@ -1622,7 +1623,11 @@ func InitializeModuleServer(cfg *setting.Cfg, opts Options, apiOpts api.ServerOp
hooksService := hooks.ProvideService()
ossLicensingService := licensing.ProvideService(cfg, hooksService)
moduleRegisterer := ProvideNoopModuleRegisterer()
moduleServer, err := NewModule(opts, apiOpts, featureToggles, cfg, storageMetrics, bleveIndexMetrics, registerer, gatherer, tracingService, ossLicensingService, moduleRegisterer)
storageBackend, err := sql.ProvideStorageBackend(cfg)
if err != nil {
return nil, err
}
moduleServer, err := NewModule(opts, apiOpts, featureToggles, cfg, storageMetrics, bleveIndexMetrics, registerer, gatherer, tracingService, ossLicensingService, moduleRegisterer, storageBackend)
if err != nil {
return nil, err
}
pkg/server/wireexts_oss.go
+3
View File
@@ -64,6 +64,7 @@ import (
"github.com/grafana/grafana/pkg/storage/unified"
"github.com/grafana/grafana/pkg/storage/unified/resource"
search2 "github.com/grafana/grafana/pkg/storage/unified/search"
"github.com/grafana/grafana/pkg/storage/unified/sql"
)
var provisioningExtras = wire.NewSet(
@@ -141,6 +142,7 @@ var wireExtsBasicSet = wire.NewSet(
wire.Bind(new(sandbox.Sandbox), new(*sandbox.Service)),
wire.Struct(new(unified.Options), "*"),
unified.ProvideUnifiedStorageClient,
sql.ProvideStorageBackend,
builder.ProvideDefaultBuildHandlerChainFuncFromBuilders,
aggregatorrunner.ProvideNoopAggregatorConfigurator,
apisregistry.WireSetExts,
@@ -193,6 +195,7 @@ var wireExtsModuleServerSet = wire.NewSet(
resource.ProvideIndexMetrics,
// Overriden by enterprise
ProvideNoopModuleRegisterer,
sql.ProvideStorageBackend,
)
var wireExtsStandaloneAPIServerSet = wire.NewSet(
pkg/storage/unified/resource/storage_backend.go
+2 -2
View File
@@ -50,7 +50,7 @@ type kvStorageBackend struct {
var _ StorageBackend = &kvStorageBackend{}
type KvBackendOptions struct {
type KVBackendOptions struct {
KvStore KV
WithPruner bool
EventRetentionPeriod time.Duration // How long to keep events (default: 1 hour)
@@ -59,7 +59,7 @@ type KvBackendOptions struct {
Reg prometheus.Registerer // TODO add metrics
}
func NewKvStorageBackend(opts KvBackendOptions) (StorageBackend, error) {
func NewKVStorageBackend(opts KVBackendOptions) (StorageBackend, error) {
ctx := context.Background()
kv := opts.KvStore
pkg/storage/unified/resource/storage_backend_test.go
+2 -2
View File
@@ -27,11 +27,11 @@ var appsNamespace = NamespacedResource{
func setupTestStorageBackend(t *testing.T) *kvStorageBackend {
kv := setupTestKV(t)
opts := KvBackendOptions{
opts := KVBackendOptions{
KvStore: kv,
WithPruner: true,
}
backend, err := NewKvStorageBackend(opts)
backend, err := NewKVStorageBackend(opts)
kvBackend := backend.(*kvStorageBackend)
require.NoError(t, err)
return kvBackend
pkg/storage/unified/sql/backend.go
+9
View File
@@ -20,6 +20,7 @@ import (
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime/schema"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/util/sqlite"
"github.com/grafana/grafana-app-sdk/logging"
@@ -37,6 +38,14 @@ const defaultPollingInterval = 100 * time.Millisecond
const defaultWatchBufferSize = 100 // number of events to buffer in the watch stream
const defaultPrunerHistoryLimit = 20
func ProvideStorageBackend(
cfg *setting.Cfg,
) (resource.StorageBackend, error) {
// TODO: make this the central place to provide SQL backend
// Currently it is skipped as we need to handle the cases of Diagnostics and Lifecycle
return nil, nil
}
type Backend interface {
resource.StorageBackend
resourcepb.DiagnosticsServer
pkg/storage/unified/sql/server.go
+28 -21
View File
@@ -30,6 +30,7 @@ type QOSEnqueueDequeuer interface {
// ServerOptions contains the options for creating a new ResourceServer
type ServerOptions struct {
Backend resource.StorageBackend
DB infraDB.DB
Cfg *setting.Cfg
Tracer trace.Tracer
@@ -87,29 +88,35 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
maxPageSizeBytes := unifiedStorageCfg.Key("max_page_size_bytes")
serverOptions.MaxPageSizeBytes = maxPageSizeBytes.MustInt(0)
eDB, err := dbimpl.ProvideResourceDB(opts.DB, opts.Cfg, opts.Tracer)
if err != nil {
return nil, err
if opts.Backend != nil {
serverOptions.Backend = opts.Backend
// TODO: we should probably have a proper interface for diagnostics/lifecycle
} else {
eDB, err := dbimpl.ProvideResourceDB(opts.DB, opts.Cfg, opts.Tracer)
if err != nil {
return nil, err
}
isHA := isHighAvailabilityEnabled(opts.Cfg.SectionWithEnvOverrides("database"),
opts.Cfg.SectionWithEnvOverrides("resource_api"))
withPruner := opts.Features.IsEnabledGlobally(featuremgmt.FlagUnifiedStorageHistoryPruner)
backend, err := NewBackend(BackendOptions{
DBProvider: eDB,
Tracer: opts.Tracer,
Reg: opts.Reg,
IsHA: isHA,
withPruner: withPruner,
storageMetrics: opts.StorageMetrics,
})
if err != nil {
return nil, err
}
serverOptions.Backend = backend
serverOptions.Diagnostics = backend
serverOptions.Lifecycle = backend
}
isHA := isHighAvailabilityEnabled(opts.Cfg.SectionWithEnvOverrides("database"),
opts.Cfg.SectionWithEnvOverrides("resource_api"))
withPruner := opts.Features.IsEnabledGlobally(featuremgmt.FlagUnifiedStorageHistoryPruner)
store, err := NewBackend(BackendOptions{
DBProvider: eDB,
Tracer: opts.Tracer,
Reg: opts.Reg,
IsHA: isHA,
withPruner: withPruner,
storageMetrics: opts.StorageMetrics,
})
if err != nil {
return nil, err
}
serverOptions.Backend = store
serverOptions.Diagnostics = store
serverOptions.Lifecycle = store
serverOptions.Search = opts.SearchOptions
serverOptions.IndexMetrics = opts.IndexMetrics
serverOptions.QOSQueue = opts.QOSQueue
pkg/storage/unified/sql/service.go
+4
View File
@@ -59,6 +59,7 @@ type service struct {
subservicesWatcher *services.FailureWatcher
hasSubservices bool
backend resource.StorageBackend
cfg *setting.Cfg
features featuremgmt.FeatureToggles
db infraDB.DB
@@ -97,6 +98,7 @@ func ProvideUnifiedStorageGrpcService(
searchRing *ring.Ring,
memberlistKVConfig kv.Config,
httpServerRouter *mux.Router,
backend resource.StorageBackend,
) (UnifiedStorageGrpcService, error) {
var err error
tracer := otel.Tracer("unified-storage")
@@ -109,6 +111,7 @@ func ProvideUnifiedStorageGrpcService(
})
s := &service{
backend: backend,
cfg: cfg,
features: features,
stopCh: make(chan struct{}),
@@ -263,6 +266,7 @@ func (s *service) starting(ctx context.Context) error {
}
serverOptions := ServerOptions{
Backend: s.backend,
DB: s.db,
Cfg: s.cfg,
Tracer: s.tracing,
pkg/storage/unified/sql/test/integration_test.go
+1 -1
View File
@@ -130,7 +130,7 @@ func TestClientServer(t *testing.T) {
features := featuremgmt.WithFeatures()
svc, err := sql.ProvideUnifiedStorageGrpcService(cfg, features, dbstore, nil, prometheus.NewPedanticRegistry(), nil, nil, nil, nil, kv.Config{}, nil)
svc, err := sql.ProvideUnifiedStorageGrpcService(cfg, features, dbstore, nil, prometheus.NewPedanticRegistry(), nil, nil, nil, nil, kv.Config{}, nil, nil)
require.NoError(t, err)
var client resourcepb.ResourceStoreClient
pkg/storage/unified/testing/storage_backend_test.go
+2 -2
View File
@@ -18,10 +18,10 @@ func TestBadgerKVStorageBackend(t *testing.T) {
t.Cleanup(func() {
_ = db.Close()
})
kvOpts := resource.KvBackendOptions{
kvOpts := resource.KVBackendOptions{
KvStore: resource.NewBadgerKV(db),
}
backend, err := resource.NewKvStorageBackend(kvOpts)
backend, err := resource.NewKVStorageBackend(kvOpts)
require.NoError(t, err)
return backend
}, &TestOptions{
pkg/tests/testinfra/testinfra.go
+1 -1
View File
@@ -129,7 +129,7 @@ func StartGrafanaEnv(t *testing.T, grafDir, cfgPath string) (string, *server.Tes
var storage sql.UnifiedStorageGrpcService
if runstore {
storage, err = sql.ProvideUnifiedStorageGrpcService(env.Cfg, env.FeatureToggles, env.SQLStore,
env.Cfg.Logger, prometheus.NewPedanticRegistry(), nil, nil, nil, nil, kv.Config{}, nil)
env.Cfg.Logger, prometheus.NewPedanticRegistry(), nil, nil, nil, nil, kv.Config{}, nil, nil)
require.NoError(t, err)
ctx := context.Background()
err = storage.StartAsync(ctx)