add fetched samples and chunks count into querier stats

Signed-off-by: Ben Ye <[email protected]>

Author: yeya24

pkg/distributor/query.go

@@ -400,6 +400,8 @@ func (d *Distributor) queryIngesterStream(ctx context.Context, replicationSet ri
 	reqStats.AddFetchedSeries(uint64(len(resp.Chunkseries) + len(resp.Timeseries)))
 	reqStats.AddFetchedChunkBytes(uint64(resp.ChunksSize()))
 	reqStats.AddFetchedDataBytes(uint64(resp.Size()))
+	reqStats.AddFetchedChunks(uint64(resp.ChunksCount()))
+	reqStats.AddFetchedSamples(uint64(resp.SamplesCount()))
 
 	return resp, nil
 }

pkg/frontend/transport/handler.go

@@ -64,6 +64,8 @@ type Handler struct {
 	// Metrics.
 	querySeconds   *prometheus.CounterVec
 	querySeries    *prometheus.CounterVec
+	querySamples   *prometheus.CounterVec
+	queryChunks    *prometheus.CounterVec
 	queryBytes     *prometheus.CounterVec
 	queryDataBytes *prometheus.CounterVec
 	activeUsers    *util.ActiveUsersCleanupService
@@ -230,13 +232,15 @@ func (f *Handler) reportQueryStats(r *http.Request, queryString url.Values, quer
 	userID := tenant.JoinTenantIDs(tenantIDs)
 	wallTime := stats.LoadWallTime()
 	numSeries := stats.LoadFetchedSeries()
-	numBytes := stats.LoadFetchedChunkBytes()
+	numChunks := stats.LoadFetchedChunks()
+	numSamples := stats.LoadFetchedSamples()
+	numChunkBytes := stats.LoadFetchedChunkBytes()
 	numDataBytes := stats.LoadFetchedDataBytes()
 
 	// Track stats.
 	f.querySeconds.WithLabelValues(userID).Add(wallTime.Seconds())
 	f.querySeries.WithLabelValues(userID).Add(float64(numSeries))
-	f.queryBytes.WithLabelValues(userID).Add(float64(numBytes))
+	f.queryBytes.WithLabelValues(userID).Add(float64(numChunkBytes))
 	f.queryDataBytes.WithLabelValues(userID).Add(float64(numDataBytes))
 	f.activeUsers.UpdateUserTimestamp(userID, time.Now())
 
@@ -249,7 +253,9 @@ func (f *Handler) reportQueryStats(r *http.Request, queryString url.Values, quer
 		"response_time", queryResponseTime,
 		"query_wall_time_seconds", wallTime.Seconds(),
 		"fetched_series_count", numSeries,
-		"fetched_chunks_bytes", numBytes,
+		"fetched_chunks_count", numChunks,
+		"fetched_samples_count", numSamples,
+		"fetched_chunks_bytes", numChunkBytes,
 		"fetched_data_bytes", numDataBytes,
 		"status_code", statusCode,
 	}, stats.LoadExtraFields()...)

pkg/ingester/client/custom.go

@@ -1,5 +1,10 @@
 package client
 
+import (
+	"encoding/binary"
+	"github.com/cortexproject/cortex/pkg/chunk/encoding"
+)
+
 // ChunksCount returns the number of chunks in response.
 func (m *QueryStreamResponse) ChunksCount() int {
 	if len(m.Chunkseries) == 0 {
@@ -27,3 +32,17 @@ func (m *QueryStreamResponse) ChunksSize() int {
 	}
 	return size
 }
+
+func (m *QueryStreamResponse) SamplesCount() (count int) {
+	for _, ts := range m.Timeseries {
+		count += len(ts.Samples)
+	}
+	for _, cs := range m.Chunkseries {
+		for _, c := range cs.Chunks {
+			if c.Encoding == int32(encoding.PrometheusXorChunk) {
+				count += int(binary.BigEndian.Uint16(c.Data))
+			}
+		}
+	}
+	return
+}

pkg/querier/blocks_store_queryable.go

@@ -2,6 +2,7 @@ package querier
 
 import (
 	"context"
+	"encoding/binary"
 	"fmt"
 	"io"
 	"sort"
@@ -692,16 +693,21 @@ func (q *blocksStoreQuerier) fetchSeriesFromStores(
 			}
 
 			numSeries := len(mySeries)
+			numSamples, chunksCount := countSamplesAndChunks(mySeries...)
 			chunkBytes := countChunkBytes(mySeries...)
 			dataBytes := countDataBytes(mySeries...)
 
 			reqStats.AddFetchedSeries(uint64(numSeries))
+			reqStats.AddFetchedChunks(chunksCount)
+			reqStats.AddFetchedSamples(numSamples)
 			reqStats.AddFetchedChunkBytes(uint64(chunkBytes))
 			reqStats.AddFetchedDataBytes(uint64(dataBytes))
 
 			level.Debug(spanLog).Log("msg", "received series from store-gateway",
 				"instance", c.RemoteAddress(),
 				"fetched series", numSeries,
+				"fetched chunks", chunksCount,
+				"fetched samples", numSamples,
 				"fetched chunk bytes", chunkBytes,
 				"fetched data bytes", dataBytes,
 				"requested blocks", strings.Join(convertULIDsToString(blockIDs), " "),
@@ -1018,6 +1024,27 @@ func countDataBytes(series ...*storepb.Series) (count int) {
 	return count
 }
 
+// countChunks
+func countChunks(series ...*storepb.Series) (count int) {
+	for _, s := range series {
+		count += len(s.Chunks)
+	}
+	return
+}
+
+// countSamples
+func countSamplesAndChunks(series ...*storepb.Series) (samplesCount, chunksCount uint64) {
+	for _, s := range series {
+		chunksCount += uint64(len(s.Chunks))
+		for _, c := range s.Chunks {
+			if c.Raw != nil && c.Raw.Type == storepb.Chunk_XOR {
+				samplesCount += uint64(binary.BigEndian.Uint16(c.Raw.Data))
+			}
+		}
+	}
+	return
+}
+
 // only retry connection issues
 func isRetryableError(err error) bool {
 	switch status.Code(err) {

pkg/querier/stats/stats.go

@@ -148,7 +148,39 @@ func (s *QueryStats) LoadFetchedDataBytes() uint64 {
 	return atomic.LoadUint64(&s.FetchedDataBytes)
 }
 
-// Merge the provide Stats into this one.
+func (s *QueryStats) AddFetchedSamples(count uint64) {
+	if s == nil {
+		return
+	}
+
+	atomic.AddUint64(&s.FetchedSamplesCount, count)
+}
+
+func (s *QueryStats) LoadFetchedSamples() uint64 {
+	if s == nil {
+		return 0
+	}
+
+	return atomic.LoadUint64(&s.FetchedSamplesCount)
+}
+
+func (s *QueryStats) AddFetchedChunks(count uint64) {
+	if s == nil {
+		return
+	}
+
+	atomic.AddUint64(&s.FetchedChunksCount, count)
+}
+
+func (s *QueryStats) LoadFetchedChunks() uint64 {
+	if s == nil {
+		return 0
+	}
+
+	return atomic.LoadUint64(&s.FetchedChunksCount)
+}
+
+// Merge the provided Stats into this one.
 func (s *QueryStats) Merge(other *QueryStats) {
 	if s == nil || other == nil {
 		return
@@ -158,6 +190,8 @@ func (s *QueryStats) Merge(other *QueryStats) {
 	s.AddFetchedSeries(other.LoadFetchedSeries())
 	s.AddFetchedChunkBytes(other.LoadFetchedChunkBytes())
 	s.AddFetchedDataBytes(other.LoadFetchedDataBytes())
+	s.AddFetchedSamples(other.LoadFetchedSamples())
+	s.AddFetchedChunks(other.LoadFetchedChunks())
 	s.AddExtraFields(other.LoadExtraFields()...)
 }
 

pkg/querier/stats/stats.pb.go

@@ -22,4 +22,8 @@ message Stats {
   uint64 fetched_data_bytes = 4;
   // Extra fields to be reported on the stats log
   map<string, string> extra_fields = 5;
+  // The number of chunks fetched for the query
+  uint64 fetched_chunks_count = 6;
+  // The number of samples fetched for the query
+  uint64 fetched_samples_count = 7;
 }