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

fix(unified-storage): remove bleve index metric from global scope (#101825)

Browse Source 
* 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
This commit is contained in:
Will Assis
2025-03-13 10:09:38 -04:00
committed by GitHub
parent a9634f9b12
commit 31a371e385
16 changed files with 152 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/cmd/grafana-cli/commands/datamigrations/to_unified_storage.go
+1 -1
View File
@@ -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) {
pkg/server/module_server.go
+6 -4
View File
@@ -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) {
pkg/server/wire.go
+1
View File
@@ -404,6 +404,7 @@ var wireBasicSet = wire.NewSet(
authz.WireSet,
// Unified storage
resource.ProvideStorageMetrics,
resource.ProvideIndexMetrics,
// Kubernetes API server
grafanaapiserver.WireSet,
apiregistry.WireSet,
pkg/server/wireexts_oss.go
+1
View File
@@ -159,6 +159,7 @@ var wireExtsModuleServerSet = wire.NewSet(
wireExtsBaseCLISet,
// Unified storage
resource.ProvideStorageMetrics,
resource.ProvideIndexMetrics,
)
var wireExtsStandaloneAPIServerSet = wire.NewSet(
pkg/storage/unified/client.go
+5 -4
View File
@@ -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
}
pkg/storage/unified/resource/bleve_index_metrics.go
+36 -139
View File
@@ -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
}
pkg/storage/unified/resource/search.go
+29 -25
View File
@@ -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 {
pkg/storage/unified/resource/server.go
+5 -1
View File
@@ -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
pkg/storage/unified/search/bleve.go
+55 -12
View File
@@ -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
pkg/storage/unified/search/bleve_search_test.go
+1 -4
View File
@@ -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"})
pkg/storage/unified/search/bleve_test.go
+1 -4
View File
@@ -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
pkg/storage/unified/search/options.go
+2 -14
View File
@@ -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,
pkg/storage/unified/sql/server.go
+2 -1
View File
@@ -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 {
pkg/storage/unified/sql/service.go
+5 -2
View File
@@ -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
}
pkg/storage/unified/sql/test/integration_test.go
+1 -1
View File
@@ -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
pkg/tests/testinfra/testinfra.go
+1 -1
View File
@@ -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)