Query: Add +Inf bucket to query duration metrics

CHANGELOG.md

@@ -47,6 +47,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#6163](https://github.com/thanos-io/thanos/pull/6163) Receiver: changed max backoff from 30s to 5s for forwarding requests. Can be configured with `--receive-forward-max-backoff`.
 - [#6327](https://github.com/thanos-io/thanos/pull/6327) *: *breaking :warning:* Use histograms instead of summaries for instrumented handlers.
 - [#6322](https://github.com/thanos-io/thanos/pull/6322) Logging: Avoid expensive log.Valuer evaluation for disallowed levels.
+- [#6358](https://github.com/thanos-io/thanos/pull/6358) Query: Add +Inf bucket to query duration metrics
 
 ### Removed
 

cmd/thanos/query.go

@@ -215,8 +215,8 @@ func registerQuery(app *extkingpin.App) {
 	grpcProxyStrategy := cmd.Flag("grpc.proxy-strategy", "Strategy to use when proxying Series requests to leaf nodes. Hidden and only used for testing, will be removed after lazy becomes the default.").Default(string(store.EagerRetrieval)).Hidden().Enum(string(store.EagerRetrieval), string(store.LazyRetrieval))
 
 	queryTelemetryDurationQuantiles := cmd.Flag("query.telemetry.request-duration-seconds-quantiles", "The quantiles for exporting metrics about the request duration quantiles.").Default("0.1", "0.25", "0.75", "1.25", "1.75", "2.5", "3", "5", "10").Float64List()
-	queryTelemetrySamplesQuantiles := cmd.Flag("query.telemetry.request-samples-quantiles", "The quantiles for exporting metrics about the samples count quantiles.").Default("100", "1000", "10000", "100000", "1000000").Int64List()
-	queryTelemetrySeriesQuantiles := cmd.Flag("query.telemetry.request-series-seconds-quantiles", "The quantiles for exporting metrics about the series count quantiles.").Default("10", "100", "1000", "10000", "100000").Int64List()
+	queryTelemetrySamplesQuantiles := cmd.Flag("query.telemetry.request-samples-quantiles", "The quantiles for exporting metrics about the samples count quantiles.").Default("100", "1000", "10000", "100000", "1000000").Float64List()
+	queryTelemetrySeriesQuantiles := cmd.Flag("query.telemetry.request-series-seconds-quantiles", "The quantiles for exporting metrics about the series count quantiles.").Default("10", "100", "1000", "10000", "100000").Float64List()
 
 	var storeRateLimits store.SeriesSelectLimits
 	storeRateLimits.RegisterFlags(cmd)
@@ -408,8 +408,8 @@ func runQuery(
 	grpcProxyStrategy string,
 	comp component.Component,
 	queryTelemetryDurationQuantiles []float64,
-	queryTelemetrySamplesQuantiles []int64,
-	queryTelemetrySeriesQuantiles []int64,
+	queryTelemetrySamplesQuantiles []float64,
+	queryTelemetrySeriesQuantiles []float64,
 	defaultEngine string,
 	storeRateLimits store.SeriesSelectLimits,
 	queryMode queryMode,

pkg/store/telemetry.go

@@ -4,6 +4,7 @@
 package store
 
 import (
+	"sort"
 	"strconv"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -17,17 +18,17 @@ import (
 type seriesStatsAggregator struct {
 	queryDuration *prometheus.HistogramVec
 
-	seriesLeBuckets  []int64
-	samplesLeBuckets []int64
+	seriesLeBuckets  []float64
+	samplesLeBuckets []float64
 	seriesStats      storepb.SeriesStatsCounter
 }
 
 // NewSeriesStatsAggregator is a constructor for seriesStatsAggregator.
 func NewSeriesStatsAggregator(
 	reg prometheus.Registerer,
 	durationQuantiles []float64,
-	sampleQuantiles []int64,
-	seriesQuantiles []int64,
+	sampleQuantiles []float64,
+	seriesQuantiles []float64,
 ) *seriesStatsAggregator {
 	return &seriesStatsAggregator{
 		queryDuration: promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
@@ -54,11 +55,11 @@ func (s *seriesStatsAggregator) Observe(duration float64) {
 		return
 	}
 	// Bucket matching for series/labels matchSeriesBucket/matchSamplesBucket => float64, float64
-	seriesLeBucket := s.findBucket(float64(s.seriesStats.Series), s.seriesLeBuckets)
-	samplesLeBucket := s.findBucket(float64(s.seriesStats.Samples), s.samplesLeBuckets)
+	seriesLeBucket := findBucket(float64(s.seriesStats.Series), s.seriesLeBuckets)
+	samplesLeBucket := findBucket(float64(s.seriesStats.Samples), s.samplesLeBuckets)
 	s.queryDuration.With(prometheus.Labels{
-		"series_le":  strconv.Itoa(int(seriesLeBucket)),
-		"samples_le": strconv.Itoa(int(samplesLeBucket)),
+		"series_le":  seriesLeBucket,
+		"samples_le": samplesLeBucket,
 	}).Observe(duration)
 	s.reset()
 }
@@ -67,18 +68,18 @@ func (s *seriesStatsAggregator) reset() {
 	s.seriesStats = storepb.SeriesStatsCounter{}
 }
 
-func (s *seriesStatsAggregator) findBucket(value float64, quantiles []int64) int64 {
+func findBucket(value float64, quantiles []float64) string {
 	if len(quantiles) == 0 {
-		return 0
+		return "+Inf"
 	}
-	var foundBucket int64
-	for _, bucket := range quantiles {
-		foundBucket = bucket
-		if value < float64(bucket) {
-			break
-		}
+
+	// If the value is bigger than the largest bucket we return +Inf
+	if value >= float64(quantiles[len(quantiles)-1]) {
+		return "+Inf"
 	}
-	return foundBucket
+
+	// SearchFloats64s gets the appropriate index in the quantiles array based on the value
+	return strconv.FormatFloat(quantiles[sort.SearchFloat64s(quantiles, value)], 'f', -1, 64)
 }
 
 // NoopSeriesStatsAggregator is a query performance series aggregator that does nothing.

pkg/testutil/e2eutil/prometheus.go

@@ -18,6 +18,7 @@ import (
 	"runtime"
 	"sort"
 	"strings"
+	"sync"
 	"syscall"
 	"testing"
 	"time"
@@ -485,6 +486,7 @@ func createBlock(
 	var timeStepSize = (maxt - mint) / int64(numSamples+1)
 	var batchSize = len(series) / runtime.GOMAXPROCS(0)
 	r := rand.New(rand.NewSource(int64(numSamples)))
+	var randMutex sync.Mutex
 
 	for len(series) > 0 {
 		l := batchSize
@@ -507,7 +509,9 @@ func createBlock(
 
 					var err error
 					if sampleType == chunkenc.ValFloat {
+						randMutex.Lock()
 						_, err = app.Append(0, lset, t, r.Float64())
+						randMutex.Unlock()
 					} else if sampleType == chunkenc.ValHistogram {
 						_, err = app.AppendHistogram(0, lset, t, &histogramSample, nil)
 					}

test/e2e/e2ethanos/services.go

@@ -260,6 +260,10 @@ type QuerierBuilder struct {
 	replicaLabels []string
 	tracingConfig string
 
+	telemetryDurationQuantiles []float64
+	telemetrySamplesQuantiles  []float64
+	telemetrySeriesQuantiles   []float64
+
 	e2e.Linkable
 	f e2e.FutureRunnable
 }
@@ -369,6 +373,13 @@ func (q *QuerierBuilder) WithQueryMode(mode string) *QuerierBuilder {
 	return q
 }
 
+func (q *QuerierBuilder) WithTelemetryQuantiles(duration []float64, samples []float64, series []float64) *QuerierBuilder {
+	q.telemetryDurationQuantiles = duration
+	q.telemetrySamplesQuantiles = samples
+	q.telemetrySeriesQuantiles = series
+	return q
+}
+
 func (q *QuerierBuilder) Init() *e2emon.InstrumentedRunnable {
 	args, err := q.collectArgs()
 	if err != nil {
@@ -459,6 +470,15 @@ func (q *QuerierBuilder) collectArgs() ([]string, error) {
 	if q.tracingConfig != "" {
 		args = append(args, "--tracing.config="+q.tracingConfig)
 	}
+	for _, bucket := range q.telemetryDurationQuantiles {
+		args = append(args, "--query.telemetry.request-duration-seconds-quantiles="+strconv.FormatFloat(bucket, 'f', -1, 64))
+	}
+	for _, bucket := range q.telemetrySamplesQuantiles {
+		args = append(args, "--query.telemetry.request-samples-quantiles="+strconv.FormatFloat(bucket, 'f', -1, 64))
+	}
+	for _, bucket := range q.telemetrySeriesQuantiles {
+		args = append(args, "--query.telemetry.request-series-seconds-quantiles="+strconv.FormatFloat(bucket, 'f', -1, 64))
+	}
 	return args, nil
 }
 

test/e2e/query_test.go

@@ -598,7 +598,7 @@ func TestQueryStoreMetrics(t *testing.T) {
 	bkt, err := s3.NewBucketWithConfig(l, e2ethanos.NewS3Config(bucket, minio.Endpoint("http"), minio.Dir()), "test")
 	testutil.Ok(t, err)
 
-	// Preparing 2 different blocks for the tests.
+	// Preparing 3 different blocks for the tests.
 	{
 		blockSizes := []struct {
 			samples int
@@ -607,6 +607,7 @@ func TestQueryStoreMetrics(t *testing.T) {
 		}{
 			{samples: 10, series: 1, name: "one_series"},
 			{samples: 10, series: 1001, name: "thousand_one_series"},
+			{samples: 10, series: 10001, name: "inf_series"},
 		}
 		now := time.Now()
 		externalLabels := labels.FromStrings("prometheus", "p1", "replica", "0")
@@ -644,9 +645,12 @@ func TestQueryStoreMetrics(t *testing.T) {
 		"",
 		nil,
 	)
-	querier := e2ethanos.NewQuerierBuilder(e, "1", storeGW.InternalEndpoint("grpc")).Init()
+
+	sampleBuckets := []float64{100, 1000, 10000, 100000}
+	seriesBuckets := []float64{10, 100, 1000, 10000}
+	querier := e2ethanos.NewQuerierBuilder(e, "1", storeGW.InternalEndpoint("grpc")).WithTelemetryQuantiles(nil, sampleBuckets, seriesBuckets).Init()
 	testutil.Ok(t, e2e.StartAndWaitReady(storeGW, querier))
-	testutil.Ok(t, storeGW.WaitSumMetrics(e2emon.Equals(2), "thanos_blocks_meta_synced"))
+	testutil.Ok(t, storeGW.WaitSumMetrics(e2emon.Equals(3), "thanos_blocks_meta_synced"))
 
 	// Querying the series in the previously created blocks to ensure we produce Store API query metrics.
 	{
@@ -663,6 +667,13 @@ func TestQueryStoreMetrics(t *testing.T) {
 			Deduplicate: true,
 		}, 1001)
 		testutil.Ok(t, err)
+
+		instantQuery(t, ctx, querier.Endpoint("http"), func() string {
+			return "max_over_time(inf_series[2h])"
+		}, time.Now, promclient.QueryOptions{
+			Deduplicate: true,
+		}, 10001)
+		testutil.Ok(t, err)
 	}
 
 	mon, err := e2emon.Start(e)
@@ -701,6 +712,24 @@ func TestQueryStoreMetrics(t *testing.T) {
 			Value: model.SampleValue(1),
 		},
 	})
+
+	queryWaitAndAssert(t, ctx, mon.GetMonitoringRunnable().Endpoint(e2edb.AccessPortName), func() string {
+		return "thanos_store_api_query_duration_seconds_count{samples_le='+Inf',series_le='+Inf'}"
+	}, time.Now, promclient.QueryOptions{
+		Deduplicate: true,
+	}, model.Vector{
+		&model.Sample{
+			Metric: model.Metric{
+				"__name__":   "thanos_store_api_query_duration_seconds_count",
+				"instance":   "storemetrics01-querier-1:8080",
+				"job":        "querier-1",
+				"samples_le": "+Inf",
+				"series_le":  "+Inf",
+			},
+			Value: model.SampleValue(1),
+		},
+	})
+
 }
 
 // Regression test for https://github.com/thanos-io/thanos/issues/5033.