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

chore: use only needed methods in storage interface

Browse Source 
- continue cleanup and separation
This commit is contained in:
Rafael Paulovic
2026-01-14 21:14:24 +01:00
parent b0bb71f834
commit 2fab497c18
16 changed files with 161 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/cmd/grafana-cli/commands/datamigrations/to_unified_storage.go
+4 -4
View File
@@ -74,7 +74,7 @@ func ToUnifiedStorage(c utils.CommandLine, cfg *setting.Cfg, sqlStore db.DB) err
return err
}
grpcClient, err := newUnifiedClient(cfg, sqlStore, featureToggles)
grpcClient, err := newUnifiedMigratorClient(cfg, sqlStore, featureToggles)
if err != nil {
return err
}
@@ -92,7 +92,7 @@ func ToUnifiedStorage(c utils.CommandLine, cfg *setting.Cfg, sqlStore db.DB) err
return runInteractiveMigration(ctx, cfg, opts, dashboardAccess, grpcClient, start)
}
func runNonInteractiveMigration(ctx context.Context, opts legacy.MigrateOptions, dashboardAccess legacy.MigrationDashboardAccessor, grpcClient resource.ResourceClient, start time.Time) error {
func runNonInteractiveMigration(ctx context.Context, opts legacy.MigrateOptions, dashboardAccess legacy.MigrationDashboardAccessor, grpcClient resource.MigratorClient, start time.Time) error {
migrator := migrations.ProvideUnifiedMigrator(dashboardAccess, grpcClient)
opts.WithHistory = true // always include history in non-interactive mode
@@ -109,7 +109,7 @@ func runNonInteractiveMigration(ctx context.Context, opts legacy.MigrateOptions,
return nil
}
func runInteractiveMigration(ctx context.Context, cfg *setting.Cfg, opts legacy.MigrateOptions, dashboardAccess legacy.MigrationDashboardAccessor, grpcClient resource.ResourceClient, start time.Time) error {
func runInteractiveMigration(ctx context.Context, cfg *setting.Cfg, opts legacy.MigrateOptions, dashboardAccess legacy.MigrationDashboardAccessor, grpcClient resource.MigratorClient, start time.Time) error {
yes, err := promptYesNo(fmt.Sprintf("Count legacy resources for namespace: %s?", opts.Namespace))
if err != nil {
return err
@@ -225,7 +225,7 @@ func promptYesNo(prompt string) (bool, error) {
}
}
func newUnifiedClient(cfg *setting.Cfg, sqlStore db.DB, featureToggles featuremgmt.FeatureToggles) (resource.ResourceClient, error) {
func newUnifiedMigratorClient(cfg *setting.Cfg, sqlStore db.DB, featureToggles featuremgmt.FeatureToggles) (resource.MigratorClient, error) {
return unified.ProvideUnifiedStorageClient(&unified.Options{
Cfg: cfg,
Features: featureToggles,
pkg/registry/apis/dashboard/legacy/client.go
+14 -48
View File
@@ -11,93 +11,59 @@ import (
)
var (
_ resource.ResourceClient = (*directResourceClient)(nil)
_ resource.StorageClient = (*DirectStorageClient)(nil)
)
// The direct client passes requests directly to the server using the *same* context
func NewDirectResourceClient(server resource.ResourceServer) resource.ResourceClient {
return &directResourceClient{server}
// NewDirectStorageClient creates a client that passes requests directly to the server using the *same* context
func NewDirectStorageClient(server resource.ResourceServer) *DirectStorageClient {
return &DirectStorageClient{server}
}
type directResourceClient struct {
type DirectStorageClient struct {
server resource.ResourceServer
}
// Create implements ResourceClient.
func (d *directResourceClient) Create(ctx context.Context, in *resourcepb.CreateRequest, opts ...grpc.CallOption) (*resourcepb.CreateResponse, error) {
func (d *DirectStorageClient) Create(ctx context.Context, in *resourcepb.CreateRequest, _ ...grpc.CallOption) (*resourcepb.CreateResponse, error) {
return d.server.Create(ctx, in)
}
// Delete implements ResourceClient.
func (d *directResourceClient) Delete(ctx context.Context, in *resourcepb.DeleteRequest, opts ...grpc.CallOption) (*resourcepb.DeleteResponse, error) {
func (d *DirectStorageClient) Delete(ctx context.Context, in *resourcepb.DeleteRequest, _ ...grpc.CallOption) (*resourcepb.DeleteResponse, error) {
return d.server.Delete(ctx, in)
}
// GetBlob implements ResourceClient.
func (d *directResourceClient) GetBlob(ctx context.Context, in *resourcepb.GetBlobRequest, opts ...grpc.CallOption) (*resourcepb.GetBlobResponse, error) {
func (d *DirectStorageClient) GetBlob(ctx context.Context, in *resourcepb.GetBlobRequest, _ ...grpc.CallOption) (*resourcepb.GetBlobResponse, error) {
return d.server.GetBlob(ctx, in)
}
// GetStats implements ResourceClient (SearchClient).
func (d *directResourceClient) GetStats(ctx context.Context, in *resourcepb.ResourceStatsRequest, opts ...grpc.CallOption) (*resourcepb.ResourceStatsResponse, error) {
return nil, fmt.Errorf("GetStats not supported with direct resource client")
}
// IsHealthy implements ResourceClient.
func (d *directResourceClient) IsHealthy(ctx context.Context, in *resourcepb.HealthCheckRequest, opts ...grpc.CallOption) (*resourcepb.HealthCheckResponse, error) {
func (d *DirectStorageClient) IsHealthy(ctx context.Context, in *resourcepb.HealthCheckRequest, _ ...grpc.CallOption) (*resourcepb.HealthCheckResponse, error) {
return d.server.IsHealthy(ctx, in)
}
// List implements ResourceClient.
func (d *directResourceClient) List(ctx context.Context, in *resourcepb.ListRequest, opts ...grpc.CallOption) (*resourcepb.ListResponse, error) {
func (d *DirectStorageClient) List(ctx context.Context, in *resourcepb.ListRequest, _ ...grpc.CallOption) (*resourcepb.ListResponse, error) {
return d.server.List(ctx, in)
}
// ListManagedObjects implements ResourceClient (SearchClient).
func (d *directResourceClient) ListManagedObjects(ctx context.Context, in *resourcepb.ListManagedObjectsRequest, opts ...grpc.CallOption) (*resourcepb.ListManagedObjectsResponse, error) {
return nil, fmt.Errorf("ListManagedObjects not supported with direct resource client")
}
// CountManagedObjects implements ResourceClient (SearchClient).
func (d *directResourceClient) CountManagedObjects(ctx context.Context, in *resourcepb.CountManagedObjectsRequest, opts ...grpc.CallOption) (*resourcepb.CountManagedObjectsResponse, error) {
return nil, fmt.Errorf("CountManagedObjects not supported with direct resource client")
}
// PutBlob implements ResourceClient.
func (d *directResourceClient) PutBlob(ctx context.Context, in *resourcepb.PutBlobRequest, opts ...grpc.CallOption) (*resourcepb.PutBlobResponse, error) {
func (d *DirectStorageClient) PutBlob(ctx context.Context, in *resourcepb.PutBlobRequest, _ ...grpc.CallOption) (*resourcepb.PutBlobResponse, error) {
return d.server.PutBlob(ctx, in)
}
// Read implements ResourceClient.
func (d *directResourceClient) Read(ctx context.Context, in *resourcepb.ReadRequest, opts ...grpc.CallOption) (*resourcepb.ReadResponse, error) {
func (d *DirectStorageClient) Read(ctx context.Context, in *resourcepb.ReadRequest, _ ...grpc.CallOption) (*resourcepb.ReadResponse, error) {
return d.server.Read(ctx, in)
}
// Search implements ResourceClient (SearchClient).
func (d *directResourceClient) Search(ctx context.Context, in *resourcepb.ResourceSearchRequest, opts ...grpc.CallOption) (*resourcepb.ResourceSearchResponse, error) {
return nil, fmt.Errorf("Search not supported with direct resource client")
}
// Update implements ResourceClient.
func (d *directResourceClient) Update(ctx context.Context, in *resourcepb.UpdateRequest, opts ...grpc.CallOption) (*resourcepb.UpdateResponse, error) {
func (d *DirectStorageClient) Update(ctx context.Context, in *resourcepb.UpdateRequest, _ ...grpc.CallOption) (*resourcepb.UpdateResponse, error) {
return d.server.Update(ctx, in)
}
// Watch implements ResourceClient.
func (d *directResourceClient) Watch(ctx context.Context, in *resourcepb.WatchRequest, opts ...grpc.CallOption) (resourcepb.ResourceStore_WatchClient, error) {
func (d *DirectStorageClient) Watch(_ context.Context, _ *resourcepb.WatchRequest, _ ...grpc.CallOption) (resourcepb.ResourceStore_WatchClient, error) {
return nil, fmt.Errorf("watch not supported with direct resource client")
}
// BulkProcess implements resource.ResourceClient.
func (d *directResourceClient) BulkProcess(ctx context.Context, opts ...grpc.CallOption) (resourcepb.BulkStore_BulkProcessClient, error) {
return nil, fmt.Errorf("BulkProcess not supported with direct resource client")
}
// RebuildIndexes implements resource.ResourceClient.
func (b *directResourceClient) RebuildIndexes(ctx context.Context, req *resourcepb.RebuildIndexesRequest, opts ...grpc.CallOption) (*resourcepb.RebuildIndexesResponse, error) {
return nil, fmt.Errorf("not implemented")
}
func (b *directResourceClient) GetQuotaUsage(ctx context.Context, req *resourcepb.QuotaUsageRequest, opts ...grpc.CallOption) (*resourcepb.QuotaUsageResponse, error) {
return nil, fmt.Errorf("not implemented")
}
pkg/registry/apis/dashboard/legacy_storage.go
+1 -1
View File
@@ -41,7 +41,7 @@ func (s *DashboardStorage) NewStore(dash utils.ResourceInfo, scheme *runtime.Sch
if err != nil {
return nil, err
}
client := legacy.NewDirectResourceClient(server) // same context
client := legacy.NewDirectStorageClient(server) // same context
optsGetter := apistore.NewRESTOptionsGetterForClient(client, nil,
defaultOpts.StorageConfig.Config, nil,
)
pkg/server/module_server.go
+1 -1
View File
@@ -210,7 +210,7 @@ func (s *ModuleServer) Run() error {
if err != nil {
return nil, err
}
return sql.ProvideSearchGrpcService(s.cfg, s.features, nil, s.log, s.registerer, docBuilders, s.indexMetrics, s.searchServerRing, s.MemberlistKVConfig, s.storageBackend)
return sql.ProvideUnifiedSearchGrpcService(s.cfg, s.features, nil, s.log, s.registerer, docBuilders, s.indexMetrics, s.searchServerRing, s.MemberlistKVConfig, s.storageBackend)
})
m.RegisterModule(modules.ZanzanaServer, func() (services.Service, error) {
pkg/server/search_server_distributor_test.go
+2 -2
View File
@@ -372,7 +372,7 @@ func initModuleServerForTest(
return testModuleServer{server: ms, grpcAddress: cfg.GRPCServer.Address, httpPort: cfg.HTTPPort, healthClient: healthClient, id: cfg.InstanceID}
}
func createBaselineServer(t *testing.T, dbType, dbConnStr string, testNamespaces []string) resource.ResourceServer {
func createBaselineServer(t *testing.T, dbType, dbConnStr string, testNamespaces []string) resource.SearchServer {
cfg := setting.NewCfg()
section, err := cfg.Raw.NewSection("database")
require.NoError(t, err)
@@ -422,7 +422,7 @@ func createBaselineServer(t *testing.T, dbType, dbConnStr string, testNamespaces
}
}
return server
return searchServer
}
var counter int
pkg/server/wireexts_oss.go
+1
View File
@@ -148,6 +148,7 @@ var wireExtsBasicSet = wire.NewSet(
wire.Struct(new(unified.Options), "*"),
unified.ProvideUnifiedStorageClient,
wire.Bind(new(resourcepb.ResourceIndexClient), new(resource.ResourceClient)),
wire.Bind(new(resource.MigratorClient), new(resource.ResourceClient)),
sql.ProvideStorageBackend,
builder.ProvideDefaultBuildHandlerChainFuncFromBuilders,
aggregatorrunner.ProvideNoopAggregatorConfigurator,
pkg/storage/unified/apistore/restoptions.go
+2 -2
View File
@@ -26,7 +26,7 @@ var _ generic.RESTOptionsGetter = (*RESTOptionsGetter)(nil)
type StorageOptionsRegister func(gr schema.GroupResource, opts StorageOptions)
type RESTOptionsGetter struct {
client resource.ResourceClient
client resource.StorageClient
secrets secret.InlineSecureValueSupport
original storagebackend.Config
configProvider RestConfigProvider
@@ -36,7 +36,7 @@ type RESTOptionsGetter struct {
}
func NewRESTOptionsGetterForClient(
client resource.ResourceClient,
client resource.StorageClient,
secrets secret.InlineSecureValueSupport,
original storagebackend.Config,
configProvider RestConfigProvider,
pkg/storage/unified/apistore/store.go
+2 -2
View File
@@ -88,7 +88,7 @@ type Storage struct {
trigger storage.IndexerFuncs
indexers *cache.Indexers
store resource.ResourceClient
store resource.StorageClient
getKey func(string) (*resourcepb.ResourceKey, error)
snowflake *snowflake.Node // used to enforce internal ids
configProvider RestConfigProvider // used for provisioning
@@ -112,7 +112,7 @@ type RestConfigProvider interface {
// NewStorage instantiates a new Storage.
func NewStorage(
config *storagebackend.ConfigForResource,
store resource.ResourceClient,
store resource.StorageClient,
keyFunc func(obj runtime.Object) (string, error),
keyParser func(key string) (*resourcepb.ResourceKey, error),
newFunc func() runtime.Object,
pkg/storage/unified/migrations/migrator.go
+2 -2
View File
@@ -48,7 +48,7 @@ func buildCollectionSettings(opts legacy.MigrateOptions) resource.BulkSettings {
}
type resourceClientStreamProvider struct {
client resource.ResourceClient
client resource.MigratorClient
}
func (r *resourceClientStreamProvider) createStream(ctx context.Context, opts legacy.MigrateOptions) (resourcepb.BulkStore_BulkProcessClient, error) {
@@ -71,7 +71,7 @@ func (b *bulkStoreClientStreamProvider) createStream(ctx context.Context, opts l
// This can migrate Folders, Dashboards and LibraryPanels
func ProvideUnifiedMigrator(
dashboardAccess legacy.MigrationDashboardAccessor,
client resource.ResourceClient,
client resource.MigratorClient,
) UnifiedMigrator {
return newUnifiedMigrator(
dashboardAccess,
pkg/storage/unified/resource/client.go
+18 -5
View File
@@ -31,20 +31,33 @@ import (
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
)
// SearchClient is used to interact with unified search
type SearchClient interface {
resourcepb.ResourceIndexClient
resourcepb.ManagedObjectIndexClient
}
// StorageClient is used to interact with unified storage
type StorageClient interface {
resourcepb.ResourceStoreClient
resourcepb.BlobStoreClient
}
// MigratorClient is used to perform migrations to unified storage
type MigratorClient interface {
resourcepb.BulkStoreClient
GetStats(ctx context.Context, in *resourcepb.ResourceStatsRequest, opts ...grpc.CallOption) (*resourcepb.ResourceStatsResponse, error)
}
// ResourceClient combines all resource-related clients and should be avoided in favor of more specific interfaces when possible
//
//go:generate mockery --name ResourceClient --structname MockResourceClient --inpackage --filename client_mock.go --with-expecter
type ResourceClient interface {
resourcepb.ResourceStoreClient
resourcepb.BulkStoreClient
resourcepb.BlobStoreClient
StorageClient
SearchClient
MigratorClient
resourcepb.DiagnosticsClient
resourcepb.QuotasClient
// SearchClient methods are included for convenience - the client typically needs both
SearchClient
}
// Internal implementation
pkg/storage/unified/resource/search.go
+7
View File
@@ -614,6 +614,13 @@ func (s *searchSupport) Stop(_ context.Context) error {
return nil
}
// IsHealthy implements resourcepb.DiagnosticsServer
func (s *searchSupport) IsHealthy(ctx context.Context, req *resourcepb.HealthCheckRequest) (*resourcepb.HealthCheckResponse, error) {
return &resourcepb.HealthCheckResponse{
Status: resourcepb.HealthCheckResponse_SERVING,
}, nil
}
func (s *searchSupport) init(ctx context.Context) error {
origCtx := ctx
pkg/storage/unified/resource/server.go
+6 -13
View File
@@ -39,6 +39,7 @@ type SearchServer interface {
resourcepb.ResourceIndexServer
resourcepb.ManagedObjectIndexServer
resourcepb.DiagnosticsServer
}
// ResourceServer implements all gRPC services
@@ -227,8 +228,7 @@ type ResourceServerOptions struct {
Blob BlobConfig
// Search options
SearchOptions SearchOptions // TODO: needed?
Search SearchServer
Search SearchServer
// Quota service
OverridesService *OverridesService
@@ -259,6 +259,9 @@ type ResourceServerOptions struct {
// MaxPageSizeBytes is the maximum size of a page in bytes.
MaxPageSizeBytes int
// IndexMinUpdateInterval is the time to wait after a successful write operation to ensure read-after-write consistency in search.
// This config is shared with search
IndexMinUpdateInterval time.Duration
// QOSQueue is the quality of service queue used to enqueue
QOSQueue QOSEnqueuer
@@ -350,9 +353,8 @@ func NewResourceServer(opts ResourceServerOptions) (*server, error) {
queue: opts.QOSQueue,
queueConfig: opts.QOSConfig,
overridesService: opts.OverridesService,
search: opts.Search,
artificialSuccessfulWriteDelay: opts.SearchOptions.IndexMinUpdateInterval,
artificialSuccessfulWriteDelay: opts.IndexMinUpdateInterval,
}
/*
@@ -381,7 +383,6 @@ type server struct {
backend StorageBackend
blob BlobSupport
secure secrets.InlineSecureValueSupport
search SearchServer
diagnostics resourcepb.DiagnosticsServer
access claims.AccessClient
writeHooks WriteAccessHooks
@@ -1522,14 +1523,6 @@ func (s *server) runInQueue(ctx context.Context, tenantID string, runnable func(
}
}
func (s *server) RebuildIndexes(ctx context.Context, req *resourcepb.RebuildIndexesRequest) (*resourcepb.RebuildIndexesResponse, error) {
if s.search == nil {
return nil, fmt.Errorf("search index not configured")
}
return s.search.RebuildIndexes(ctx, req)
}
func (s *server) checkQuota(ctx context.Context, nsr NamespacedResource) {
span := trace.SpanFromContext(ctx)
span.AddEvent("checkQuota", trace.WithAttributes(
pkg/storage/unified/sql/remote_search.go
+13 -6
View File
@@ -16,9 +16,10 @@ var _ resource.SearchServer = (*remoteSearchClient)(nil)
// remoteSearchClient wraps gRPC search clients to implement the SearchServer interface.
// This allows the storage server to delegate search operations to a remote search server.
type remoteSearchClient struct {
conn *grpc.ClientConn
index resourcepb.ResourceIndexClient
moiClient resourcepb.ManagedObjectIndexClient
conn *grpc.ClientConn
index resourcepb.ResourceIndexClient
moiClient resourcepb.ManagedObjectIndexClient
diagnostics resourcepb.DiagnosticsClient
}
// newRemoteSearchClient creates a new remote search client that connects to a search server at the given address.
@@ -36,9 +37,10 @@ func newRemoteSearchClient(address string) (*remoteSearchClient, error) {
}
return &remoteSearchClient{
conn: conn,
index: resourcepb.NewResourceIndexClient(conn),
moiClient: resourcepb.NewManagedObjectIndexClient(conn),
conn: conn,
index: resourcepb.NewResourceIndexClient(conn),
moiClient: resourcepb.NewManagedObjectIndexClient(conn),
diagnostics: resourcepb.NewDiagnosticsClient(conn),
}, nil
}
@@ -81,3 +83,8 @@ func (r *remoteSearchClient) CountManagedObjects(ctx context.Context, req *resou
func (r *remoteSearchClient) ListManagedObjects(ctx context.Context, req *resourcepb.ListManagedObjectsRequest) (*resourcepb.ListManagedObjectsResponse, error) {
return r.moiClient.ListManagedObjects(ctx, req)
}
// IsHealthy implements resourcepb.DiagnosticsServer.
func (r *remoteSearchClient) IsHealthy(ctx context.Context, req *resourcepb.HealthCheckRequest) (*resourcepb.HealthCheckResponse, error) {
return r.diagnostics.IsHealthy(ctx, req)
}
pkg/storage/unified/sql/search_service.go
+18 -38
View File
@@ -35,11 +35,12 @@ import (
)
var (
_ SearchGrpcService = (*searchService)(nil)
_ UnifiedSearchGrpcService = (*searchService)(nil)
)
// SearchGrpcService is the interface for the standalone search gRPC service.
type SearchGrpcService interface {
// UnifiedSearchGrpcService is the interface for the standalone search gRPC service.
// This follows the same naming convention as UnifiedStorageGrpcService.
type UnifiedSearchGrpcService interface {
services.NamedService
// GetAddress returns the address where this service is running
@@ -54,10 +55,10 @@ type searchService struct {
subservicesWatcher *services.FailureWatcher
hasSubservices bool
cfg *setting.Cfg
features featuremgmt.FeatureToggles
db infraDB.DB
stopCh chan struct{}
cfg *setting.Cfg
features featuremgmt.FeatureToggles
db infraDB.DB
stopCh chan struct{}
stoppedCh chan error
handler grpcserver.Provider
@@ -78,9 +79,10 @@ type searchService struct {
backend resource.StorageBackend
}
// ProvideSearchGrpcService creates a standalone search gRPC service.
// ProvideUnifiedSearchGrpcService creates a standalone search gRPC service.
// This is used when running search-server as a separate target.
func ProvideSearchGrpcService(
// It follows the same naming convention as ProvideUnifiedStorageGrpcService.
func ProvideUnifiedSearchGrpcService(
cfg *setting.Cfg,
features featuremgmt.FeatureToggles,
db infraDB.DB,
@@ -91,7 +93,7 @@ func ProvideSearchGrpcService(
searchRing *ring.Ring,
memberlistKVConfig kv.Config,
backend resource.StorageBackend,
) (SearchGrpcService, error) {
) (UnifiedSearchGrpcService, error) {
tracer := otel.Tracer("search-server")
authn := NewAuthenticatorWithFallback(cfg, reg, tracer, func(ctx context.Context) (context.Context, error) {
@@ -227,7 +229,12 @@ func (s *searchService) starting(ctx context.Context) error {
srv := s.handler.GetServer()
resourcepb.RegisterResourceIndexServer(srv, searchServer)
resourcepb.RegisterManagedObjectIndexServer(srv, searchServer)
grpc_health_v1.RegisterHealthServer(srv, &searchHealthService{searchServer: searchServer})
resourcepb.RegisterDiagnosticsServer(srv, searchServer)
healthService, err := resource.ProvideHealthService(searchServer)
if err != nil {
return fmt.Errorf("failed to create health service: %w", err)
}
grpc_health_v1.RegisterHealthServer(srv, healthService)
// register reflection service
_, err = grpcserver.ProvideReflectionService(s.cfg, s.handler)
@@ -324,30 +331,3 @@ func toSearchLifecyclerConfig(cfg *setting.Cfg, logger log.Logger) (ring.BasicLi
NumTokens: resource.RingNumTokens,
}, nil
}
// searchHealthService implements the health check for the search service.
type searchHealthService struct {
searchServer resource.SearchServer
}
func (h *searchHealthService) Check(ctx context.Context, req *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) {
return &grpc_health_v1.HealthCheckResponse{
Status: grpc_health_v1.HealthCheckResponse_SERVING,
}, nil
}
func (h *searchHealthService) Watch(req *grpc_health_v1.HealthCheckRequest, server grpc_health_v1.Health_WatchServer) error {
return fmt.Errorf("watch not implemented")
}
func (h *searchHealthService) List(ctx context.Context, req *grpc_health_v1.HealthListRequest) (*grpc_health_v1.HealthListResponse, error) {
check, err := h.Check(ctx, &grpc_health_v1.HealthCheckRequest{})
if err != nil {
return nil, err
}
return &grpc_health_v1.HealthListResponse{
Statuses: map[string]*grpc_health_v1.HealthCheckResponse{
"": check,
},
}, nil
}
pkg/storage/unified/sql/server.go
+70 -7
View File
@@ -30,6 +30,19 @@ type QOSEnqueueDequeuer interface {
Dequeue(ctx context.Context) (func(), error)
}
// SearchServerOptions contains the options for creating a new SearchServer
type SearchServerOptions struct {
Backend resource.StorageBackend
DB infraDB.DB
Cfg *setting.Cfg
Tracer trace.Tracer
Reg prometheus.Registerer
AccessClient types.AccessClient
SearchOptions resource.SearchOptions
IndexMetrics *resource.BleveIndexMetrics
OwnsIndexFn func(key resource.NamespacedResource) (bool, error)
}
// ServerOptions contains the options for creating a new ResourceServer
type ServerOptions struct {
Backend resource.StorageBackend
@@ -46,6 +59,51 @@ type ServerOptions struct {
QOSQueue QOSEnqueueDequeuer
SecureValues secrets.InlineSecureValueSupport
OwnsIndexFn func(key resource.NamespacedResource) (bool, error)
// Search is an optional pre-created search server. If nil, one will be created.
Search resource.SearchServer
}
// NewSearchServer creates a new SearchServer with the given options.
// This can be used to create a standalone search server or to create a search server
// that will be passed to NewResourceServer.
func NewSearchServer(opts SearchServerOptions) (resource.SearchServer, error) {
backend := opts.Backend
if backend == nil {
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"))
b, err := NewBackend(BackendOptions{
DBProvider: eDB,
Reg: opts.Reg,
IsHA: isHA,
LastImportTimeMaxAge: opts.SearchOptions.MaxIndexAge,
})
if err != nil {
return nil, err
}
// Initialize the backend before creating search server
if err := b.Init(context.Background()); err != nil {
return nil, fmt.Errorf("failed to initialize backend: %w", err)
}
backend = b
}
search, err := resource.NewSearchServer(opts.SearchOptions, backend, opts.AccessClient, nil, opts.IndexMetrics, opts.OwnsIndexFn)
if err != nil {
return nil, fmt.Errorf("failed to create search server: %w", err)
}
if err := search.Init(context.Background()); err != nil {
return nil, fmt.Errorf("failed to initialize search server: %w", err)
}
return search, nil
}
func NewResourceServer(opts ServerOptions) (resource.ResourceServer, resource.SearchServer, error) {
@@ -148,7 +206,7 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, resource.Se
Reg: opts.Reg,
IsHA: isHA,
storageMetrics: opts.StorageMetrics,
LastImportTimeMaxAge: opts.SearchOptions.MaxIndexAge, // No need to keep last_import_times older than max index age.
LastImportTimeMaxAge: opts.Cfg.MaxFileIndexAge, // No need to keep last_import_times older than max index age.
})
if err != nil {
return nil, nil, err
@@ -166,13 +224,18 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, resource.Se
}
}
search, err := resource.NewSearchServer(opts.SearchOptions, serverOptions.Backend, opts.AccessClient, nil, opts.IndexMetrics, opts.OwnsIndexFn)
if err != nil {
return nil, nil, fmt.Errorf("failed to initialize search: %w", err)
}
// Use pre-created search server if provided, otherwise create one
search := opts.Search
if search == nil {
var err error
search, err = resource.NewSearchServer(opts.SearchOptions, serverOptions.Backend, opts.AccessClient, nil, opts.IndexMetrics, opts.OwnsIndexFn)
if err != nil {
return nil, nil, fmt.Errorf("failed to create search server: %w", err)
}
if err := search.Init(context.Background()); err != nil {
return nil, nil, fmt.Errorf("failed to initialize search: %w", err)
if err := search.Init(context.Background()); err != nil {
return nil, nil, fmt.Errorf("failed to initialize search server: %w", err)
}
}
serverOptions.Search = search
pkg/storage/unified/sql/service.go
-9
View File
@@ -36,7 +36,6 @@ import (
"github.com/grafana/grafana/pkg/storage/unified/resource"
"github.com/grafana/grafana/pkg/storage/unified/resource/grpc"
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
"github.com/grafana/grafana/pkg/storage/unified/search"
"github.com/grafana/grafana/pkg/util/scheduler"
)
@@ -261,11 +260,6 @@ func (s *service) starting(ctx context.Context) error {
return err
}
searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.docBuilders, s.indexMetrics, s.OwnsIndex)
if err != nil {
return err
}
serverOptions := ServerOptions{
Backend: s.backend,
DB: s.db,
@@ -273,12 +267,9 @@ func (s *service) starting(ctx context.Context) error {
Tracer: s.tracing,
Reg: s.reg,
AccessClient: authzClient,
SearchOptions: searchOptions,
StorageMetrics: s.storageMetrics,
IndexMetrics: s.indexMetrics,
Features: s.features,
QOSQueue: s.queue,
OwnsIndexFn: s.OwnsIndex,
}
if s.cfg.OverridesFilePath != "" {