add basic caching bucket config & benchmark stats to BenchmarkLabelValuesOffsetsIndexV2_WithPrefix - TODO separate benchmark & make it more interesting

Author: francoposa

pkg/storage/indexheader/bucket_binary_reader.go

@@ -29,9 +29,8 @@ import (
 
 // BucketBinaryReader reads index header data directly from object storage.
 type BucketBinaryReader struct {
-	bkt            objstore.BucketReader
-	factory        *streamencoding.BucketDecbufFactory
-	bufReaderStats *streamencoding.BufReaderStats
+	bkt     objstore.BucketReader
+	factory *streamencoding.BucketDecbufFactory
 
 	toc *index.TOC
 
@@ -81,10 +80,8 @@ func NewBucketBinaryReader(
 	indexPath := filepath.Join(blockID.String(), block.IndexFilename)
 
 	r := &BucketBinaryReader{
-		bkt: bkt,
-
-		factory:        streamencoding.NewBucketDecbufFactory(ctx, bkt, indexPath),
-		bufReaderStats: &streamencoding.BufReaderStats{},
+		bkt:     bkt,
+		factory: streamencoding.NewBucketDecbufFactory(ctx, bkt, indexPath),
 	}
 
 	indexAttrs, err := bkt.Attributes(ctx, indexPath)

pkg/storage/indexheader/reader_benchmarks_test.go

@@ -13,7 +13,10 @@ import (
 	"time"
 
 	"github.com/go-kit/log"
+	"github.com/grafana/dskit/cache"
+	"github.com/grafana/mimir/pkg/storage/tsdb/bucketcache"
 	"github.com/oklog/ulid/v2"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/stretchr/testify/require"
 	"github.com/thanos-io/objstore"
@@ -267,11 +270,23 @@ func BenchmarkLabelValuesOffsetsIndexV2_WithPrefix(b *testing.B) {
 		require.NoError(b, bkt.Close())
 	})
 
-	diskReader, err := NewStreamBinaryReader(ctx, log.NewNopLogger(), instBkt, bucketDir, blockID, 32, NewStreamBinaryReaderMetrics(nil), Config{})
+	// We reuse cache between tests (!)
+	cache := cache.NewMockCache()
+
+	const cfgName = "index"
+	subrangeSize := int64(16000)
+	cfg := bucketcache.NewCachingBucketConfig()
+	cfg.CacheGetRange(cfgName, cache, func(string) bool { return true }, subrangeSize, cache, time.Hour, time.Hour, 0)
+
+	cacheBucketReg := prometheus.NewPedanticRegistry()
+	cachingBucket, err := bucketcache.NewCachingBucket("test", instBkt, cfg, log.NewNopLogger(), cacheBucketReg)
+	require.NoError(b, err)
+
+	diskReader, err := NewStreamBinaryReader(ctx, log.NewNopLogger(), cachingBucket, bucketDir, blockID, 32, NewStreamBinaryReaderMetrics(nil), Config{})
 	require.NoError(b, err)
 	b.Cleanup(func() { require.NoError(b, diskReader.Close()) })
 
-	bucketReader, err := NewBucketBinaryReader(ctx, log.NewNopLogger(), instBkt, bucketDir, blockID, 32, Config{})
+	bucketReader, err := NewBucketBinaryReader(ctx, log.NewNopLogger(), cachingBucket, bucketDir, blockID, 32, Config{})
 	require.NoError(b, err)
 	b.Cleanup(func() { require.NoError(b, bucketReader.Close()) })
 
@@ -296,9 +311,30 @@ func BenchmarkLabelValuesOffsetsIndexV2_WithPrefix(b *testing.B) {
 	b.ReportAllocs()
 	for _, lbl := range sortedTestLbls {
 		tcs := tests[lbl]
-		for readerName, reader := range readers {
+		for readerName, _ := range readers {
 			for _, tc := range tcs {
 				b.Run(fmt.Sprintf("Reader=%s/Label=%s/Prefix='%s'/Desc=%s", readerName, lbl, tc.prefix, tc.desc), func(b *testing.B) {
+					cacheBucketReg := prometheus.NewPedanticRegistry()
+					cachingBucket, err := bucketcache.NewCachingBucket("test", instBkt, cfg, log.NewNopLogger(), cacheBucketReg)
+					require.NoError(b, err)
+
+					var reader Reader
+					switch readerName {
+					case "disk":
+						r, err := NewStreamBinaryReader(ctx, log.NewNopLogger(), cachingBucket, bucketDir, blockID, 32, NewStreamBinaryReaderMetrics(nil), Config{})
+						require.NoError(b, err)
+						b.Cleanup(func() { require.NoError(b, r.Close()) })
+						reader = r
+					case "bucket":
+						r, err := NewBucketBinaryReader(ctx, log.NewNopLogger(), cachingBucket, bucketDir, blockID, 32, Config{})
+						require.NoError(b, err)
+						b.Cleanup(func() { require.NoError(b, r.Close()) })
+						reader = r
+					default:
+						b.Fatalf("unknown reader: %s", readerName)
+					}
+
+					b.ResetTimer()
 					for i := 0; i < b.N; i++ {
 						startBytesDiscarded := reader.BufReaderStats().BytesDiscarded.Load()
 						values, err := reader.LabelValuesOffsets(context.Background(), lbl, tc.prefix, tc.filter)
@@ -309,6 +345,39 @@ func BenchmarkLabelValuesOffsetsIndexV2_WithPrefix(b *testing.B) {
 						diff := endBytesDiscarded - startBytesDiscarded
 						output := float64(diff)
 						b.ReportMetric(output, "buffered-bytes-discarded/op")
+
+						metrics, err := cacheBucketReg.Gather()
+						require.NoError(b, err)
+
+						promToBenchmarkMetrics := map[string]string{
+							"thanos_store_bucket_cache_operation_requests_total":       "get-range-ops",
+							"thanos_store_bucket_cache_operation_hits_total":           "get-range-hits",
+							"thanos_store_bucket_cache_getrange_requested_bytes_total": "get-range-bytes-requested",
+							"thanos_store_bucket_cache_getrange_fetched_bytes_total":   "get-range-bytes-fetched",
+							"thanos_store_bucket_cache_getrange_refetched_bytes_total": "get-range-bytes-refetched",
+						}
+					MetricFamilyLoop:
+						for _, mf := range metrics {
+						MetricLoop:
+							for _, m := range mf.GetMetric() {
+								name := mf.GetName()
+
+								benchMetricName, ok := promToBenchmarkMetrics[name]
+								if !ok {
+									continue MetricFamilyLoop
+								}
+
+								for _, l := range m.GetLabel() {
+									lName, lValue := l.GetName(), l.GetValue()
+									if lName == "operation" && lValue != "get_range" {
+										continue MetricLoop
+									}
+								}
+
+								value := m.GetCounter().GetValue()
+								b.ReportMetric(value, benchMetricName+"/op")
+							}
+						}
 					}
 				})
 			}

pkg/storage/indexheader/stream_binary_reader.go

@@ -43,9 +43,8 @@ func NewStreamBinaryReaderMetrics(reg prometheus.Registerer) *StreamBinaryReader
 }
 
 type StreamBinaryReader struct {
-	factory        *streamencoding.DecbufFactory
-	bufReaderStats *streamencoding.BufReaderStats
-	toc            *BinaryTOC
+	factory *streamencoding.DecbufFactory
+	toc     *BinaryTOC
 
 	// Symbols struct that keeps only 1/postingOffsetsInMemSampling in the memory, then looks up the
 	// rest via seeking to offsets in the index-header.
@@ -109,8 +108,7 @@ func NewFileStreamBinaryReader(ctx context.Context, binPath string, id ulid.ULID
 	logger = log.With(logger, "id", id, "path", sparseHeadersPath, "inmem_sampling_rate", postingOffsetsInMemSampling)
 
 	r := &StreamBinaryReader{
-		factory:        streamencoding.NewDecbufFactory(binPath, cfg.MaxIdleFileHandles, metrics.decbufFactory),
-		bufReaderStats: &streamencoding.BufReaderStats{},
+		factory: streamencoding.NewDecbufFactory(binPath, cfg.MaxIdleFileHandles, metrics.decbufFactory),
 	}
 
 	// Create a new raw decoding buffer with access to the entire index-header file to