chore(blooms): adds more instrumentation to block building (#12779)

Author: owen-d

pkg/bloomcompactor/spec.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"time"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -74,12 +75,16 @@ func NewSimpleBloomGenerator(
 	logger log.Logger,
 ) *SimpleBloomGenerator {
 	return &SimpleBloomGenerator{
-		userID:       userID,
-		opts:         opts,
-		store:        store,
-		chunkLoader:  chunkLoader,
-		blocksIter:   blocksIter,
-		logger:       log.With(logger, "component", "bloom_generator"),
+		userID:      userID,
+		opts:        opts,
+		store:       store,
+		chunkLoader: chunkLoader,
+		blocksIter:  blocksIter,
+		logger: log.With(
+			logger,
+			"component", "bloom_generator",
+			"org_id", userID,
+		),
 		readWriterFn: readWriterFn,
 		metrics:      metrics,
 		reporter:     reporter,
@@ -90,6 +95,13 @@ func NewSimpleBloomGenerator(
 
 func (s *SimpleBloomGenerator) populator(ctx context.Context) func(series *v1.Series, bloom *v1.Bloom) (int, error) {
 	return func(series *v1.Series, bloom *v1.Bloom) (int, error) {
+		start := time.Now()
+		level.Debug(s.logger).Log(
+			"msg", "populating bloom filter",
+			"stage", "before",
+			"fp", series.Fingerprint,
+			"chunks", len(series.Chunks),
+		)
 		chunkItersWithFP, err := s.chunkLoader.Load(ctx, s.userID, series)
 		if err != nil {
 			return 0, errors.Wrapf(err, "failed to load chunks for series: %+v", series)
@@ -103,6 +115,16 @@ func (s *SimpleBloomGenerator) populator(ctx context.Context) func(series *v1.Se
 			chunkItersWithFP.itr,
 		)
 
+		level.Debug(s.logger).Log(
+			"msg", "populating bloom filter",
+			"stage", "after",
+			"fp", series.Fingerprint,
+			"chunks", len(series.Chunks),
+			"series_bytes", bytesAdded,
+			"duration", time.Since(start),
+			"err", err,
+		)
+
 		if s.reporter != nil {
 			s.reporter(series.Fingerprint)
 		}

pkg/storage/bloom/v1/builder.go

@@ -157,26 +157,26 @@ func (b *BlockBuilder) AddSeries(series SeriesWithBloom) (bool, error) {
 		return false, errors.Wrapf(err, "writing index for series %v", series.Series.Fingerprint)
 	}
 
-	full, err := b.isBlockFull()
+	full, _, err := b.IsBlockFull()
 	if err != nil {
 		return false, errors.Wrap(err, "checking if block is full")
 	}
 
 	return full, nil
 }
 
-func (b *BlockBuilder) isBlockFull() (bool, error) {
-	// if the block size is 0, the max size is unlimited
-	if b.opts.BlockSize == 0 {
-		return false, nil
+func (b *BlockBuilder) IsBlockFull() (full bool, size int, err error) {
+	size, err = b.writer.Size()
+	if err != nil {
+		return false, 0, errors.Wrap(err, "getting block size")
 	}
 
-	size, err := b.writer.Size()
-	if err != nil {
-		return false, errors.Wrap(err, "getting block size")
+	// if the block size is 0, the max size is unlimited
+	if b.opts.BlockSize == 0 {
+		return false, size, nil
 	}
 
-	return uint64(size) >= b.opts.BlockSize, nil
+	return uint64(size) >= b.opts.BlockSize, size, nil
 }
 
 type BloomBlockBuilder struct {
@@ -657,6 +657,14 @@ func (mb *MergeBuilder) Build(builder *BlockBuilder) (checksum uint32, totalByte
 		return 0, totalBytes, errors.Wrap(err, "iterating store")
 	}
 
+	flushedFor := blockFlushReasonFinished
+	full, sz, _ := builder.IsBlockFull()
+	if full {
+		flushedFor = blockFlushReasonFull
+	}
+	mb.metrics.blockSize.Observe(float64(sz))
+	mb.metrics.blockFlushReason.WithLabelValues(flushedFor).Inc()
+
 	checksum, err = builder.Close()
 	if err != nil {
 		return 0, totalBytes, errors.Wrap(err, "closing block")

pkg/storage/bloom/v1/metrics.go

@@ -18,6 +18,9 @@ type Metrics struct {
 	tokensTotal         prometheus.Counter
 	insertsTotal        *prometheus.CounterVec
 
+	blockSize        prometheus.Histogram
+	blockFlushReason *prometheus.CounterVec
+
 	pagesRead    *prometheus.CounterVec
 	pagesSkipped *prometheus.CounterVec
 	bytesRead    *prometheus.CounterVec
@@ -34,6 +37,9 @@ const (
 	collisionTypeTrue      = "true"
 	collisionTypeCache     = "cache"
 
+	blockFlushReasonFull     = "full"
+	blockFlushReasonFinished = "finished"
+
 	pageTypeBloom  = "bloom"
 	pageTypeSeries = "series"
 
@@ -94,6 +100,18 @@ func NewMetrics(r prometheus.Registerer) *Metrics {
 			Help:      "Number of inserts into the bloom filter. collision type may be `false` (no collision), `cache` (found in token cache) or true (found in bloom filter). token_type may be either `raw` (the original ngram) or `chunk_prefixed` (the ngram with the chunk prefix)",
 		}, []string{"token_type", "collision"}),
 
+		blockSize: promauto.With(r).NewHistogram(prometheus.HistogramOpts{
+			Namespace: constants.Loki,
+			Name:      "bloom_block_size",
+			Help:      "Size of the bloom block in bytes",
+			Buckets:   prometheus.ExponentialBucketsRange(1<<20, 1<<30, 8),
+		}),
+		blockFlushReason: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
+			Namespace: constants.Loki,
+			Name:      "bloom_block_flush_reason_total",
+			Help:      "Reason the block was finished. Can be either `full` (the block hit its maximum size) or `finished` (the block was finished due to the end of the series).",
+		}, []string{"reason"}),
+
 		pagesRead: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
 			Namespace: constants.Loki,
 			Name:      "bloom_pages_read_total",