From 70c9c3889f5587e0dc2911293bcb9614a4ddbd5b Mon Sep 17 00:00:00 2001 From: owensmallwood Date: Wed, 4 Dec 2024 15:02:40 -0600 Subject: [PATCH] Unified Storage Indexer: Add back metrics (#97310) * Adds back indexer metrics. Uses config values instead of hardcoded ones. * cast to int64 * remove unused func * Index metrics impl doesn't depend on Bleve. Adds a TotalDocs func to SearchBackend interface. * adds config setting for index_min_count * rename arg * rename metric label to namespace instead of slug * adds default "do nothing" case to satisfy linter * moves bleve index metrics to search package * make bleve backend private, dont need to pass in prom reg * imports * adds bleve metrics to resource package to avoid circular deps --- pkg/setting/setting.go | 11 +- pkg/setting/setting_unified_storage.go | 3 +- .../unified/resource/bleve_index_metrics.go | 128 ++++++++++++++++++ pkg/storage/unified/resource/search.go | 61 ++++++++- pkg/storage/unified/search/bleve.go | 27 +++- pkg/storage/unified/search/bleve_test.go | 12 +- pkg/storage/unified/sql/server.go | 21 +-- 7 files changed, 231 insertions(+), 32 deletions(-) create mode 100644 pkg/storage/unified/resource/bleve_index_metrics.go diff --git a/pkg/setting/setting.go b/pkg/setting/setting.go index bc26b0f7263..8527c1e9df3 100644 --- a/pkg/setting/setting.go +++ b/pkg/setting/setting.go @@ -527,11 +527,12 @@ type Cfg struct { ShortLinkExpiration int // Unified Storage - UnifiedStorage map[string]UnifiedStorageConfig - IndexPath string - IndexWorkers int - IndexMaxBatchSize int - IndexListLimit int + UnifiedStorage map[string]UnifiedStorageConfig + IndexPath string + IndexWorkers int + IndexMaxBatchSize int + IndexFileThreshold int + IndexMinCount int } type UnifiedStorageConfig struct { diff --git a/pkg/setting/setting_unified_storage.go b/pkg/setting/setting_unified_storage.go index 15468fa0c6c..51040160743 100644 --- a/pkg/setting/setting_unified_storage.go +++ b/pkg/setting/setting_unified_storage.go @@ -41,5 +41,6 @@ func (cfg *Cfg) setUnifiedStorageConfig() { cfg.IndexPath = section.Key("index_path").String() cfg.IndexWorkers = section.Key("index_workers").MustInt(10) cfg.IndexMaxBatchSize = section.Key("index_max_batch_size").MustInt(100) - cfg.IndexListLimit = section.Key("index_list_limit").MustInt(1000) + cfg.IndexFileThreshold = section.Key("index_file_threshold").MustInt(10) + cfg.IndexMinCount = section.Key("index_min_count").MustInt(1) } diff --git a/pkg/storage/unified/resource/bleve_index_metrics.go b/pkg/storage/unified/resource/bleve_index_metrics.go new file mode 100644 index 00000000000..c6a96e5dd6d --- /dev/null +++ b/pkg/storage/unified/resource/bleve_index_metrics.go @@ -0,0 +1,128 @@ +package resource + +import ( + "os" + "path/filepath" + "sync" + "time" + + "github.com/grafana/dskit/instrument" + "github.com/prometheus/client_golang/prometheus" +) + +var ( + onceIndex sync.Once + IndexMetrics *BleveIndexMetrics +) + +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 +} + +var IndexCreationBuckets = []float64{1, 5, 10, 25, 50, 75, 100, 200, 300, 400, 500, 600, 700, 800, 900, 1000} + +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{"namespace", "index_storage"}), // index_storage is either "file" or "memory" + } + }) + + return IndexMetrics +} + +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) + } + + // 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 c37f5124fd1..e1f2ed8cb22 100644 --- a/pkg/storage/unified/resource/search.go +++ b/pkg/storage/unified/resource/search.go @@ -4,9 +4,11 @@ import ( "context" "fmt" "log/slog" + "slices" "sync" "time" + "github.com/grafana/grafana/pkg/cmd/grafana-cli/logger" "github.com/hashicorp/golang-lru/v2/expirable" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" @@ -44,6 +46,9 @@ type ResourceIndex interface { // Execute an origin query -- access control is not not checked for each item // NOTE: this will likely be used for provisioning, or it will be removed Origin(ctx context.Context, req *OriginRequest) (*OriginResponse, error) + + // Get the number of documents in the index + DocCount() (int, error) } // SearchBackend contains the technology specific logic to support search @@ -68,6 +73,9 @@ type SearchBackend interface { // The builder will write all documents before returning builder func(index ResourceIndex) (int64, error), ) (ResourceIndex, error) + + // Gets the total number of documents across all indexes + TotalDocs() int64 } const tracingPrexfixSearch = "unified_search." @@ -165,6 +173,7 @@ func (s *searchSupport) Search(ctx context.Context, req *ResourceSearchRequest) func (s *searchSupport) init(ctx context.Context) error { _, span := s.tracer.Start(ctx, tracingPrexfixSearch+"Init") defer span.End() + start := time.Now().Unix() totalBatchesIndexed := 0 group := errgroup.Group{} @@ -204,11 +213,21 @@ func (s *searchSupport) init(ctx context.Context) error { } }() + end := time.Now().Unix() + if IndexMetrics != nil { + IndexMetrics.IndexCreationTime.WithLabelValues().Observe(float64(end - start)) + } + return nil } // Async event func (s *searchSupport) handleEvent(ctx context.Context, evt *WrittenEvent) { + if !slices.Contains([]WatchEvent_Type{WatchEvent_ADDED, WatchEvent_MODIFIED, WatchEvent_DELETED}, evt.Type) { + s.log.Info("ignoring watch event", "type", evt.Type) + return + } + nsr := NamespacedResource{ Namespace: evt.Key.Namespace, Group: evt.Key.Group, @@ -233,10 +252,35 @@ func (s *searchSupport) handleEvent(ctx context.Context, evt *WrittenEvent) { return } - err = index.Write(doc) - if err != nil { - s.log.Warn("error writing document watch event", "error", err) - return + switch evt.Type { + case WatchEvent_ADDED, WatchEvent_MODIFIED: + err = index.Write(doc) + if err != nil { + s.log.Warn("error writing document watch event", "error", err) + return + } + if evt.Type == WatchEvent_ADDED { + IndexMetrics.IndexedKinds.WithLabelValues(evt.Key.Resource).Inc() + } + case WatchEvent_DELETED: + err = index.Delete(evt.Key) + if err != nil { + s.log.Warn("error deleting document watch event", "error", err) + return + } + IndexMetrics.IndexedKinds.WithLabelValues(evt.Key.Resource).Dec() + default: + // do nothing + s.log.Warn("unknown watch event", "type", evt.Type) + } + + // record latency from when event was created to when it was indexed + latencySeconds := float64(time.Now().UnixMicro()-evt.ResourceVersion) / 1e6 + if latencySeconds > 5 { + logger.Warn("high index latency", "latency", latencySeconds) + } + if IndexMetrics != nil { + IndexMetrics.IndexLatency.WithLabelValues(evt.Key.Resource).Observe(latencySeconds) } } @@ -311,6 +355,15 @@ func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size return rv, err }) + // Record the number of objects indexed for the kind/resource + docCount, err := index.DocCount() + if err != nil { + s.log.Warn("error getting doc count", "error", err) + } + if IndexMetrics != nil { + IndexMetrics.IndexedKinds.WithLabelValues(key.Resource).Add(float64(docCount)) + } + if err != nil { return nil, 0, err } diff --git a/pkg/storage/unified/search/bleve.go b/pkg/storage/unified/search/bleve.go index f563cad9499..b8047142fe6 100644 --- a/pkg/storage/unified/search/bleve.go +++ b/pkg/storage/unified/search/bleve.go @@ -11,7 +11,6 @@ import ( "github.com/blevesearch/bleve/v2" "github.com/blevesearch/bleve/v2/search" "github.com/blevesearch/bleve/v2/search/query" - "github.com/prometheus/client_golang/prometheus" "go.opentelemetry.io/otel/trace" "k8s.io/apimachinery/pkg/selection" @@ -46,7 +45,7 @@ type bleveBackend struct { cacheMu sync.RWMutex } -func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, reg prometheus.Registerer) *bleveBackend { +func NewBleveBackend(opts BleveOptions, tracer trace.Tracer) *bleveBackend { b := &bleveBackend{ log: slog.Default().With("logger", "bleve-backend"), tracer: tracer, @@ -54,10 +53,6 @@ func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, reg prometheus.Regi opts: opts, } - if reg != nil { - b.log.Info("TODO, register metrics collectors!") - } - return b } @@ -107,8 +102,10 @@ func (b *bleveBackend) BuildIndex(ctx context.Context, if err == nil { b.log.Info("TODO, check last RV so we can see if the numbers have changed", "dir", dir) } + resource.IndexMetrics.IndexTenants.WithLabelValues(key.Namespace, "file").Inc() } else { index, err = bleve.NewMemOnly(mapper) + resource.IndexMetrics.IndexTenants.WithLabelValues(key.Namespace, "memory").Inc() } if err != nil { return nil, err @@ -144,6 +141,19 @@ func (b *bleveBackend) BuildIndex(ctx context.Context, return idx, nil } +// TotalDocs returns the total number of documents across all indices +func (b *bleveBackend) TotalDocs() int64 { + var totalDocs int64 + for _, v := range b.cache { + c, err := v.index.DocCount() + if err != nil { + continue + } + totalDocs += int64(c) + } + return totalDocs +} + type bleveIndex struct { key resource.NamespacedResource index bleve.Index @@ -285,6 +295,11 @@ func (b *bleveIndex) Search( return response, nil } +func (b *bleveIndex) DocCount() (int, error) { + count, err := b.index.DocCount() + return int(count), err +} + // make sure the request key matches the index func (b *bleveIndex) verifyKey(key *resource.ResourceKey) *resource.ErrorResult { if key.Namespace != b.key.Namespace { diff --git a/pkg/storage/unified/search/bleve_test.go b/pkg/storage/unified/search/bleve_test.go index 2a383393818..d5d3db1d7d3 100644 --- a/pkg/storage/unified/search/bleve_test.go +++ b/pkg/storage/unified/search/bleve_test.go @@ -30,14 +30,10 @@ func TestBleveBackend(t *testing.T) { tmpdir, err := os.CreateTemp("", "bleve-test") require.NoError(t, err) - backend := NewBleveBackend( - BleveOptions{ - Root: tmpdir.Name(), - FileThreshold: 5, // with more than 5 items we create a file on disk - }, - tracing.NewNoopTracerService(), - nil, - ) + backend := NewBleveBackend(BleveOptions{ + Root: tmpdir.Name(), + FileThreshold: 5, // with more than 5 items we create a file on disk + }, tracing.NewNoopTracerService()) rv := int64(10) ctx := context.Background() diff --git a/pkg/storage/unified/sql/server.go b/pkg/storage/unified/sql/server.go index a0fe7e230ad..19dbfca1f12 100644 --- a/pkg/storage/unified/sql/server.go +++ b/pkg/storage/unified/sql/server.go @@ -2,10 +2,11 @@ package sql import ( "context" + "log/slog" "os" - "path/filepath" "strings" + "github.com/grafana/grafana/pkg/storage/unified/search" "github.com/prometheus/client_golang/prometheus" infraDB "github.com/grafana/grafana/pkg/infra/db" @@ -14,7 +15,6 @@ import ( "github.com/grafana/grafana/pkg/services/featuremgmt" "github.com/grafana/grafana/pkg/setting" "github.com/grafana/grafana/pkg/storage/unified/resource" - "github.com/grafana/grafana/pkg/storage/unified/search" "github.com/grafana/grafana/pkg/storage/unified/sql/db/dbimpl" ) @@ -59,13 +59,18 @@ func NewResourceServer(ctx context.Context, db infraDB.DB, cfg *setting.Cfg, if features.IsEnabledGlobally(featuremgmt.FlagUnifiedStorageSearch) { opts.Search = resource.SearchOptions{ Backend: search.NewBleveBackend(search.BleveOptions{ - Root: filepath.Join(cfg.DataPath, "unified-search", "bleve"), - FileThreshold: 10, // fewer than X items will use a memory index - BatchSize: 500, // This is the batch size for how many objects to add to the index at once - }, tracer, reg), + Root: cfg.IndexPath, + 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), Resources: docs, - WorkerThreads: 5, // from cfg? - InitMinCount: 1, + WorkerThreads: cfg.IndexWorkers, + InitMinCount: cfg.IndexMinCount, + } + + err = reg.Register(resource.NewIndexMetrics(cfg.IndexPath, opts.Search.Backend)) + if err != nil { + slog.Warn("Failed to register indexer metrics", "error", err) } }