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

@@ -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 {

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)
 }

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
+}

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
 	}

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 {

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()

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)
 		}
 	}