chore(blooms): Add setting to fetch bloom blocks async or synchronous…

…ly (#14862)

In case bloom blocks are small, they could be downloaded synchronously to avoid missing blocks on cold start.
This is an undocumented setting.

Signed-off-by: Christian Haudum <christian.haudum@gmail.com>

pkg/bloomgateway/bloomgateway.go

@@ -82,6 +82,7 @@ func New(cfg Config, store bloomshipper.Store, logger log.Logger, reg prometheus
 		workerConfig: workerConfig{
 			maxItems:         cfg.NumMultiplexItems,
 			queryConcurrency: cfg.BlockQueryConcurrency,
+			async:            cfg.FetchBlocksAsync,
 		},
 		pendingTasks: &atomic.Int64{},
 

pkg/bloomgateway/config.go

@@ -11,10 +11,11 @@ type Config struct {
 	// Client configures the Bloom Gateway client
 	Client ClientConfig `yaml:"client,omitempty" doc:""`
 
-	WorkerConcurrency       int `yaml:"worker_concurrency"`
-	BlockQueryConcurrency   int `yaml:"block_query_concurrency"`
-	MaxOutstandingPerTenant int `yaml:"max_outstanding_per_tenant"`
-	NumMultiplexItems       int `yaml:"num_multiplex_tasks"`
+	WorkerConcurrency       int  `yaml:"worker_concurrency"`
+	BlockQueryConcurrency   int  `yaml:"block_query_concurrency"`
+	MaxOutstandingPerTenant int  `yaml:"max_outstanding_per_tenant"`
+	NumMultiplexItems       int  `yaml:"num_multiplex_tasks"`
+	FetchBlocksAsync        bool `yaml:"fetch_blocks_async" doc:"hidden"`
 }
 
 // RegisterFlags registers flags for the Bloom Gateway configuration.
@@ -29,6 +30,7 @@ func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	f.IntVar(&cfg.BlockQueryConcurrency, prefix+"block-query-concurrency", 8, "Number of blocks processed concurrently on a single worker. Usually set to 2x number of CPU cores.")
 	f.IntVar(&cfg.MaxOutstandingPerTenant, prefix+"max-outstanding-per-tenant", 1024, "Maximum number of outstanding tasks per tenant.")
 	f.IntVar(&cfg.NumMultiplexItems, prefix+"num-multiplex-tasks", 512, "How many tasks are multiplexed at once.")
+	f.BoolVar(&cfg.FetchBlocksAsync, prefix+"fetch-blocks-async", true, "Whether blocks should be fetched asynchronously.")
 	// TODO(chaudum): Figure out what the better place is for registering flags
 	// -bloom-gateway.client.* or -bloom-gateway-client.*
 	cfg.Client.RegisterFlags(f)

pkg/bloomgateway/processor.go

@@ -16,10 +16,11 @@ import (
 	"github.com/grafana/loki/v3/pkg/storage/stores/shipper/bloomshipper"
 )
 
-func newProcessor(id string, concurrency int, store bloomshipper.Store, logger log.Logger, metrics *workerMetrics) *processor {
+func newProcessor(id string, concurrency int, async bool, store bloomshipper.Store, logger log.Logger, metrics *workerMetrics) *processor {
 	return &processor{
 		id:          id,
 		concurrency: concurrency,
+		async:       async,
 		store:       store,
 		logger:      logger,
 		metrics:     metrics,
@@ -28,7 +29,8 @@ func newProcessor(id string, concurrency int, store bloomshipper.Store, logger l
 
 type processor struct {
 	id          string
-	concurrency int // concurrency at which bloom blocks are processed
+	concurrency int  // concurrency at which bloom blocks are processed
+	async       bool // whether blocks should be downloaded asynchronously
 	store       bloomshipper.Store
 	logger      log.Logger
 	metrics     *workerMetrics
@@ -71,7 +73,7 @@ func (p *processor) processTasksForDay(ctx context.Context, _ string, _ config.D
 	bqs, err := p.store.FetchBlocks(
 		ctx,
 		refs,
-		bloomshipper.WithFetchAsync(true),
+		bloomshipper.WithFetchAsync(p.async),
 		bloomshipper.WithIgnoreNotFound(true),
 		// NB(owen-d): we relinquish bloom pages to a pool
 		// after iteration for performance (alloc reduction).

pkg/bloomgateway/processor_test.go

@@ -128,7 +128,7 @@ func TestProcessor(t *testing.T) {
 		refs, metas, queriers, data := createBlocks(t, tenant, 10, now.Add(-1*time.Hour), now, 0x0000, 0x0fff)
 
 		mockStore := newMockBloomStore(refs, queriers, metas)
-		p := newProcessor("worker", 1, mockStore, log.NewNopLogger(), metrics)
+		p := newProcessor("worker", 1, false, mockStore, log.NewNopLogger(), metrics)
 
 		chunkRefs := createQueryInputFromBlockData(t, tenant, data, 10)
 		swb := seriesWithInterval{
@@ -179,7 +179,7 @@ func TestProcessor(t *testing.T) {
 		}
 
 		mockStore := newMockBloomStore(refs, queriers, metas)
-		p := newProcessor("worker", 1, mockStore, log.NewNopLogger(), metrics)
+		p := newProcessor("worker", 1, false, mockStore, log.NewNopLogger(), metrics)
 
 		chunkRefs := createQueryInputFromBlockData(t, tenant, data, 10)
 		swb := seriesWithInterval{
@@ -226,7 +226,7 @@ func TestProcessor(t *testing.T) {
 		mockStore := newMockBloomStore(refs, queriers, metas)
 		mockStore.err = errors.New("store failed")
 
-		p := newProcessor("worker", 1, mockStore, log.NewNopLogger(), metrics)
+		p := newProcessor("worker", 1, false, mockStore, log.NewNopLogger(), metrics)
 
 		chunkRefs := createQueryInputFromBlockData(t, tenant, data, 10)
 		swb := seriesWithInterval{

pkg/bloomgateway/worker.go

@@ -22,6 +22,7 @@ const (
 type workerConfig struct {
 	maxItems         int
 	queryConcurrency int
+	async            bool
 }
 
 // worker is a datastructure that consumes tasks from the request queue,
@@ -64,7 +65,7 @@ func (w *worker) starting(_ context.Context) error {
 func (w *worker) running(_ context.Context) error {
 	idx := queue.StartIndexWithLocalQueue
 
-	p := newProcessor(w.id, w.cfg.queryConcurrency, w.store, w.logger, w.metrics)
+	p := newProcessor(w.id, w.cfg.queryConcurrency, w.cfg.async, w.store, w.logger, w.metrics)
 
 	for st := w.State(); st == services.Running || st == services.Stopping; {
 		taskCtx := context.Background()