Add more metrics and logging

Author: fionaliao

pkg/streamingpromql/operators/functions/split_model.go

@@ -3,6 +3,7 @@
 package functions
 
 import (
+	"github.com/go-kit/log"
 	"github.com/grafana/mimir/pkg/streamingpromql/optimize/plan/querysplitting/cache"
 	"github.com/grafana/mimir/pkg/streamingpromql/planning"
 	"github.com/grafana/mimir/pkg/streamingpromql/types"
@@ -49,6 +50,7 @@ type SplittableOperatorFactory func(
 	annotations *annotations.Annotations,
 	memoryConsumptionTracker *limiter.MemoryConsumptionTracker,
 	enableDelayedNameRemoval bool,
+	logger log.Logger,
 ) (types.Operator, error)
 
 func NewSplitOperatorFactory[T any](
@@ -69,6 +71,7 @@ func NewSplitOperatorFactory[T any](
 		annotations *annotations.Annotations,
 		memoryConsumptionTracker *limiter.MemoryConsumptionTracker,
 		enableDelayedNameRemoval bool,
+		logger log.Logger,
 	) (types.Operator, error) {
 		return NewSplittingFunctionOverRangeVector[T](
 			innerNode,
@@ -86,6 +89,7 @@ func NewSplitOperatorFactory[T any](
 			annotations,
 			memoryConsumptionTracker,
 			enableDelayedNameRemoval,
+			logger,
 		)
 	}
 }

pkg/streamingpromql/operators/functions/split_operator.go

@@ -11,7 +11,10 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
 	"github.com/grafana/mimir/pkg/streamingpromql/optimize/plan/querysplitting/cache"
+	"github.com/grafana/mimir/pkg/util/spanlogger"
 	promts "github.com/prometheus/prometheus/model/timestamp"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/promql/parser/posrange"
@@ -53,6 +56,9 @@ type FunctionOverRangeVectorSplit[T any] struct {
 	// seriesToSplits is ordered the same way as SeriesMetadata
 	seriesToSplits   [][]SplitSeries
 	currentSeriesIdx int
+
+	logger     log.Logger
+	cacheStats *cache.CacheStats
 }
 
 var _ types.InstantVectorOperator = (*FunctionOverRangeVectorSplit[any])(nil)
@@ -73,6 +79,7 @@ func NewSplittingFunctionOverRangeVector[T any](
 	annotations *annotations.Annotations,
 	memoryConsumptionTracker *limiter.MemoryConsumptionTracker,
 	enableDelayedNameRemoval bool,
+	logger log.Logger,
 ) (*FunctionOverRangeVectorSplit[T], error) {
 	if !timeRange.IsInstant {
 		return nil, fmt.Errorf("FunctionOverRangeVectorSplit only supports instant queries")
@@ -95,6 +102,8 @@ func NewSplittingFunctionOverRangeVector[T any](
 		expressionPosition:       expressionPosition,
 		timeRange:                timeRange,
 		enableDelayedNameRemoval: enableDelayedNameRemoval,
+		logger:                   logger,
+		cacheStats:               &cache.CacheStats{},
 	}
 
 	if funcDef.SeriesValidationFuncFactory != nil {
@@ -141,7 +150,7 @@ func (m *FunctionOverRangeVectorSplit[T]) createSplits(ctx context.Context) ([]S
 	for _, splitRange := range m.splitRanges {
 		// For cacheable (aligned) ranges, check the cache
 		if splitRange.Cacheable {
-			cacheEntry, found, err := cache.NewReadEntry[T](m.cache, m.codec, ctx, int32(m.FuncId), m.innerCacheKey, splitRange.Start, splitRange.End)
+			cacheEntry, found, err := cache.NewReadEntry[T](m.cache, m.codec, ctx, int32(m.FuncId), m.innerCacheKey, splitRange.Start, splitRange.End, m.cacheStats)
 			if err != nil {
 				return nil, err
 			}
@@ -375,12 +384,45 @@ func (m *FunctionOverRangeVectorSplit[T]) emitAnnotation(generator types.Annotat
 }
 
 func (m *FunctionOverRangeVectorSplit[T]) Finalize(ctx context.Context) error {
+	logger := spanlogger.FromContext(ctx, m.logger)
+
+	// Count cached vs uncached splits
+	var cachedCount, uncachedCount int
+	for _, split := range m.splits {
+		if split.IsCached() {
+			cachedCount++
+		} else {
+			uncachedCount++
+		}
+	}
+
+	// Finalize all splits to ensure cache writes are complete
 	for _, split := range m.splits {
 		if err := split.Finalize(ctx); err != nil {
 			return err
 		}
 	}
 
+	// TODO: currently at info level while testing, may also modify and remove some stats post tests
+	level.Info(logger).Log(
+		"msg", "query splitting stats",
+		"function", m.FuncId.PromQLName(),
+		"inner_cache_key", m.innerCacheKey,
+		"query_start_ms", m.queryTimeRange.StartT,
+		"query_end_ms", m.queryTimeRange.EndT,
+		"splits_total", len(m.splits),
+		"splits_cached", cachedCount,
+		"splits_uncached", uncachedCount,
+		"cache_entries_cached", m.cacheStats.CachedEntries,
+		"cache_entries_uncached", m.cacheStats.UncachedEntries,
+		"max_series_per_entry", m.cacheStats.MaxSeries,
+		"min_series_per_entry", m.cacheStats.MinSeries,
+		"total_series", m.cacheStats.TotalSeries,
+		"max_bytes_per_entry", m.cacheStats.MaxBytes,
+		"min_bytes_per_entry", m.cacheStats.MinBytes,
+		"total_cache_bytes", m.cacheStats.TotalBytes,
+	)
+
 	return nil
 }
 
@@ -396,6 +438,7 @@ type Split[T any] interface {
 	ReadResultsAt(ctx context.Context, idx int) ([]T, error)
 	Finalize(ctx context.Context) error
 	Close()
+	IsCached() bool
 }
 
 type SplitSeries struct {
@@ -440,6 +483,10 @@ func (c *CachedSplit[T]) Finalize(ctx context.Context) error {
 func (c *CachedSplit[T]) Close() {
 }
 
+func (c *CachedSplit[T]) IsCached() bool {
+	return true
+}
+
 type UncachedSplit[T any] struct {
 	ranges   []Range
 	operator types.RangeVectorOperator
@@ -466,7 +513,7 @@ func NewUncachedSplit[T any](
 			continue
 		}
 
-		cacheEntries[i], err = cache.NewWriteEntry[T](parent.cache, parent.codec, ctx, int32(parent.FuncId), parent.innerCacheKey, splitRange.Start, splitRange.End)
+		cacheEntries[i], err = cache.NewWriteEntry[T](parent.cache, parent.codec, ctx, int32(parent.FuncId), parent.innerCacheKey, splitRange.Start, splitRange.End, parent.cacheStats)
 		if err != nil {
 			return nil, err
 		}
@@ -568,6 +615,10 @@ func (p *UncachedSplit[T]) Close() {
 	}
 }
 
+func (p *UncachedSplit[T]) IsCached() bool {
+	return false
+}
+
 type ResultGetter[T any] struct {
 	resultBuffer   map[int][]T
 	nextSeriesIdx  int

pkg/streamingpromql/optimize/plan/querysplitting/cache/cache.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/grafana/dskit/tenant"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -29,9 +30,28 @@ type Backend interface {
 
 type Cache struct {
 	backend Backend
+	metrics *resultsCacheMetrics
 	logger  log.Logger
 }
 
+type resultsCacheMetrics struct {
+	cacheRequests prometheus.Counter
+	cacheHits     prometheus.Counter
+}
+
+func newResultsCacheMetrics(reg prometheus.Registerer) *resultsCacheMetrics {
+	return &resultsCacheMetrics{
+		cacheRequests: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "mimir_query_engine_intermediate_result_cache_requests_total",
+			Help: "Total number of requests (or partial requests) looked up in the results cache.",
+		}),
+		cacheHits: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "mimir_query_engine_intermediate_result_cache_hits_total",
+			Help: "Total number of requests (or partial requests) fetched from the results cache.",
+		}),
+	}
+}
+
 func NewResultsCache(cfg Config, logger log.Logger, reg prometheus.Registerer) (*Cache, error) {
 	client, err := cache.CreateClient("intermediate-result-cache", cfg.BackendConfig, logger, prometheus.WrapRegistererWithPrefix("mimir_", reg))
 	if err != nil {
@@ -46,12 +66,13 @@ func NewResultsCache(cfg Config, logger log.Logger, reg prometheus.Registerer) (
 	)
 
 	logger.Log("msg", "intermediate results cache initialized", "backend", cfg.Backend)
-	return NewResultsCacheWithBackend(backend, logger), nil
+	return NewResultsCacheWithBackend(backend, reg, logger), nil
 }
 
-func NewResultsCacheWithBackend(backend Backend, logger log.Logger) *Cache {
+func NewResultsCacheWithBackend(backend Backend, reg prometheus.Registerer, logger log.Logger) *Cache {
 	return &Cache{
 		backend: backend,
+		metrics: newResultsCacheMetrics(reg),
 		logger:  logger,
 	}
 }
@@ -90,12 +111,14 @@ func NewReadEntry[T any](
 	function int32,
 	innerKey string,
 	start, end int64,
+	stats *CacheStats,
 ) (ReadEntry[T], bool, error) {
 	tenant, err := user.ExtractOrgID(ctx)
 	if err != nil {
 		return nil, false, err
 	}
 
+	c.metrics.cacheRequests.Inc()
 	cacheKey := generateCacheKey(tenant, function, innerKey, start, end)
 	hashedKey := cacheHashKey(cacheKey)
 
@@ -107,19 +130,25 @@ func NewReadEntry[T any](
 
 	var cached CachedSeries
 	if err := cached.Unmarshal(data); err != nil {
+		level.Warn(c.logger).Log("msg", "failed to decode cached result", "hashed_cache_key", hashedKey, "cache_key", cacheKey, "err", err)
 		return nil, false, nil
 	}
 
 	if cached.CacheKey != cacheKey {
+		level.Warn(c.logger).Log("msg", "skipped cached result because a cache key collision has been found", "hashed_cache_key", hashedKey, "cache_key", cacheKey)
+
 		return nil, false, nil
 	}
 
+	c.metrics.cacheHits.Inc()
+	level.Debug(c.logger).Log("msg", "cache hit", "tenant", tenant, "function", function, "innerKey", innerKey, "start", start, "end", end)
+
 	reader, err := codec.NewReader(cached.Results)
 	if err != nil {
 		return nil, false, err
 	}
 
-	level.Debug(c.logger).Log("msg", "cache hit", "tenant", tenant, "function", function, "innerKey", innerKey, "start", start, "end", end)
+	stats.AddCachedEntryStat(len(cached.Series), len(data))
 
 	return &bufferedReadEntry[T]{
 		cached: cached,
@@ -134,6 +163,7 @@ func NewWriteEntry[T any](
 	function int32,
 	innerKey string,
 	start, end int64,
+	stats *CacheStats,
 ) (WriteEntry[T], error) {
 	tenant, err := user.ExtractOrgID(ctx)
 	if err != nil {
@@ -151,6 +181,7 @@ func NewWriteEntry[T any](
 		},
 		finalized: false,
 		logger:    c.logger,
+		stats:     stats,
 	}
 
 	writer, err := codec.NewWriter(func(data []byte) {
@@ -195,6 +226,7 @@ type bufferedWriteEntry[T any] struct {
 	cached    CachedSeries
 	writer    SplitWriter[T]
 	finalized bool
+	stats     *CacheStats
 	logger    log.Logger
 }
 
@@ -231,6 +263,8 @@ func (e *bufferedWriteEntry[T]) Finalize() error {
 
 	level.Debug(e.logger).Log("msg", "cache entry written", "cache_key", e.cached.CacheKey, "series_count", len(e.cached.Series), "entry_size", len(data))
 
+	e.stats.AddUncachedEntryStat(len(e.cached.Series), len(data))
+
 	e.finalized = true
 	return nil
 }

pkg/streamingpromql/optimize/plan/querysplitting/cache/stats.go

@@ -0,0 +1,49 @@
+package cache
+
+// TODO: this may just be temporary during testing
+type CacheStats struct {
+	CachedEntries   int // Entries read from cache
+	UncachedEntries int // Entries written to cache
+	TotalSeries     int // not deduped
+	MinSeries       int
+	MaxSeries       int
+	TotalBytes      int
+	MinBytes        int
+	MaxBytes        int
+}
+
+func (c *CacheStats) AddCachedEntryStat(seriesCount, size int) {
+	c.CachedEntries++
+	c.TotalSeries += seriesCount
+	if c.MinSeries == 0 || seriesCount < c.MinSeries {
+		c.MinSeries = seriesCount
+	}
+	if seriesCount > c.MaxSeries {
+		c.MaxSeries = seriesCount
+	}
+	c.TotalBytes += size
+	if c.MinBytes == 0 || size < c.MinBytes {
+		c.MinBytes = size
+	}
+	if size > c.MaxBytes {
+		c.MaxBytes = size
+	}
+}
+
+func (c *CacheStats) AddUncachedEntryStat(seriesCount, size int) {
+	c.UncachedEntries++
+	c.TotalSeries += seriesCount
+	if c.MinSeries == 0 || seriesCount < c.MinSeries {
+		c.MinSeries = seriesCount
+	}
+	if seriesCount > c.MaxSeries {
+		c.MaxSeries = seriesCount
+	}
+	c.TotalBytes += size
+	if c.MinBytes == 0 || size < c.MinBytes {
+		c.MinBytes = size
+	}
+	if size > c.MaxBytes {
+		c.MaxBytes = size
+	}
+}

pkg/streamingpromql/optimize/plan/querysplitting/node.go

@@ -183,6 +183,7 @@ func (m Materializer) Materialize(n planning.Node, materializer *planning.Materi
 		params.Annotations,
 		params.MemoryConsumptionTracker,
 		params.EnableDelayedNameRemoval,
+		params.Logger,
 	)
 	if err != nil {
 		return nil, err