Store Gateway index stats response (#3206)

* WIP gateway stats response

* Update comment and fix linter issue

* update license header for new rpc.proto

* Refactor for moved proto files

* rename stats entry to reflect other usages

* add series fetched to the index fetch bytes total

* add a comment and move to uint64

* Update pkg/storegateway/storepb/rpc.proto

Co-authored-by: Marco Pracucci <marco@pracucci.com>

pkg/frontend/transport/handler.go

@@ -200,6 +200,7 @@ func (f *Handler) reportQueryStats(r *http.Request, queryString url.Values, quer
 	numSeries := stats.LoadFetchedSeries()
 	numBytes := stats.LoadFetchedChunkBytes()
 	numChunks := stats.LoadFetchedChunks()
+	numIndexBytes := stats.LoadFetchedIndexBytes()
 	sharded := strconv.FormatBool(stats.GetShardedQueries() > 0)
 
 	if stats != nil {
@@ -222,6 +223,7 @@ func (f *Handler) reportQueryStats(r *http.Request, queryString url.Values, quer
 		"fetched_series_count", numSeries,
 		"fetched_chunk_bytes", numBytes,
 		"fetched_chunks_count", numChunks,
+		"fetched_index_bytes", numIndexBytes,
 		"sharded_queries", stats.LoadShardedQueries(),
 		"split_queries", stats.LoadSplitQueries(),
 	}, formatQueryString(queryString)...)

pkg/querier/blocks_store_queryable.go

@@ -728,6 +728,7 @@ func (q *blocksStoreQuerier) fetchSeriesFromStores(
 			mySeries := []*storepb.Series(nil)
 			myWarnings := storage.Warnings(nil)
 			myQueriedBlocks := []ulid.ULID(nil)
+			indexBytesFetched := uint64(0)
 
 			for {
 				// Ensure the context hasn't been canceled in the meanwhile (eg. an error occurred
@@ -789,6 +790,10 @@ func (q *blocksStoreQuerier) fetchSeriesFromStores(
 
 					myQueriedBlocks = append(myQueriedBlocks, ids...)
 				}
+
+				if s := resp.GetStats(); s != nil {
+					indexBytesFetched += s.FetchedIndexBytes
+				}
 			}
 
 			numSeries := len(mySeries)
@@ -797,12 +802,14 @@ func (q *blocksStoreQuerier) fetchSeriesFromStores(
 			reqStats.AddFetchedSeries(uint64(numSeries))
 			reqStats.AddFetchedChunkBytes(uint64(chunkBytes))
 			reqStats.AddFetchedChunks(uint64(chunksFetched))
+			reqStats.AddFetchedIndexBytes(indexBytesFetched)
 
 			level.Debug(spanLog).Log("msg", "received series from store-gateway",
 				"instance", c.RemoteAddress(),
 				"fetched series", numSeries,
 				"fetched chunk bytes", chunkBytes,
 				"fetched chunks", chunksFetched,
+				"fetched index bytes", indexBytesFetched,
 				"requested blocks", strings.Join(convertULIDsToString(blockIDs), " "),
 				"queried blocks", strings.Join(convertULIDsToString(myQueriedBlocks), " "))
 

pkg/querier/blocks_store_queryable_test.go

@@ -117,6 +117,7 @@ func TestBlocksStoreQuerier_Select(t *testing.T) {
 						mockSeriesResponse(labels.Labels{metricNameLabel}, minT, 1),
 						mockSeriesResponse(labels.Labels{metricNameLabel}, minT+1, 2),
 						mockHintsResponse(block1, block2),
+						mockStatsResponse(50),
 					}}: {block1, block2},
 				},
 			},
@@ -1928,6 +1929,14 @@ func mockSeriesResponseWithChunks(lbls labels.Labels, chunks ...storepb.AggrChun
 	}
 }
 
+func mockStatsResponse(fetchedIndexBytes int) *storepb.SeriesResponse {
+	return &storepb.SeriesResponse{
+		Result: &storepb.SeriesResponse_Stats{
+			Stats: &storepb.Stats{FetchedIndexBytes: uint64(fetchedIndexBytes)},
+		},
+	}
+}
+
 func mockHintsResponse(ids ...ulid.ULID) *storepb.SeriesResponse {
 	hints := &hintspb.SeriesResponseHints{}
 	for _, id := range ids {

pkg/querier/stats/stats.go

@@ -107,6 +107,22 @@ func (s *Stats) LoadFetchedChunks() uint64 {
 	return atomic.LoadUint64(&s.FetchedChunksCount)
 }
 
+func (s *Stats) AddFetchedIndexBytes(indexBytes uint64) {
+	if s == nil {
+		return
+	}
+
+	atomic.AddUint64(&s.FetchedIndexBytes, indexBytes)
+}
+
+func (s *Stats) LoadFetchedIndexBytes() uint64 {
+	if s == nil {
+		return 0
+	}
+
+	return atomic.LoadUint64(&s.FetchedIndexBytes)
+}
+
 func (s *Stats) AddShardedQueries(num uint32) {
 	if s == nil {
 		return
@@ -151,6 +167,7 @@ func (s *Stats) Merge(other *Stats) {
 	s.AddFetchedChunks(other.LoadFetchedChunks())
 	s.AddShardedQueries(other.LoadShardedQueries())
 	s.AddSplitQueries(other.LoadSplitQueries())
+	s.AddFetchedIndexBytes(other.LoadFetchedIndexBytes())
 }
 
 func ShouldTrackHTTPGRPCResponse(r *httpgrpc.HTTPResponse) bool {

pkg/querier/stats/stats.proto

@@ -28,4 +28,6 @@ message Stats {
   uint32 sharded_queries = 5;
   // The number of split partial queries executed. 0 if splitting is disabled or the query can't be split.
   uint32 split_queries = 6;
+  // The number of index bytes fetched on the store-gateway for the query
+  uint64 fetched_index_bytes = 7;
 }

pkg/storegateway/bucket.go

@@ -1063,6 +1063,11 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 		}
 	}
 
+	if err = srv.Send(storepb.NewStatsResponse(stats.postingsFetchedSizeSum + stats.seriesFetchedSizeSum)); err != nil {
+		err = status.Error(codes.Unknown, errors.Wrap(err, "sends series response stats").Error())
+		return
+	}
+
 	return err
 }
 

pkg/storegateway/storepb/custom.go

@@ -32,6 +32,14 @@ func NewHintsSeriesResponse(hints *types.Any) *SeriesResponse {
 	}
 }
 
+func NewStatsResponse(indexBytesFetched int) *SeriesResponse {
+	return &SeriesResponse{
+		Result: &SeriesResponse_Stats{
+			Stats: &Stats{FetchedIndexBytes: uint64(indexBytesFetched)},
+		},
+	}
+}
+
 type emptySeriesSet struct{}
 
 func (emptySeriesSet) Next() bool                       { return false }