From 31a371e385e29b6216b5f3f6afde5a7bde58a49e Mon Sep 17 00:00:00 2001 From: Will Assis <35489495+gassiss@users.noreply.github.com> Date: Thu, 13 Mar 2025 11:09:38 -0300 Subject: [PATCH] fix(unified-storage): remove bleve index metric from global scope (#101825) * refactor grafana_index_server_index_size to calculate in a goroutine instead of at scrape time and remove grafana_index_server_indexed_docs metric * use wire to inject bleve index metrics * remove sprinkles metrics from bleve index metrics * log error when trying to calculate file index size and bump interval to 1m instead of 5s --- .../datamigrations/to_unified_storage.go | 2 +- pkg/server/module_server.go | 10 +- pkg/server/wire.go | 1 + pkg/server/wireexts_oss.go | 1 + pkg/storage/unified/client.go | 9 +- .../unified/resource/bleve_index_metrics.go | 175 ++++-------------- pkg/storage/unified/resource/search.go | 54 +++--- pkg/storage/unified/resource/server.go | 6 +- pkg/storage/unified/search/bleve.go | 67 +++++-- .../unified/search/bleve_search_test.go | 5 +- pkg/storage/unified/search/bleve_test.go | 5 +- pkg/storage/unified/search/options.go | 16 +- pkg/storage/unified/sql/server.go | 3 +- pkg/storage/unified/sql/service.go | 7 +- .../unified/sql/test/integration_test.go | 2 +- pkg/tests/testinfra/testinfra.go | 2 +- 16 files changed, 152 insertions(+), 213 deletions(-) diff --git a/pkg/cmd/grafana-cli/commands/datamigrations/to_unified_storage.go b/pkg/cmd/grafana-cli/commands/datamigrations/to_unified_storage.go index 5d39ee165c1..faae73c2ba8 100644 --- a/pkg/cmd/grafana-cli/commands/datamigrations/to_unified_storage.go +++ b/pkg/cmd/grafana-cli/commands/datamigrations/to_unified_storage.go @@ -196,7 +196,7 @@ func newUnifiedClient(cfg *setting.Cfg, sqlStore db.DB) (resource.ResourceClient Reg: prometheus.NewPedanticRegistry(), Authzc: authlib.FixedAccessClient(true), // always true! Docs: nil, // document supplier - }, nil) + }, nil, nil) } func newParquetClient(file *os.File) (resource.BulkStoreClient, error) { diff --git a/pkg/server/module_server.go b/pkg/server/module_server.go index 2f0e7a1a03c..82fd314f5e1 100644 --- a/pkg/server/module_server.go +++ b/pkg/server/module_server.go @@ -22,8 +22,8 @@ import ( // NewModule returns an instance of a ModuleServer, responsible for managing // dskit modules (services). -func NewModule(opts Options, apiOpts api.ServerOptions, features featuremgmt.FeatureToggles, cfg *setting.Cfg, storageMetrics *resource.StorageMetrics) (*ModuleServer, error) { - s, err := newModuleServer(opts, apiOpts, features, cfg, storageMetrics) +func NewModule(opts Options, apiOpts api.ServerOptions, features featuremgmt.FeatureToggles, cfg *setting.Cfg, storageMetrics *resource.StorageMetrics, indexMetrics *resource.BleveIndexMetrics) (*ModuleServer, error) { + s, err := newModuleServer(opts, apiOpts, features, cfg, storageMetrics, indexMetrics) if err != nil { return nil, err } @@ -35,7 +35,7 @@ func NewModule(opts Options, apiOpts api.ServerOptions, features featuremgmt.Fea return s, nil } -func newModuleServer(opts Options, apiOpts api.ServerOptions, features featuremgmt.FeatureToggles, cfg *setting.Cfg, storageMetrics *resource.StorageMetrics) (*ModuleServer, error) { +func newModuleServer(opts Options, apiOpts api.ServerOptions, features featuremgmt.FeatureToggles, cfg *setting.Cfg, storageMetrics *resource.StorageMetrics, indexMetrics *resource.BleveIndexMetrics) (*ModuleServer, error) { rootCtx, shutdownFn := context.WithCancel(context.Background()) s := &ModuleServer{ @@ -52,6 +52,7 @@ func newModuleServer(opts Options, apiOpts api.ServerOptions, features featuremg commit: opts.Commit, buildBranch: opts.BuildBranch, storageMetrics: storageMetrics, + indexMetrics: indexMetrics, } return s, nil @@ -74,6 +75,7 @@ type ModuleServer struct { isInitialized bool mtx sync.Mutex storageMetrics *resource.StorageMetrics + indexMetrics *resource.BleveIndexMetrics pidFile string version string @@ -138,7 +140,7 @@ func (s *ModuleServer) Run() error { if err != nil { return nil, err } - return sql.ProvideUnifiedStorageGrpcService(s.cfg, s.features, nil, s.log, nil, docBuilders, s.storageMetrics) + return sql.ProvideUnifiedStorageGrpcService(s.cfg, s.features, nil, s.log, nil, docBuilders, s.storageMetrics, s.indexMetrics) }) m.RegisterModule(modules.ZanzanaServer, func() (services.Service, error) { diff --git a/pkg/server/wire.go b/pkg/server/wire.go index c24a40e80d4..e211b739ecb 100644 --- a/pkg/server/wire.go +++ b/pkg/server/wire.go @@ -404,6 +404,7 @@ var wireBasicSet = wire.NewSet( authz.WireSet, // Unified storage resource.ProvideStorageMetrics, + resource.ProvideIndexMetrics, // Kubernetes API server grafanaapiserver.WireSet, apiregistry.WireSet, diff --git a/pkg/server/wireexts_oss.go b/pkg/server/wireexts_oss.go index 75e6f6842fb..77ff00b960c 100644 --- a/pkg/server/wireexts_oss.go +++ b/pkg/server/wireexts_oss.go @@ -159,6 +159,7 @@ var wireExtsModuleServerSet = wire.NewSet( wireExtsBaseCLISet, // Unified storage resource.ProvideStorageMetrics, + resource.ProvideIndexMetrics, ) var wireExtsStandaloneAPIServerSet = wire.NewSet( diff --git a/pkg/storage/unified/client.go b/pkg/storage/unified/client.go index 75c6cbdf9fa..3eebaffeac8 100644 --- a/pkg/storage/unified/client.go +++ b/pkg/storage/unified/client.go @@ -51,7 +51,7 @@ type clientMetrics struct { } // This adds a UnifiedStorage client into the wire dependency tree -func ProvideUnifiedStorageClient(opts *Options, storageMetrics *resource.StorageMetrics) (resource.ResourceClient, error) { +func ProvideUnifiedStorageClient(opts *Options, storageMetrics *resource.StorageMetrics, indexMetrics *resource.BleveIndexMetrics) (resource.ResourceClient, error) { // See: apiserver.ApplyGrafanaConfig(cfg, features, o) apiserverCfg := opts.Cfg.SectionWithEnvOverrides("grafana-apiserver") client, err := newClient(options.StorageOptions{ @@ -59,7 +59,7 @@ func ProvideUnifiedStorageClient(opts *Options, storageMetrics *resource.Storage DataPath: apiserverCfg.Key("storage_path").MustString(filepath.Join(opts.Cfg.DataPath, "grafana-apiserver")), Address: apiserverCfg.Key("address").MustString(""), // client address BlobStoreURL: apiserverCfg.Key("blob_url").MustString(""), - }, opts.Cfg, opts.Features, opts.DB, opts.Tracer, opts.Reg, opts.Authzc, opts.Docs, storageMetrics) + }, opts.Cfg, opts.Features, opts.DB, opts.Tracer, opts.Reg, opts.Authzc, opts.Docs, storageMetrics, indexMetrics) if err == nil { // Used to get the folder stats client = federated.NewFederatedClient( @@ -80,6 +80,7 @@ func newClient(opts options.StorageOptions, authzc types.AccessClient, docs resource.DocumentBuilderSupplier, storageMetrics *resource.StorageMetrics, + indexMetrics *resource.BleveIndexMetrics, ) (resource.ResourceClient, error) { ctx := context.Background() switch opts.StorageType { @@ -131,11 +132,11 @@ func newClient(opts options.StorageOptions, // Use the local SQL default: - searchOptions, err := search.NewSearchOptions(features, cfg, tracer, docs, reg) + searchOptions, err := search.NewSearchOptions(features, cfg, tracer, docs, indexMetrics) if err != nil { return nil, err } - server, err := sql.NewResourceServer(db, cfg, tracer, reg, authzc, searchOptions, storageMetrics) + server, err := sql.NewResourceServer(db, cfg, tracer, reg, authzc, searchOptions, storageMetrics, indexMetrics) if err != nil { return nil, err } diff --git a/pkg/storage/unified/resource/bleve_index_metrics.go b/pkg/storage/unified/resource/bleve_index_metrics.go index a15f36be543..310886666cc 100644 --- a/pkg/storage/unified/resource/bleve_index_metrics.go +++ b/pkg/storage/unified/resource/bleve_index_metrics.go @@ -1,160 +1,57 @@ package resource import ( - "os" - "path/filepath" - "sync" "time" "github.com/grafana/dskit/instrument" "github.com/prometheus/client_golang/prometheus" -) - -var ( - onceIndex sync.Once - onceSprinkles sync.Once - IndexMetrics *BleveIndexMetrics - SprinklesIndexMetrics *SprinklesMetrics + "github.com/prometheus/client_golang/prometheus/promauto" ) type BleveIndexMetrics struct { - IndexDir string - Backend SearchBackend - - // metrics IndexLatency *prometheus.HistogramVec IndexSize prometheus.Gauge - IndexedDocs prometheus.Gauge IndexedKinds *prometheus.GaugeVec IndexCreationTime *prometheus.HistogramVec IndexTenants *prometheus.CounterVec } -type SprinklesMetrics struct { - SprinklesLatency prometheus.Histogram -} - var IndexCreationBuckets = []float64{1, 5, 10, 25, 50, 75, 100, 200, 300, 400, 500, 600, 700, 800, 900, 1000} -func NewSprinklesMetrics() *SprinklesMetrics { - onceSprinkles.Do(func() { - SprinklesIndexMetrics = &SprinklesMetrics{ - SprinklesLatency: prometheus.NewHistogram(prometheus.HistogramOpts{ - Namespace: "index_server", - Name: "sprinkles_latency_seconds", - Help: "Time (in seconds) it takes until sprinkles are fetched", - Buckets: instrument.DefBuckets, - NativeHistogramBucketFactor: 1.1, // enable native histograms - NativeHistogramMaxBucketNumber: 160, - NativeHistogramMinResetDuration: time.Hour, - }), - } - }) - - return SprinklesIndexMetrics -} - -func NewIndexMetrics(indexDir string, searchBackend SearchBackend) *BleveIndexMetrics { - onceIndex.Do(func() { - IndexMetrics = &BleveIndexMetrics{ - IndexDir: indexDir, - Backend: searchBackend, - IndexLatency: prometheus.NewHistogramVec(prometheus.HistogramOpts{ - Namespace: "index_server", - Name: "index_latency_seconds", - Help: "Time (in seconds) until index is updated with new event", - Buckets: instrument.DefBuckets, - NativeHistogramBucketFactor: 1.1, // enable native histograms - NativeHistogramMaxBucketNumber: 160, - NativeHistogramMinResetDuration: time.Hour, - }, []string{"resource"}), - IndexSize: prometheus.NewGauge(prometheus.GaugeOpts{ - Namespace: "index_server", - Name: "index_size", - Help: "Size of the index in bytes - only for file-based indices", - }), - IndexedDocs: prometheus.NewGauge(prometheus.GaugeOpts{ - Namespace: "index_server", - Name: "indexed_docs", - Help: "Number of indexed documents by resource", - }), - IndexedKinds: prometheus.NewGaugeVec(prometheus.GaugeOpts{ - Namespace: "index_server", - Name: "indexed_kinds", - Help: "Number of indexed documents by kind", - }, []string{"kind"}), - IndexCreationTime: prometheus.NewHistogramVec(prometheus.HistogramOpts{ - Namespace: "index_server", - Name: "index_creation_time_seconds", - Help: "Time (in seconds) it takes until index is created", - Buckets: IndexCreationBuckets, - NativeHistogramBucketFactor: 1.1, // enable native histograms - NativeHistogramMaxBucketNumber: 160, - NativeHistogramMinResetDuration: time.Hour, - }, []string{}), - IndexTenants: prometheus.NewCounterVec(prometheus.CounterOpts{ - Namespace: "index_server", - Name: "index_tenants", - Help: "Number of tenants in the index", - }, []string{"index_storage"}), // index_storage is either "file" or "memory" - } - }) - - return IndexMetrics -} - -func (s *SprinklesMetrics) Collect(ch chan<- prometheus.Metric) { - s.SprinklesLatency.Collect(ch) -} - -func (s *SprinklesMetrics) Describe(ch chan<- *prometheus.Desc) { - s.SprinklesLatency.Describe(ch) -} - -func (s *BleveIndexMetrics) Collect(ch chan<- prometheus.Metric) { - s.IndexLatency.Collect(ch) - s.IndexCreationTime.Collect(ch) - s.IndexedKinds.Collect(ch) - s.IndexTenants.Collect(ch) - - // collect index size - totalSize, err := getTotalIndexSize(s.IndexDir) - if err == nil { - s.IndexSize.Set(float64(totalSize)) - s.IndexSize.Collect(ch) +func ProvideIndexMetrics(reg prometheus.Registerer) *BleveIndexMetrics { + return &BleveIndexMetrics{ + IndexLatency: promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "index_server", + Name: "index_latency_seconds", + Help: "Time (in seconds) until index is updated with new event", + Buckets: instrument.DefBuckets, + NativeHistogramBucketFactor: 1.1, // enable native histograms + NativeHistogramMaxBucketNumber: 160, + NativeHistogramMinResetDuration: time.Hour, + }, []string{"resource"}), + IndexSize: promauto.With(reg).NewGauge(prometheus.GaugeOpts{ + Namespace: "index_server", + Name: "index_size", + Help: "Size of the index in bytes - only for file-based indices", + }), + IndexedKinds: promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{ + Namespace: "index_server", + Name: "indexed_kinds", + Help: "Number of indexed documents by kind", + }, []string{"kind"}), + IndexCreationTime: promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "index_server", + Name: "index_creation_time_seconds", + Help: "Time (in seconds) it takes until index is created", + Buckets: IndexCreationBuckets, + NativeHistogramBucketFactor: 1.1, // enable native histograms + NativeHistogramMaxBucketNumber: 160, + NativeHistogramMinResetDuration: time.Hour, + }, []string{}), + IndexTenants: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Namespace: "index_server", + Name: "index_tenants", + Help: "Number of tenants in the index", + }, []string{"index_storage"}), // index_storage is either "file" or "memory" } - - // collect index docs - s.IndexedDocs.Set(float64(s.Backend.TotalDocs())) - s.IndexedDocs.Collect(ch) -} - -func (s *BleveIndexMetrics) Describe(ch chan<- *prometheus.Desc) { - s.IndexLatency.Describe(ch) - s.IndexSize.Describe(ch) - s.IndexedDocs.Describe(ch) - s.IndexedKinds.Describe(ch) - s.IndexCreationTime.Describe(ch) - s.IndexTenants.Describe(ch) -} - -// getTotalIndexSize returns the total size of all file-based indices. -func getTotalIndexSize(dir string) (int64, error) { - var totalSize int64 - - err := filepath.WalkDir(dir, func(path string, info os.DirEntry, err error) error { - if err != nil { - return err - } - if !info.IsDir() { - fileInfo, err := info.Info() - if err != nil { - return err - } - totalSize += fileInfo.Size() - } - return nil - }) - - return totalSize, err } diff --git a/pkg/storage/unified/resource/search.go b/pkg/storage/unified/resource/search.go index 63acc10be7c..196b33e2492 100644 --- a/pkg/storage/unified/resource/search.go +++ b/pkg/storage/unified/resource/search.go @@ -88,14 +88,15 @@ const tracingPrexfixSearch = "unified_search." // This supports indexing+search regardless of implementation type searchSupport struct { - tracer trace.Tracer - log *slog.Logger - storage StorageBackend - search SearchBackend - access types.AccessClient - builders *builderCache - initWorkers int - initMinSize int + tracer trace.Tracer + log *slog.Logger + storage StorageBackend + search SearchBackend + indexMetrics *BleveIndexMetrics + access types.AccessClient + builders *builderCache + initWorkers int + initMinSize int } var ( @@ -103,7 +104,7 @@ var ( _ ManagedObjectIndexServer = (*searchSupport)(nil) ) -func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.AccessClient, blob BlobSupport, tracer trace.Tracer) (support *searchSupport, err error) { +func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.AccessClient, blob BlobSupport, tracer trace.Tracer, indexMetrics *BleveIndexMetrics) (support *searchSupport, err error) { // No backend search support if opts.Backend == nil { return nil, nil @@ -117,13 +118,14 @@ func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.A } support = &searchSupport{ - access: access, - tracer: tracer, - storage: storage, - search: opts.Backend, - log: slog.Default().With("logger", "resource-search"), - initWorkers: opts.WorkerThreads, - initMinSize: opts.InitMinCount, + access: access, + tracer: tracer, + storage: storage, + search: opts.Backend, + log: slog.Default().With("logger", "resource-search"), + initWorkers: opts.WorkerThreads, + initMinSize: opts.InitMinCount, + indexMetrics: indexMetrics, } info, err := opts.Resources.GetDocumentBuilders() @@ -399,8 +401,8 @@ func (s *searchSupport) init(ctx context.Context) error { end := time.Now().Unix() s.log.Info("search index initialized", "duration_secs", end-start, "total_docs", s.search.TotalDocs()) - if IndexMetrics != nil { - IndexMetrics.IndexCreationTime.WithLabelValues().Observe(float64(end - start)) + if s.indexMetrics != nil { + s.indexMetrics.IndexCreationTime.WithLabelValues().Observe(float64(end - start)) } return nil @@ -457,8 +459,8 @@ func (s *searchSupport) handleEvent(ctx context.Context, evt *WrittenEvent) { s.log.Warn("error writing document watch event", "error", err) return } - if evt.Type == WatchEvent_ADDED { - IndexMetrics.IndexedKinds.WithLabelValues(evt.Key.Resource).Inc() + if evt.Type == WatchEvent_ADDED && s.indexMetrics != nil { + s.indexMetrics.IndexedKinds.WithLabelValues(evt.Key.Resource).Inc() } case WatchEvent_DELETED: _, deleteSpan := s.tracer.Start(ctx, tracingPrexfixSearch+"DeleteDocument") @@ -468,7 +470,9 @@ func (s *searchSupport) handleEvent(ctx context.Context, evt *WrittenEvent) { s.log.Warn("error deleting document watch event", "error", err) return } - IndexMetrics.IndexedKinds.WithLabelValues(evt.Key.Resource).Dec() + if s.indexMetrics != nil { + s.indexMetrics.IndexedKinds.WithLabelValues(evt.Key.Resource).Dec() + } default: // do nothing s.log.Warn("unknown watch event", "type", evt.Type) @@ -481,8 +485,8 @@ func (s *searchSupport) handleEvent(ctx context.Context, evt *WrittenEvent) { if latencySeconds > 1 { s.log.Warn("high index latency object details", "resource", evt.Key.Resource, "latency_seconds", latencySeconds, "name", evt.Key.Name, "namespace", evt.Key.Namespace, "rv", evt.ResourceVersion) } - if IndexMetrics != nil { - IndexMetrics.IndexLatency.WithLabelValues(evt.Key.Resource).Observe(latencySeconds) + if s.indexMetrics != nil { + s.indexMetrics.IndexLatency.WithLabelValues(evt.Key.Resource).Observe(latencySeconds) } } @@ -573,8 +577,8 @@ func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size if err != nil { s.log.Warn("error getting doc count", "error", err) } - if IndexMetrics != nil { - IndexMetrics.IndexedKinds.WithLabelValues(key.Resource).Add(float64(docCount)) + if s.indexMetrics != nil { + s.indexMetrics.IndexedKinds.WithLabelValues(key.Resource).Add(float64(docCount)) } if err == nil { diff --git a/pkg/storage/unified/resource/server.go b/pkg/storage/unified/resource/server.go index 0acc0dd2520..0a983de75aa 100644 --- a/pkg/storage/unified/resource/server.go +++ b/pkg/storage/unified/resource/server.go @@ -184,6 +184,8 @@ type ResourceServerOptions struct { Reg prometheus.Registerer storageMetrics *StorageMetrics + + IndexMetrics *BleveIndexMetrics } func NewResourceServer(opts ResourceServerOptions) (ResourceServer, error) { @@ -248,11 +250,12 @@ func NewResourceServer(opts ResourceServerOptions) (ResourceServer, error) { ctx: ctx, cancel: cancel, storageMetrics: opts.storageMetrics, + indexMetrics: opts.IndexMetrics, } if opts.Search.Resources != nil { var err error - s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.Tracer) + s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.Tracer, opts.IndexMetrics) if err != nil { return nil, err } @@ -282,6 +285,7 @@ type server struct { now func() int64 mostRecentRV atomic.Int64 // The most recent resource version seen by the server storageMetrics *StorageMetrics + indexMetrics *BleveIndexMetrics // Background watch task -- this has permissions for everything ctx context.Context diff --git a/pkg/storage/unified/search/bleve.go b/pkg/storage/unified/search/bleve.go index 322abff0516..4976449f548 100644 --- a/pkg/storage/unified/search/bleve.go +++ b/pkg/storage/unified/search/bleve.go @@ -59,10 +59,11 @@ type bleveBackend struct { cache map[resource.NamespacedResource]*bleveIndex cacheMu sync.RWMutex - features featuremgmt.FeatureToggles + features featuremgmt.FeatureToggles + indexMetrics *resource.BleveIndexMetrics } -func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, features featuremgmt.FeatureToggles) (*bleveBackend, error) { +func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, features featuremgmt.FeatureToggles, indexMetrics *resource.BleveIndexMetrics) (*bleveBackend, error) { if opts.Root == "" { return nil, fmt.Errorf("bleve backend missing root folder configuration") } @@ -74,14 +75,19 @@ func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, features featuremgm return nil, fmt.Errorf("bleve root is configured against a file (not folder)") } - return &bleveBackend{ - log: slog.Default().With("logger", "bleve-backend"), - tracer: tracer, - cache: make(map[resource.NamespacedResource]*bleveIndex), - opts: opts, - start: time.Now(), - features: features, - }, nil + bleveBackend := &bleveBackend{ + log: slog.Default().With("logger", "bleve-backend"), + tracer: tracer, + cache: make(map[resource.NamespacedResource]*bleveIndex), + opts: opts, + start: time.Now(), + features: features, + indexMetrics: indexMetrics, + } + + go bleveBackend.updateIndexSizeMetric(opts.Root) + + return bleveBackend, nil } // This will return nil if the key does not exist @@ -96,6 +102,39 @@ func (b *bleveBackend) GetIndex(ctx context.Context, key resource.NamespacedReso return nil, nil } +// updateIndexSizeMetric sets the total size of all file-based indices metric. +func (b *bleveBackend) updateIndexSizeMetric(indexPath string) { + if b.indexMetrics == nil { + return + } + + for { + var totalSize int64 + + err := filepath.WalkDir(indexPath, func(path string, info os.DirEntry, err error) error { + if err != nil { + return err + } + if !info.IsDir() { + fileInfo, err := info.Info() + if err != nil { + return err + } + totalSize += fileInfo.Size() + } + return nil + }) + + if err == nil { + b.indexMetrics.IndexSize.Set(float64(totalSize)) + } else { + b.log.Error("got error while trying to calculate bleve file index size", "error", err) + } + + time.Sleep(60 * time.Second) + } +} + // Build an index from scratch func (b *bleveBackend) BuildIndex(ctx context.Context, key resource.NamespacedResource, @@ -166,10 +205,14 @@ func (b *bleveBackend) BuildIndex(ctx context.Context, if index != nil && err == nil { go b.cleanOldIndexes(resourceDir, fname) } - resource.IndexMetrics.IndexTenants.WithLabelValues("file").Inc() + if b.indexMetrics != nil { + b.indexMetrics.IndexTenants.WithLabelValues("file").Inc() + } } else { index, err = bleve.NewMemOnly(mapper) - resource.IndexMetrics.IndexTenants.WithLabelValues("memory").Inc() + if b.indexMetrics != nil { + b.indexMetrics.IndexTenants.WithLabelValues("memory").Inc() + } } if err != nil { return nil, err diff --git a/pkg/storage/unified/search/bleve_search_test.go b/pkg/storage/unified/search/bleve_search_test.go index 39eb0638995..29287de97e4 100644 --- a/pkg/storage/unified/search/bleve_search_test.go +++ b/pkg/storage/unified/search/bleve_search_test.go @@ -335,12 +335,9 @@ func newTestDashboardsIndex(t *testing.T) resource.ResourceIndex { backend, err := NewBleveBackend(BleveOptions{ Root: tmpdir, FileThreshold: 9999, // use in-memory for tests - }, tracing.NewNoopTracerService(), featuremgmt.WithFeatures(featuremgmt.FlagUnifiedStorageSearchPermissionFiltering)) + }, tracing.NewNoopTracerService(), featuremgmt.WithFeatures(featuremgmt.FlagUnifiedStorageSearchPermissionFiltering), nil) require.NoError(t, err) - // AVOID NPE in test - resource.NewIndexMetrics(backend.opts.Root, backend) - rv := int64(10) ctx := identity.WithRequester(context.Background(), &user.SignedInUser{Namespace: "ns"}) diff --git a/pkg/storage/unified/search/bleve_test.go b/pkg/storage/unified/search/bleve_test.go index f1d6fec51b6..204a4a07cbd 100644 --- a/pkg/storage/unified/search/bleve_test.go +++ b/pkg/storage/unified/search/bleve_test.go @@ -39,12 +39,9 @@ func TestBleveBackend(t *testing.T) { backend, err := NewBleveBackend(BleveOptions{ Root: tmpdir, FileThreshold: 5, // with more than 5 items we create a file on disk - }, tracing.NewNoopTracerService(), featuremgmt.WithFeatures(featuremgmt.FlagUnifiedStorageSearchPermissionFiltering)) + }, tracing.NewNoopTracerService(), featuremgmt.WithFeatures(featuremgmt.FlagUnifiedStorageSearchPermissionFiltering), nil) require.NoError(t, err) - // AVOID NPE in test - resource.NewIndexMetrics(backend.opts.Root, backend) - rv := int64(10) ctx := identity.WithRequester(context.Background(), &user.SignedInUser{Namespace: "ns"}) var dashboardsIndex resource.ResourceIndex diff --git a/pkg/storage/unified/search/options.go b/pkg/storage/unified/search/options.go index 13a2612319d..f3103e3c94e 100644 --- a/pkg/storage/unified/search/options.go +++ b/pkg/storage/unified/search/options.go @@ -1,19 +1,16 @@ package search import ( - "log/slog" "os" "path/filepath" - "github.com/prometheus/client_golang/prometheus" - "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/services/featuremgmt" "github.com/grafana/grafana/pkg/setting" "github.com/grafana/grafana/pkg/storage/unified/resource" ) -func NewSearchOptions(features featuremgmt.FeatureToggles, cfg *setting.Cfg, tracer tracing.Tracer, docs resource.DocumentBuilderSupplier, reg prometheus.Registerer) (resource.SearchOptions, error) { +func NewSearchOptions(features featuremgmt.FeatureToggles, cfg *setting.Cfg, tracer tracing.Tracer, docs resource.DocumentBuilderSupplier, indexMetrics *resource.BleveIndexMetrics) (resource.SearchOptions, error) { // Setup the search server if features.IsEnabledGlobally(featuremgmt.FlagUnifiedStorageSearch) { root := cfg.IndexPath @@ -28,21 +25,12 @@ func NewSearchOptions(features featuremgmt.FeatureToggles, cfg *setting.Cfg, tra Root: root, FileThreshold: int64(cfg.IndexFileThreshold), // fewer than X items will use a memory index BatchSize: cfg.IndexMaxBatchSize, // This is the batch size for how many objects to add to the index at once - }, tracer, features) + }, tracer, features, indexMetrics) if err != nil { return resource.SearchOptions{}, err } - err = reg.Register(resource.NewIndexMetrics(cfg.IndexPath, bleve)) - if err != nil { - slog.Warn("Failed to register indexer metrics", "error", err) - } - err = reg.Register(resource.NewSprinklesMetrics()) - if err != nil { - slog.Warn("Failed to register sprinkles metrics", "error", err) - } - return resource.SearchOptions{ Backend: bleve, Resources: docs, diff --git a/pkg/storage/unified/sql/server.go b/pkg/storage/unified/sql/server.go index 69738f96f21..330e265d9f7 100644 --- a/pkg/storage/unified/sql/server.go +++ b/pkg/storage/unified/sql/server.go @@ -18,7 +18,7 @@ import ( // Creates a new ResourceServer func NewResourceServer(db infraDB.DB, cfg *setting.Cfg, - tracer tracing.Tracer, reg prometheus.Registerer, ac types.AccessClient, searchOptions resource.SearchOptions, storageMetrics *resource.StorageMetrics) (resource.ResourceServer, error) { + tracer tracing.Tracer, reg prometheus.Registerer, ac types.AccessClient, searchOptions resource.SearchOptions, storageMetrics *resource.StorageMetrics, indexMetrics *resource.BleveIndexMetrics) (resource.ResourceServer, error) { apiserverCfg := cfg.SectionWithEnvOverrides("grafana-apiserver") opts := resource.ResourceServerOptions{ Tracer: tracer, @@ -55,6 +55,7 @@ func NewResourceServer(db infraDB.DB, cfg *setting.Cfg, opts.Diagnostics = store opts.Lifecycle = store opts.Search = searchOptions + opts.IndexMetrics = indexMetrics rs, err := resource.NewResourceServer(opts) if err != nil { diff --git a/pkg/storage/unified/sql/service.go b/pkg/storage/unified/sql/service.go index e6ad6cd6096..6b4e3dc2a87 100644 --- a/pkg/storage/unified/sql/service.go +++ b/pkg/storage/unified/sql/service.go @@ -52,6 +52,7 @@ type service struct { log log.Logger reg prometheus.Registerer storageMetrics *resource.StorageMetrics + indexMetrics *resource.BleveIndexMetrics docBuilders resource.DocumentBuilderSupplier } @@ -64,6 +65,7 @@ func ProvideUnifiedStorageGrpcService( reg prometheus.Registerer, docBuilders resource.DocumentBuilderSupplier, storageMetrics *resource.StorageMetrics, + indexMetrics *resource.BleveIndexMetrics, ) (UnifiedStorageGrpcService, error) { tracingCfg, err := tracing.ProvideTracingConfig(cfg) if err != nil { @@ -96,6 +98,7 @@ func ProvideUnifiedStorageGrpcService( reg: reg, docBuilders: docBuilders, storageMetrics: storageMetrics, + indexMetrics: indexMetrics, } // This will be used when running as a dskit service @@ -110,12 +113,12 @@ func (s *service) start(ctx context.Context) error { return err } - searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.tracing, s.docBuilders, s.reg) + searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.tracing, s.docBuilders, s.indexMetrics) if err != nil { return err } - server, err := NewResourceServer(s.db, s.cfg, s.tracing, s.reg, authzClient, searchOptions, s.storageMetrics) + server, err := NewResourceServer(s.db, s.cfg, s.tracing, s.reg, authzClient, searchOptions, s.storageMetrics, s.indexMetrics) if err != nil { return err } diff --git a/pkg/storage/unified/sql/test/integration_test.go b/pkg/storage/unified/sql/test/integration_test.go index 64164300dea..cf62ab0b0d7 100644 --- a/pkg/storage/unified/sql/test/integration_test.go +++ b/pkg/storage/unified/sql/test/integration_test.go @@ -83,7 +83,7 @@ func TestClientServer(t *testing.T) { features := featuremgmt.WithFeatures() - svc, err := sql.ProvideUnifiedStorageGrpcService(cfg, features, dbstore, nil, prometheus.NewPedanticRegistry(), nil, nil) + svc, err := sql.ProvideUnifiedStorageGrpcService(cfg, features, dbstore, nil, prometheus.NewPedanticRegistry(), nil, nil, nil) require.NoError(t, err) var client resource.ResourceStoreClient diff --git a/pkg/tests/testinfra/testinfra.go b/pkg/tests/testinfra/testinfra.go index 118cb83b7de..d8c412f9618 100644 --- a/pkg/tests/testinfra/testinfra.go +++ b/pkg/tests/testinfra/testinfra.go @@ -104,7 +104,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) + env.Cfg.Logger, prometheus.NewPedanticRegistry(), nil, nil, nil) require.NoError(t, err) ctx := context.Background() err = storage.StartAsync(ctx)