pass series to page decoder

Author: salvacorts

pkg/storage/bloom/v1/bloom.go

@@ -3,6 +3,7 @@ package v1
 import (
 	"bytes"
 	"fmt"
+	"github.com/prometheus/common/model"
 	"io"
 
 	"github.com/go-kit/log/level"
@@ -377,13 +378,18 @@ func (b *BloomBlock) DecodeHeaders(r io.ReadSeeker) (uint32, error) {
 	return checksum, nil
 }
 
-func (b *BloomBlock) BloomPageDecoder(r io.ReadSeeker, pageIdx int, maxPageSize int, metrics *Metrics) (res *BloomPageDecoder, err error) {
+func (b *BloomBlock) BloomPageDecoder(r io.ReadSeeker, pageIdx int, maxPageSize int, metrics *Metrics, s ...model.Fingerprint) (res *BloomPageDecoder, err error) {
 	if pageIdx < 0 || pageIdx >= len(b.pageHeaders) {
 		metrics.pagesSkipped.WithLabelValues(pageTypeBloom, skipReasonOOB).Inc()
 		metrics.bytesSkipped.WithLabelValues(pageTypeBloom, skipReasonOOB).Add(float64(b.pageHeaders[pageIdx].DecompressedLen))
 		return nil, fmt.Errorf("invalid page (%d) for bloom page decoding", pageIdx)
 	}
 
+	var series model.Fingerprint
+	if len(s) > 0 {
+		series = s[0]
+	}
+
 	page := b.pageHeaders[pageIdx]
 	// fmt.Printf("pageIdx=%d page=%+v size=%.2fMiB\n", pageIdx, page, float64(page.Len)/float64(1<<20))
 
@@ -465,6 +471,7 @@ func (b *BloomBlock) BloomPageDecoder(r io.ReadSeeker, pageIdx int, maxPageSize
 
 		level.Error(util_log.Logger).Log(
 			"msg", "page too large",
+			"series_fp", series.String(),
 			"version", b.schema.version,
 			"page", pageIdx,
 			"len", page.Len,
@@ -488,7 +495,8 @@ func (b *BloomBlock) BloomPageDecoder(r io.ReadSeeker, pageIdx int, maxPageSize
 			"fpRatesPerLayer", fpRatesPerLayer,
 		)
 
-		return nil, ErrPageTooLarge
+		return nil, fmt.Errorf("bloom too big for series %s - bytes(%d) origBytes(%d) layers(%d) fillRatio(%.2f) lastLayerFillRatio(%.2f) capacity(%d) capacityPerLayer(%s) countPerLayer(%s) bytesPerLayer(%s) fpRatesPerLayer(%s)",
+			series, bloom.BytesSize(), page.Stats.Bytes, len(bloom.CapacityPerLayer()), bloom.FillRatio(), fillRatioLastLayer, bloom.Capacity(), capacityPerLayer, countPerLayer, bytesPerLayer, fpRatesPerLayer)
 	}
 
 	if _, err = r.Seek(int64(page.Offset), io.SeekStart); err != nil {

pkg/storage/bloom/v1/bloom_querier.go

@@ -2,6 +2,7 @@ package v1
 
 import (
 	"github.com/pkg/errors"
+	"github.com/prometheus/common/model"
 )
 
 type BloomQuerier interface {
@@ -44,7 +45,7 @@ func (it *LazyBloomIter) ensureInit() {
 	}
 }
 
-func (it *LazyBloomIter) Seek(offset BloomOffset) {
+func (it *LazyBloomIter) Seek(offset BloomOffset, series ...model.Fingerprint) {
 	it.ensureInit()
 
 	// if we need a different page or the current page hasn't been loaded,
@@ -62,29 +63,28 @@ func (it *LazyBloomIter) Seek(offset BloomOffset) {
 			it.err = errors.Wrap(err, "getting blooms reader")
 			return
 		}
-		decoder, err := it.b.blooms.BloomPageDecoder(r, offset.Page, it.m, it.b.metrics)
+		decoder, err := it.b.blooms.BloomPageDecoder(r, offset.Page, it.m, it.b.metrics, series...)
 		if err != nil {
 			it.err = errors.Wrap(err, "loading bloom page")
 			return
 		}
 
 		it.curPageIndex = offset.Page
 		it.curPage = decoder
-
 	}
 
 	it.curPage.Seek(offset.ByteOffset)
 }
 
-func (it *LazyBloomIter) Next() bool {
+func (it *LazyBloomIter) Next(series ...model.Fingerprint) bool {
 	it.ensureInit()
 	if it.err != nil {
 		return false
 	}
-	return it.next()
+	return it.next(series...)
 }
 
-func (it *LazyBloomIter) next() bool {
+func (it *LazyBloomIter) next(series ...model.Fingerprint) bool {
 	if it.err != nil {
 		return false
 	}
@@ -103,6 +103,7 @@ func (it *LazyBloomIter) next() bool {
 				it.curPageIndex,
 				it.m,
 				it.b.metrics,
+				series...,
 			)
 			if err != nil {
 				it.err = err

pkg/storage/bloom/v1/fuse.go

@@ -101,14 +101,19 @@ func (fq *FusedQuerier) Run() error {
 		)
 
 		// Now that we've found the series, we need to find the unpack the bloom
-		fq.bq.blooms.Seek(series.Offset)
-		if !fq.bq.blooms.Next() {
-			if errors.Is(fq.bq.blooms.Err(), ErrPageTooLarge) {
-				level.Warn(util_log.Logger).Log(
-					"msg", "bloom too large for series",
-					"series", series.Fingerprint,
-				)
-			}
+		fq.bq.blooms.Seek(series.Offset, series.Fingerprint)
+		if !fq.bq.blooms.Next(series.Fingerprint) {
+			//if errors.Is(fq.bq.blooms.Err(), ErrPageTooLarge) {
+			//	level.Warn(util_log.Logger).Log(
+			//		"msg", "bloom too large for series",
+			//		"series", series.Fingerprint,
+			//	)
+			//}
+
+			level.Warn(util_log.Logger).Log(
+				"msg", "bloom too large for series",
+				err, fq.bq.blooms.Err(),
+			)
 
 			// fingerprint not found, can't remove chunks
 			level.Debug(fq.logger).Log("msg", "fingerprint not found", "fp", series.Fingerprint, "err", fq.bq.blooms.Err())

pkg/storage/bloom/v1/metrics.go

@@ -153,13 +153,13 @@ func NewMetrics(r prometheus.Registerer) *Metrics {
 			Namespace: constants.Loki,
 			Name:      "bloom_exp_source_bytes",
 			Help:      "Number of bytes in a bloom source",
-			Buckets:   prometheus.ExponentialBucketsRange(1<<10, 512<<20, 10),
+			Buckets:   prometheus.ExponentialBucketsRange(1<<10, 30<<30, 50), // 1KB to 30GB
 		}, []string{"size"}),
 		bloomLines: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{
 			Namespace: constants.Loki,
 			Name:      "bloom_exp_lines",
 			Help:      "Number of lines in a bloom",
-			Buckets:   prometheus.ExponentialBucketsRange(1, 33554432, 10),
+			Buckets:   prometheus.ExponentialBucketsRange(1, 50000000, 50),
 		}, []string{"size"}),
 		bloomLayers: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{
 			Namespace: constants.Loki,
@@ -171,7 +171,7 @@ func NewMetrics(r prometheus.Registerer) *Metrics {
 			Namespace: constants.Loki,
 			Name:      "bloom_exp_bytes_size",
 			Help:      "Number of bytes in a bloom",
-			Buckets:   prometheus.ExponentialBucketsRange(1<<10, 512<<20, 10),
+			Buckets:   prometheus.ExponentialBucketsRange(1<<10, 512<<20, 20),
 		}, []string{"size"}),
 		bloomLastLayerFillRatio: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{
 			Namespace: constants.Loki,