Add new query stats metrics to track prometheus querystats

Signed-off-by: SungJin1212 <[email protected]>

Author: SungJin1212

CHANGELOG.md

@@ -8,6 +8,7 @@
 * [FEATURE] Ruler: Minimize chances of missed rule group evaluations that can occur due to OOM kills, bad underlying nodes, or due to an unhealthy ruler that appears in the ring as healthy. This feature is enabled via `-ruler.enable-ha-evaluation` flag. #6129
 * [FEATURE] Store Gateway: Add an in-memory chunk cache. #6245
 * [FEATURE] Chunk Cache: Support multi level cache and add metrics. #6249
+* [ENHANCEMENT] Query Frontend: Add new query stats metrics `cortex_query_total_queryable_samples_total` and `cortex_query_peak_samples` to track totalQueryableSamples and peakSample per user. #6228
 * [ENHANCEMENT] Ingester: Add `blocks-storage.tsdb.wal-compression-type` to support zstd wal compression type. #6232
 * [ENHANCEMENT] Query Frontend: Add info field to query response. #6207
 * [ENHANCEMENT] Query Frontend: Add peakSample in query stats response. #6188

pkg/api/handlers.go

@@ -226,7 +226,7 @@ func NewQuerierHandler(
 		// This is used for the stats API which we should not support. Or find other ways to.
 		prometheus.GathererFunc(func() ([]*dto.MetricFamily, error) { return nil, nil }),
 		reg,
-		nil,
+		querier.StatsRenderer,
 		false,
 		nil,
 		false,

pkg/frontend/transport/handler.go

@@ -89,13 +89,15 @@ type Handler struct {
 	roundTripper http.RoundTripper
 
 	// Metrics.
-	querySeconds    *prometheus.CounterVec
-	querySeries     *prometheus.CounterVec
-	querySamples    *prometheus.CounterVec
-	queryChunkBytes *prometheus.CounterVec
-	queryDataBytes  *prometheus.CounterVec
-	rejectedQueries *prometheus.CounterVec
-	activeUsers     *util.ActiveUsersCleanupService
+	querySeconds               *prometheus.CounterVec
+	querySeries                *prometheus.CounterVec
+	queryFetchedSamples        *prometheus.CounterVec
+	queryTotalQueryableSamples *prometheus.CounterVec
+	queryPeakSamples           *prometheus.HistogramVec
+	queryChunkBytes            *prometheus.CounterVec
+	queryDataBytes             *prometheus.CounterVec
+	rejectedQueries            *prometheus.CounterVec
+	activeUsers                *util.ActiveUsersCleanupService
 }
 
 // NewHandler creates a new frontend handler.
@@ -117,11 +119,25 @@ func NewHandler(cfg HandlerConfig, roundTripper http.RoundTripper, log log.Logge
 			Help: "Number of series fetched to execute a query.",
 		}, []string{"user"})
 
-		h.querySamples = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+		h.queryFetchedSamples = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
 			Name: "cortex_query_samples_total",
 			Help: "Number of samples fetched to execute a query.",
 		}, []string{"user"})
 
+		// It tracks TotalSamples in https://github.com/prometheus/prometheus/blob/main/util/stats/query_stats.go#L237 for each user.
+		h.queryTotalQueryableSamples = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Name: "cortex_query_samples_processed_total",
+			Help: "Number of samples processed to execute a query.",
+		}, []string{"user"})
+
+		h.queryPeakSamples = promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
+			Name:                            "cortex_query_peak_samples",
+			Help:                            "Highest count of samples considered to execute a query.",
+			NativeHistogramBucketFactor:     1.1,
+			NativeHistogramMaxBucketNumber:  100,
+			NativeHistogramMinResetDuration: 1 * time.Hour,
+		}, []string{"user"})
+
 		h.queryChunkBytes = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
 			Name: "cortex_query_fetched_chunks_bytes_total",
 			Help: "Size of all chunks fetched to execute a query in bytes.",
