Stream chunks from storegateway to queriers

CHANGELOG.md

@@ -7,6 +7,7 @@
 * [CHANGE] Store-gateway: skip verifying index header integrity upon loading. To enable verification set `blocks_storage.bucket_store.index_header.verify_on_load: true`.
 * [CHANGE] Querier: change the default value of the experimental `-querier.streaming-chunks-per-ingester-buffer-size` flag to 256. #5203
 * [CHANGE] Querier: only initiate query requests to ingesters in the `ACTIVE` state in the ring. #5342
+* [CHANGE] Querier: Renamed `-querier.prefer-streaming-chunks` to `-querier.prefer-streaming-chunks-from-ingesters` to enable streaming chunks from ingesters to queriers. #5182
 * [CHANGE] Querier: `-query-frontend.cache-unaligned-requests` has been moved from a global flag to a per-tenant override. #5312
 * [CHANGE] Ingester: removed `cortex_ingester_shipper_dir_syncs_total` and `cortex_ingester_shipper_dir_sync_failures_total` metrics. The former metric was not much useful, and the latter was never incremented. #5396
 * [FEATURE] Cardinality API: Add a new `count_method` parameter which enables counting active series #5136
@@ -42,6 +43,7 @@
 * [ENHANCEMENT] Querier: add `cortex_querier_queries_rejected_total` metric that counts the number of queries rejected due to hitting a limit (eg. max series per query or max chunks per query). #5316 #5440
 * [ENHANCEMENT] Querier: add experimental `-querier.minimize-ingester-requests-hedging-delay` option to initiate requests to further ingesters when request minimisation is enabled and not all initial requests have completed. #5368
 * [ENHANCEMENT] Clarify docs for `-ingester.client.*` flags to make it clear that these are used by both queriers and distributors. #5375
+* [ENHANCEMENT] Querier and store-gateway: add experimental support for streaming chunks from store-gateways to queriers while evaluating queries. This can be enabled with `-querier.prefer-streaming-chunks-from-store-gateways=true`. #5182
 * [ENHANCEMENT] Querier: enforce `max-chunks-per-query` limit earlier in query processing when streaming chunks from ingesters to queriers to avoid unnecessarily consuming resources for queries that will be aborted. #5369 #5447
 * [ENHANCEMENT] Ingester: added `cortex_ingester_shipper_last_successful_upload_timestamp_seconds` metric tracking the last successful TSDB block uploaded to the bucket (unix timestamp in seconds). #5396
 * [BUGFIX] Ingester: Handle when previous ring state is leaving and the number of tokens has changed. #5204

cmd/mimir/config-descriptor.json

@@ -1642,12 +1642,23 @@
         },
         {
           "kind": "field",
-          "name": "prefer_streaming_chunks",
+          "name": "prefer_streaming_chunks_from_ingesters",
           "required": false,
           "desc": "Request ingesters stream chunks. Ingesters will only respond with a stream of chunks if the target ingester supports this, and this preference will be ignored by ingesters that do not support this.",
           "fieldValue": null,
           "fieldDefaultValue": false,
-          "fieldFlag": "querier.prefer-streaming-chunks",
+          "fieldFlag": "querier.prefer-streaming-chunks-from-ingesters",
+          "fieldType": "boolean",
+          "fieldCategory": "experimental"
+        },
+        {
+          "kind": "field",
+          "name": "prefer_streaming_chunks_from_store_gateways",
+          "required": false,
+          "desc": "Request store-gateways stream chunks. Store-gateways will only respond with a stream of chunks if the target store-gateway supports this, and this preference will be ignored by store-gateways that do not support this.",
+          "fieldValue": null,
+          "fieldDefaultValue": false,
+          "fieldFlag": "querier.prefer-streaming-chunks-from-store-gateways",
           "fieldType": "boolean",
           "fieldCategory": "experimental"
         },
@@ -1662,6 +1673,17 @@
           "fieldType": "int",
           "fieldCategory": "experimental"
         },