@@ -143,7 +159,9 @@ func NewHandler(cfg HandlerConfig, roundTripper http.RoundTripper, log log.Logge
 		h.activeUsers = util.NewActiveUsersCleanupWithDefaultValues(func(user string) {
 			h.querySeconds.DeleteLabelValues(user)
 			h.querySeries.DeleteLabelValues(user)
-			h.querySamples.DeleteLabelValues(user)
+			h.queryFetchedSamples.DeleteLabelValues(user)
+			h.queryTotalQueryableSamples.DeleteLabelValues(user)
+			h.queryPeakSamples.DeleteLabelValues(user)
 			h.queryChunkBytes.DeleteLabelValues(user)
 			h.queryDataBytes.DeleteLabelValues(user)
 			if err := util.DeleteMatchingLabels(h.rejectedQueries, map[string]string{"user": user}); err != nil {
@@ -301,6 +319,8 @@ func (f *Handler) reportQueryStats(r *http.Request, userID string, queryString u
 	numSeries := stats.LoadFetchedSeries()
 	numChunks := stats.LoadFetchedChunks()
 	numSamples := stats.LoadFetchedSamples()
+	numProcessedSamples := stats.LoadProcessedSamples()
+	numPeakSamples := stats.LoadPeakSamples()
 	numChunkBytes := stats.LoadFetchedChunkBytes()
 	numDataBytes := stats.LoadFetchedDataBytes()
 	numStoreGatewayTouchedPostings := stats.LoadStoreGatewayTouchedPostings()
@@ -312,7 +332,9 @@ func (f *Handler) reportQueryStats(r *http.Request, userID string, queryString u
 	// Track stats.
 	f.querySeconds.WithLabelValues(userID).Add(wallTime.Seconds())
 	f.querySeries.WithLabelValues(userID).Add(float64(numSeries))
-	f.querySamples.WithLabelValues(userID).Add(float64(numSamples))
+	f.queryFetchedSamples.WithLabelValues(userID).Add(float64(numSamples))
+	f.queryTotalQueryableSamples.WithLabelValues(userID).Add(float64(numProcessedSamples))
+	f.queryPeakSamples.WithLabelValues(userID).Observe(float64(numPeakSamples))
 	f.queryChunkBytes.WithLabelValues(userID).Add(float64(numChunkBytes))
 	f.queryDataBytes.WithLabelValues(userID).Add(float64(numDataBytes))
 	f.activeUsers.UpdateUserTimestamp(userID, time.Now())

pkg/frontend/transport/handler_test.go

@@ -188,7 +188,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:               "test handler with stats enabled",
 			cfg:                HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics:    4,
+			expectedMetrics:    6,
 			roundTripperFunc:   roundTripper,
 			expectedStatusCode: http.StatusOK,
 		},
@@ -202,7 +202,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonResponseTooLarge",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusRequestEntityTooLarge,
@@ -218,7 +218,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonTooManyRequests",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusTooManyRequests,
@@ -234,7 +234,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonTooManySamples",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusUnprocessableEntity,
@@ -250,7 +250,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonTooLongRange",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusUnprocessableEntity,
@@ -266,7 +266,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonSeriesFetched",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusUnprocessableEntity,
@@ -282,7 +282,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonChunksFetched",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusUnprocessableEntity,
@@ -298,7 +298,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonChunkBytesFetched",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusUnprocessableEntity,
@@ -314,7 +314,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonDataBytesFetched",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusUnprocessableEntity,
@@ -330,7 +330,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonSeriesLimitStoreGateway",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusUnprocessableEntity,
@@ -346,7 +346,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonChunksLimitStoreGateway",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusUnprocessableEntity,
@@ -362,7 +362,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 		{
 			name:            "test handler with reasonBytesLimitStoreGateway",
 			cfg:             HandlerConfig{QueryStatsEnabled: true},
-			expectedMetrics: 4,
+			expectedMetrics: 6,
 			roundTripperFunc: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
 				return &http.Response{
 					StatusCode: http.StatusUnprocessableEntity,
@@ -395,6 +395,8 @@ func TestHandler_ServeHTTP(t *testing.T) {
 				"cortex_query_fetched_series_total",
 				"cortex_query_samples_total",
 				"cortex_query_fetched_chunks_bytes_total",
+				"cortex_query_samples_processed_total",
+				"cortex_query_peak_samples",
 			)
 
 			assert.NoError(t, err)

pkg/querier/stats/stats.go

@@ -302,6 +302,46 @@ func (s *QueryStats) LoadStoreGatewayTouchedPostingBytes() uint64 {
 	return atomic.LoadUint64(&s.StoreGatewayTouchedPostingBytes)
 }
 
+func (s *QueryStats) AddProcessedSamples(count uint64) {
+	if s == nil {
+		return
+	}
+
+	atomic.AddUint64(&s.ProcessedSamples, count)
+}
+
+func (s *QueryStats) LoadProcessedSamples() uint64 {
+	if s == nil {
+		return 0
+	}
+
+	return atomic.LoadUint64(&s.ProcessedSamples)
+}
+
+func (s *QueryStats) AddPeakSamples(count uint64) {
+	if s == nil {
+		return
+	}
+
+	atomic.AddUint64(&s.PeakSamples, count)
+}
+
+func (s *QueryStats) SetPeakSamples(count uint64) {
+	if s == nil {
+		return
+	}
+
+	atomic.StoreUint64(&s.PeakSamples, count)
+}
+
+func (s *QueryStats) LoadPeakSamples() uint64 {
+	if s == nil {
+		return 0
+	}
+
+	return atomic.LoadUint64(&s.PeakSamples)
+}
+
 // Merge the provided Stats into this one.
 func (s *QueryStats) Merge(other *QueryStats) {
 	if s == nil || other == nil {
@@ -317,6 +357,8 @@ func (s *QueryStats) Merge(other *QueryStats) {
 	s.AddFetchedChunks(other.LoadFetchedChunks())
 	s.AddStoreGatewayTouchedPostings(other.LoadStoreGatewayTouchedPostings())
 	s.AddStoreGatewayTouchedPostingBytes(other.LoadStoreGatewayTouchedPostingBytes())
+	s.AddProcessedSamples(other.LoadProcessedSamples())
+	s.SetPeakSamples(max(s.LoadPeakSamples(), other.LoadPeakSamples()))
 	s.AddExtraFields(other.LoadExtraFields()...)
 }