+        {
+          "kind": "field",
+          "name": "streaming_chunks_per_store_gateway_series_buffer_size",
+          "required": false,
+          "desc": "Number of series to buffer per store-gateway when streaming chunks from store-gateways.",
+          "fieldValue": null,
+          "fieldDefaultValue": 256,
+          "fieldFlag": "querier.streaming-chunks-per-store-gateway-buffer-size",
+          "fieldType": "int",
+          "fieldCategory": "experimental"
+        },
         {
           "kind": "field",
           "name": "minimize_ingester_requests",

cmd/mimir/help-all.txt.tmpl

@@ -1611,8 +1611,10 @@ Usage of ./cmd/mimir/mimir:
     	[experimental] If true, when querying ingesters, only the minimum required ingesters required to reach quorum will be queried initially, with other ingesters queried only if needed due to failures from the initial set of ingesters. Enabling this option reduces resource consumption for the happy path at the cost of increased latency for the unhappy path.
   -querier.minimize-ingester-requests-hedging-delay duration
     	[experimental] Delay before initiating requests to further ingesters when request minimization is enabled and the initially selected set of ingesters have not all responded. Ignored if -querier.minimize-ingester-requests is not enabled. (default 3s)
-  -querier.prefer-streaming-chunks
+  -querier.prefer-streaming-chunks-from-ingesters
     	[experimental] Request ingesters stream chunks. Ingesters will only respond with a stream of chunks if the target ingester supports this, and this preference will be ignored by ingesters that do not support this.
+  -querier.prefer-streaming-chunks-from-store-gateways
+    	[experimental] Request store-gateways stream chunks. Store-gateways will only respond with a stream of chunks if the target store-gateway supports this, and this preference will be ignored by store-gateways that do not support this.
   -querier.query-ingesters-within duration
     	Maximum lookback beyond which queries are not sent to ingester. 0 means all queries are sent to ingester. (default 13h)
   -querier.query-store-after duration
@@ -1639,6 +1641,8 @@ Usage of ./cmd/mimir/mimir:
     	Override the expected name on the server certificate.
   -querier.streaming-chunks-per-ingester-buffer-size uint
     	[experimental] Number of series to buffer per ingester when streaming chunks from ingesters. (default 256)
+  -querier.streaming-chunks-per-store-gateway-buffer-size uint
+    	[experimental] Number of series to buffer per store-gateway when streaming chunks from store-gateways. (default 256)
   -querier.timeout duration
     	The timeout for a query. This config option should be set on query-frontend too when query sharding is enabled. This also applies to queries evaluated by the ruler (internally or remotely). (default 2m0s)
   -query-frontend.align-queries-with-step

docs/sources/mimir/references/configuration-parameters/index.md

@@ -1096,14 +1096,25 @@ store_gateway_client:
 # (experimental) Request ingesters stream chunks. Ingesters will only respond
 # with a stream of chunks if the target ingester supports this, and this
 # preference will be ignored by ingesters that do not support this.
-# CLI flag: -querier.prefer-streaming-chunks
-[prefer_streaming_chunks: <boolean> | default = false]
+# CLI flag: -querier.prefer-streaming-chunks-from-ingesters
+[prefer_streaming_chunks_from_ingesters: <boolean> | default = false]
+
+# (experimental) Request store-gateways stream chunks. Store-gateways will only
+# respond with a stream of chunks if the target store-gateway supports this, and
+# this preference will be ignored by store-gateways that do not support this.
+# CLI flag: -querier.prefer-streaming-chunks-from-store-gateways
+[prefer_streaming_chunks_from_store_gateways: <boolean> | default = false]
 
 # (experimental) Number of series to buffer per ingester when streaming chunks
 # from ingesters.
 # CLI flag: -querier.streaming-chunks-per-ingester-buffer-size
 [streaming_chunks_per_ingester_series_buffer_size: <int> | default = 256]
 
+# (experimental) Number of series to buffer per store-gateway when streaming
+# chunks from store-gateways.
+# CLI flag: -querier.streaming-chunks-per-store-gateway-buffer-size
+[streaming_chunks_per_store_gateway_series_buffer_size: <int> | default = 256]
+
 # (experimental) If true, when querying ingesters, only the minimum required
 # ingesters required to reach quorum will be queried initially, with other
 # ingesters queried only if needed due to failures from the initial set of

integration/ingester_test.go

@@ -480,9 +480,9 @@ func TestIngesterQuerying(t *testing.T) {
 					defer s.Close()
 
 					baseFlags := map[string]string{
-						"-distributor.ingestion-tenant-shard-size": "0",
-						"-ingester.ring.heartbeat-period":          "1s",
-						"-querier.prefer-streaming-chunks":         strconv.FormatBool(streamingEnabled),
+						"-distributor.ingestion-tenant-shard-size":        "0",
+						"-ingester.ring.heartbeat-period":                 "1s",
+						"-querier.prefer-streaming-chunks-from-ingesters": strconv.FormatBool(streamingEnabled),
 					}
 
 					flags := mergeFlags(
@@ -543,12 +543,12 @@ func TestIngesterQueryingWithRequestMinimization(t *testing.T) {
 			defer s.Close()
 
 			baseFlags := map[string]string{
-				"-distributor.ingestion-tenant-shard-size": "0",
-				"-ingester.ring.heartbeat-period":          "1s",
-				"-ingester.ring.zone-awareness-enabled":    "true",
-				"-ingester.ring.replication-factor":        "3",
-				"-querier.minimize-ingester-requests":      "true",
-				"-querier.prefer-streaming-chunks":         strconv.FormatBool(streamingEnabled),
+				"-distributor.ingestion-tenant-shard-size":        "0",
+				"-ingester.ring.heartbeat-period":                 "1s",
+				"-ingester.ring.zone-awareness-enabled":           "true",
+				"-ingester.ring.replication-factor":               "3",
+				"-querier.minimize-ingester-requests":             "true",
+				"-querier.prefer-streaming-chunks-from-ingesters": strconv.FormatBool(streamingEnabled),
 			}
 
 			flags := mergeFlags(

integration/querier_test.go

@@ -28,14 +28,22 @@ import (
 )
 
 func TestQuerierWithBlocksStorageRunningInMicroservicesMode(t *testing.T) {
-	testQuerierWithBlocksStorageRunningInMicroservicesMode(t, generateFloatSeries)
+	for _, streamingEnabled := range []bool{true, false} {
+		t.Run(fmt.Sprintf("streaming=%t", streamingEnabled), func(t *testing.T) {
+			testQuerierWithBlocksStorageRunningInMicroservicesMode(t, streamingEnabled, generateFloatSeries)
+		})
+	}
 }
 
 func TestQuerierWithBlocksStorageRunningInMicroservicesModeWithHistograms(t *testing.T) {
-	testQuerierWithBlocksStorageRunningInMicroservicesMode(t, generateHistogramSeries)
+	for _, streamingEnabled := range []bool{true, false} {
+		t.Run(fmt.Sprintf("streaming=%t", streamingEnabled), func(t *testing.T) {
+			testQuerierWithBlocksStorageRunningInMicroservicesMode(t, streamingEnabled, generateHistogramSeries)
+		})
+	}
 }
 
-func testQuerierWithBlocksStorageRunningInMicroservicesMode(t *testing.T, seriesGenerator func(name string, ts time.Time, additionalLabels ...prompb.Label) (series []prompb.TimeSeries, vector model.Vector, matrix model.Matrix)) {
+func testQuerierWithBlocksStorageRunningInMicroservicesMode(t *testing.T, streamingEnabled bool, seriesGenerator func(name string, ts time.Time, additionalLabels ...prompb.Label) (series []prompb.TimeSeries, vector model.Vector, matrix model.Matrix)) {
 	tests := map[string]struct {
 		tenantShardSize      int
 		indexCacheBackend    string
@@ -162,6 +170,8 @@ func testQuerierWithBlocksStorageRunningInMicroservicesMode(t *testing.T, series
 				"-store-gateway.tenant-shard-size":                             fmt.Sprintf("%d", testCfg.tenantShardSize),
 				"-query-frontend.query-stats-enabled":                          "true",
 				"-query-frontend.parallelize-shardable-queries":                strconv.FormatBool(testCfg.queryShardingEnabled),
+				"-querier.prefer-streaming-chunks-from-ingesters":              strconv.FormatBool(streamingEnabled),
+				"-querier.prefer-streaming-chunks-from-store-gateways":         strconv.FormatBool(streamingEnabled),
 			})
 
 			// Start store-gateways.
@@ -239,15 +249,21 @@ func testQuerierWithBlocksStorageRunningInMicroservicesMode(t *testing.T, series
 			// thanos_store_index_cache_requests_total: ExpandedPostings: 5, Postings: 2, Series: 2
 			instantQueriesCount++
 
+			comparingFunction := e2e.Equals
+			if streamingEnabled {
+				// Some metrics can be higher when streaming is enabled. The exact number is not deterministic in every case.
+				comparingFunction = e2e.GreaterOrEqual
+			}
+
 			// Check the in-memory index cache metrics (in the store-gateway).
-			require.NoError(t, storeGateways.WaitSumMetrics(e2e.Equals(9), "thanos_store_index_cache_requests_total")) // 5 + 2 + 2
-			require.NoError(t, storeGateways.WaitSumMetrics(e2e.Equals(0), "thanos_store_index_cache_hits_total"))     // no cache hit cause the cache was empty
+			require.NoError(t, storeGateways.WaitSumMetrics(comparingFunction(9), "thanos_store_index_cache_requests_total")) // 5 + 2 + 2
+			require.NoError(t, storeGateways.WaitSumMetrics(comparingFunction(0), "thanos_store_index_cache_hits_total"))     // no cache hit cause the cache was empty
 
 			if testCfg.indexCacheBackend == tsdb.IndexCacheBackendInMemory {
 				require.NoError(t, storeGateways.WaitSumMetrics(e2e.Equals(2*2+2+3), "thanos_store_index_cache_items"))             // 2 series both for postings and series cache, 2 expanded postings on one block, 3 on another one
 				require.NoError(t, storeGateways.WaitSumMetrics(e2e.Equals(2*2+2+3), "thanos_store_index_cache_items_added_total")) // 2 series both for postings and series cache, 2 expanded postings on one block, 3 on another one
 			} else if testCfg.indexCacheBackend == tsdb.IndexCacheBackendMemcached {
-				require.NoError(t, storeGateways.WaitSumMetrics(e2e.Equals(9*2), "thanos_memcached_operations_total")) // one set for each get
+				require.NoError(t, storeGateways.WaitSumMetrics(comparingFunction(9*2), "thanos_memcached_operations_total")) // one set for each get
 			}
 
 			// Query back again the 1st series from storage. This time it should use the index cache.
@@ -257,14 +273,14 @@ func testQuerierWithBlocksStorageRunningInMicroservicesMode(t *testing.T, series
 			assert.Equal(t, expectedVector1, result.(model.Vector))
 			expectedFetchedSeries++ // Storage only.
 
-			require.NoError(t, storeGateways.WaitSumMetrics(e2e.Equals(9+2), "thanos_store_index_cache_requests_total"))
-			require.NoError(t, storeGateways.WaitSumMetrics(e2e.Equals(2), "thanos_store_index_cache_hits_total")) // this time has used the index cache
+			require.NoError(t, storeGateways.WaitSumMetrics(comparingFunction(9+2), "thanos_store_index_cache_requests_total"))
+			require.NoError(t, storeGateways.WaitSumMetrics(comparingFunction(2), "thanos_store_index_cache_hits_total")) // this time has used the index cache
 
 			if testCfg.indexCacheBackend == tsdb.IndexCacheBackendInMemory {
 				require.NoError(t, storeGateways.WaitSumMetrics(e2e.Equals(2*2+2+3), "thanos_store_index_cache_items"))             // as before
 				require.NoError(t, storeGateways.WaitSumMetrics(e2e.Equals(2*2+2+3), "thanos_store_index_cache_items_added_total")) // as before
 			} else if testCfg.indexCacheBackend == tsdb.IndexCacheBackendMemcached {
-				require.NoError(t, storeGateways.WaitSumMetrics(e2e.Equals(9*2+2), "thanos_memcached_operations_total")) // as before + 2 gets (expanded postings and series)
+				require.NoError(t, storeGateways.WaitSumMetrics(comparingFunction(9*2+2), "thanos_memcached_operations_total")) // as before + 2 gets (expanded postings and series)
 			}
 
 			// Query range. We expect 1 data point with a value of 3 (number of series).
@@ -877,12 +893,13 @@ func TestQueryLimitsWithBlocksStorageRunningInMicroServices(t *testing.T) {
 			// Configure the blocks storage to frequently compact TSDB head
 			// and ship blocks to the storage.
 			flags := mergeFlags(BlocksStorageFlags(), BlocksStorageS3Flags(), map[string]string{
-				"-blocks-storage.tsdb.block-ranges-period":   blockRangePeriod.String(),
-				"-blocks-storage.tsdb.ship-interval":         "1s",
-				"-blocks-storage.bucket-store.sync-interval": "1s",
-				"-blocks-storage.tsdb.retention-period":      ((blockRangePeriod * 2) - 1).String(),
-				"-querier.max-fetched-series-per-query":      "3",
-				"-querier.prefer-streaming-chunks":           strconv.FormatBool(streamingEnabled),
+				"-blocks-storage.tsdb.block-ranges-period":             blockRangePeriod.String(),
+				"-blocks-storage.tsdb.ship-interval":                   "1s",
+				"-blocks-storage.bucket-store.sync-interval":           "1s",
+				"-blocks-storage.tsdb.retention-period":                ((blockRangePeriod * 2) - 1).String(),
+				"-querier.max-fetched-series-per-query":                "3",
+				"-querier.prefer-streaming-chunks-from-ingesters":      strconv.FormatBool(streamingEnabled),
+				"-querier.prefer-streaming-chunks-from-store-gateways": strconv.FormatBool(streamingEnabled),
 			})
 
 			// Start dependencies.

pkg/distributor/distributor.go

@@ -166,7 +166,7 @@ type Config struct {
 
 	// This config is dynamically injected because it is defined in the querier config.
 	ShuffleShardingLookbackPeriod              time.Duration `yaml:"-"`
-	PreferStreamingChunks                      bool          `yaml:"-"`
+	PreferStreamingChunksFromIngesters         bool          `yaml:"-"`
 	StreamingChunksPerIngesterSeriesBufferSize uint64        `yaml:"-"`
 	MinimizeIngesterRequests                   bool          `yaml:"-"`
 	MinimiseIngesterRequestsHedgingDelay       time.Duration `yaml:"-"`

pkg/distributor/distributor_test.go

@@ -3023,7 +3023,7 @@ func prepare(t *testing.T, cfg prepConfig) ([]*Distributor, []mockIngester, []*p
 		distributorCfg.DefaultLimits.MaxInflightPushRequestsBytes = cfg.maxInflightRequestsBytes
 		distributorCfg.DefaultLimits.MaxIngestionRate = cfg.maxIngestionRate
 		distributorCfg.ShuffleShardingLookbackPeriod = time.Hour
-		distributorCfg.PreferStreamingChunks = cfg.preferStreamingChunks
+		distributorCfg.PreferStreamingChunksFromIngesters = cfg.preferStreamingChunks
 		distributorCfg.StreamingChunksPerIngesterSeriesBufferSize = 128
 
 		cfg.limits.IngestionTenantShardSize = cfg.shuffleShardSize

pkg/distributor/query.go

@@ -71,7 +71,7 @@ func (d *Distributor) QueryStream(ctx context.Context, from, to model.Time, matc
 			return err
 		}
 
-		if d.cfg.PreferStreamingChunks {
+		if d.cfg.PreferStreamingChunksFromIngesters {
 			req.StreamingChunksBatchSize = d.cfg.StreamingChunksPerIngesterSeriesBufferSize
 		}
 

pkg/mimir/modules.go

@@ -361,7 +361,7 @@ func (t *Mimir) initDistributorService() (serv services.Service, err error) {
 	// ruler's dependency)
 	canJoinDistributorsRing := t.Cfg.isAnyModuleEnabled(Distributor, Write, All)
 
-	t.Cfg.Distributor.PreferStreamingChunks = t.Cfg.Querier.PreferStreamingChunks
+	t.Cfg.Distributor.PreferStreamingChunksFromIngesters = t.Cfg.Querier.PreferStreamingChunksFromIngesters
 	t.Cfg.Distributor.StreamingChunksPerIngesterSeriesBufferSize = t.Cfg.Querier.StreamingChunksPerIngesterSeriesBufferSize
 	t.Cfg.Distributor.MinimizeIngesterRequests = t.Cfg.Querier.MinimizeIngesterRequests
 	t.Cfg.Distributor.MinimiseIngesterRequestsHedgingDelay = t.Cfg.Querier.MinimiseIngesterRequestsHedgingDelay