Query: Add tenant label to exported metrics

With this commit we now add the tenant label to relevant metrics
exported by the query component.

This includes the HTTP metrics handled by the InstrumentationMiddleware
and the query latency metrics.

Signed-off-by: Jacob Baungard Hansen <jacobbaungard@redhat.com>

CHANGELOG.md

@@ -22,6 +22,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#6530](https://github.com/thanos-io/thanos/pull/6530) / [#6690](https://github.com/thanos-io/thanos/pull/6690) Query: Add command line arguments for configuring tenants and forward tenant information to Store Gateway.
 - [#6765](https://github.com/thanos-io/thanos/pull/6765) Index Cache: Add `enabled_items` to index cache config to selectively cache configured items. Available item types are `Postings`, `Series` and `ExpandedPostings`.
 - [#6773](https://github.com/thanos-io/thanos/pull/6773) Index Cache: Add `ttl` to control the ttl to store items in remote index caches like memcached and redis.
+- [#6794](https://github.com/thanos-io/thanos/pull/6794) Query: *breaking :warning:* Add tenant label to relevant exported metrics. Note that this change may cause some pre-existing custom dashboard queries to be incorrect due to the added label.
 
 ### Changed
 

cmd/thanos/query.go

@@ -715,7 +715,7 @@ func runQuery(
 		// Configure Request Logging for HTTP calls.
 		logMiddleware := logging.NewHTTPServerMiddleware(logger, httpLogOpts...)
 
-		ins := extpromhttp.NewInstrumentationMiddleware(reg, nil)
+		ins := extpromhttp.NewTenantInstrumentationMiddleware(tenantHeader, defaultTenant, reg, nil)
 		// TODO(bplotka in PR #513 review): pass all flags, not only the flags needed by prefix rewriting.
 		ui.NewQueryUI(logger, endpoints, webExternalPrefix, webPrefixHeaderName, alertQueryURL).Register(router, ins)
 

pkg/api/query/v1.go

@@ -704,7 +704,7 @@ func (qapi *QueryAPI) query(r *http.Request) (interface{}, []error, *api.ApiErro
 		return nil, nil, &api.ApiError{Typ: api.ErrorExec, Err: res.Err}, qry.Close
 	}
 
-	aggregator := qapi.seriesStatsAggregatorFactory.NewAggregator()
+	aggregator := qapi.seriesStatsAggregatorFactory.NewAggregator(tenant)
 	for i := range seriesStats {
 		aggregator.Aggregate(seriesStats[i])
 	}
@@ -1007,7 +1007,7 @@ func (qapi *QueryAPI) queryRange(r *http.Request) (interface{}, []error, *api.Ap
 		}
 		return nil, nil, &api.ApiError{Typ: api.ErrorExec, Err: res.Err}, qry.Close
 	}
-	aggregator := qapi.seriesStatsAggregatorFactory.NewAggregator()
+	aggregator := qapi.seriesStatsAggregatorFactory.NewAggregator(tenant)
 	for i := range seriesStats {
 		aggregator.Aggregate(seriesStats[i])
 	}

pkg/store/telemetry.go

@@ -19,6 +19,7 @@ type seriesStatsAggregator struct {
 	queryDuration    *prometheus.HistogramVec
 	seriesLeBuckets  []float64
 	samplesLeBuckets []float64
+	tenant           string
 
 	seriesStats storepb.SeriesStatsCounter
 }
@@ -29,11 +30,12 @@ type seriesStatsAggregatorFactory struct {
 	samplesLeBuckets []float64
 }
 
-func (f *seriesStatsAggregatorFactory) NewAggregator() SeriesQueryPerformanceMetricsAggregator {
+func (f *seriesStatsAggregatorFactory) NewAggregator(tenant string) SeriesQueryPerformanceMetricsAggregator {
 	return &seriesStatsAggregator{
 		queryDuration:    f.queryDuration,
 		seriesLeBuckets:  f.seriesLeBuckets,
 		samplesLeBuckets: f.samplesLeBuckets,
+		tenant:           tenant,
 		seriesStats:      storepb.SeriesStatsCounter{},
 	}
 }
@@ -49,7 +51,7 @@ func NewSeriesStatsAggregatorFactory(
 			Name:    "thanos_store_api_query_duration_seconds",
 			Help:    "Duration of the Thanos Store API select phase for a query.",
 			Buckets: durationQuantiles,
-		}, []string{"series_le", "samples_le"}),
+		}, []string{"series_le", "samples_le", "tenant"}),
 		seriesLeBuckets:  seriesQuantiles,
 		samplesLeBuckets: sampleQuantiles,
 	}
@@ -67,7 +69,7 @@ func NewSeriesStatsAggregator(
 			Name:    "thanos_store_api_query_duration_seconds",
 			Help:    "Duration of the Thanos Store API select phase for a query.",
 			Buckets: durationQuantiles,
-		}, []string{"series_le", "samples_le"}),
+		}, []string{"series_le", "samples_le", "tenant"}),
 		seriesLeBuckets:  seriesQuantiles,
 		samplesLeBuckets: sampleQuantiles,
 		seriesStats:      storepb.SeriesStatsCounter{},
@@ -92,6 +94,7 @@ func (s *seriesStatsAggregator) Observe(duration float64) {
 	s.queryDuration.With(prometheus.Labels{
 		"series_le":  seriesLeBucket,
 		"samples_le": samplesLeBucket,
+		"tenant":     s.tenant,
 	}).Observe(duration)
 	s.reset()
 }
@@ -115,7 +118,7 @@ func findBucket(value float64, quantiles []float64) string {
 }
 
 type SeriesQueryPerformanceMetricsAggregatorFactory interface {
-	NewAggregator() SeriesQueryPerformanceMetricsAggregator
+	NewAggregator(tenant string) SeriesQueryPerformanceMetricsAggregator
 }
 
 type SeriesQueryPerformanceMetricsAggregator interface {
@@ -133,7 +136,7 @@ func (s *NoopSeriesStatsAggregator) Observe(_ float64) {}
 // NoopSeriesStatsAggregatorFactory is a query performance series aggregator factory that does nothing.
 type NoopSeriesStatsAggregatorFactory struct{}
 
-func (s *NoopSeriesStatsAggregatorFactory) NewAggregator() SeriesQueryPerformanceMetricsAggregator {
+func (s *NoopSeriesStatsAggregatorFactory) NewAggregator(tenant string) SeriesQueryPerformanceMetricsAggregator {
 	return &NoopSeriesStatsAggregator{}
 }
 

test/e2e/query_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/efficientgo/e2e"
 	e2edb "github.com/efficientgo/e2e/db"
 	e2emon "github.com/efficientgo/e2e/monitoring"
+	"github.com/efficientgo/e2e/monitoring/matchers"
 	"github.com/go-kit/log"
 	"github.com/gogo/protobuf/proto"
 	"github.com/golang/snappy"
@@ -799,13 +800,22 @@ func TestQueryStoreMetrics(t *testing.T) {
 			Deduplicate: true,
 		}, 10001)
 		testutil.Ok(t, err)
+
+		// query with a non-default tenant
+		instantQuery(t, ctx, querier.Endpoint("http"), func() string {
+			return "max_over_time(one_series{instance='foo_0'}[2h])"
+		}, time.Now, promclient.QueryOptions{
+			Deduplicate: true,
+			HTTPHeaders: map[string][]string{"thanos-tenant": {"test-tenant-1"}},
+		}, 1)
+		testutil.Ok(t, err)
 	}
 
 	mon, err := e2emon.Start(e)
 	testutil.Ok(t, err)
 
 	queryWaitAndAssert(t, ctx, mon.GetMonitoringRunnable().Endpoint(e2edb.AccessPortName), func() string {
-		return "thanos_store_api_query_duration_seconds_count{samples_le='100000',series_le='10000'}"
+		return "thanos_store_api_query_duration_seconds_count{samples_le='100000',series_le='10000',tenant='default-tenant'}"
 	}, time.Now, promclient.QueryOptions{
 		Deduplicate: true,
 	}, model.Vector{
@@ -816,13 +826,14 @@ func TestQueryStoreMetrics(t *testing.T) {
 				"job":        "querier-1",
 				"samples_le": "100000",
 				"series_le":  "10000",
+				"tenant":     "default-tenant",
 			},
 			Value: model.SampleValue(1),
 		},
 	})
 
 	queryWaitAndAssert(t, ctx, mon.GetMonitoringRunnable().Endpoint(e2edb.AccessPortName), func() string {
-		return "thanos_store_api_query_duration_seconds_count{samples_le='100',series_le='10'}"
+		return "thanos_store_api_query_duration_seconds_count{samples_le='100',series_le='10',tenant='default-tenant'}"
 	}, time.Now, promclient.QueryOptions{
 		Deduplicate: true,
 	}, model.Vector{
@@ -833,13 +844,14 @@ func TestQueryStoreMetrics(t *testing.T) {
 				"job":        "querier-1",
 				"samples_le": "100",
 				"series_le":  "10",
+				"tenant":     "default-tenant",
 			},
 			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'}"
+		return "thanos_store_api_query_duration_seconds_count{samples_le='+Inf',series_le='+Inf',tenant='default-tenant'}"
 	}, time.Now, promclient.QueryOptions{
 		Deduplicate: true,
 	}, model.Vector{
@@ -850,6 +862,25 @@ func TestQueryStoreMetrics(t *testing.T) {
 				"job":        "querier-1",
 				"samples_le": "+Inf",
 				"series_le":  "+Inf",
+				"tenant":     "default-tenant",
+			},
+			Value: model.SampleValue(1),
+		},
+	})
+
+	queryWaitAndAssert(t, ctx, mon.GetMonitoringRunnable().Endpoint(e2edb.AccessPortName), func() string {
+		return "thanos_store_api_query_duration_seconds_count{samples_le='100',series_le='10',tenant='test-tenant-1'}"
+	}, 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": "100",
+				"series_le":  "10",
+				"tenant":     "test-tenant-1",
 			},
 			Value: model.SampleValue(1),
 		},
@@ -2304,3 +2335,87 @@ func TestSidecarPrefersExtLabels(t *testing.T) {
 		Timestamp: model.TimeFromUnixNano(now.Add(time.Hour).UnixNano()),
 	}}, retv)
 }
+
+func TestTenantHTTPMetrics(t *testing.T) {
+	t.Parallel()
+
+	e, err := e2e.NewDockerEnvironment("tenant-metrics")
+	testutil.Ok(t, err)
+	t.Cleanup(e2ethanos.CleanScenario(t, e))
+
+	// scrape the local prometheus, and our querier metrics
+	prom1, sidecar1 := e2ethanos.NewPrometheusWithSidecar(e, "alone", e2ethanos.DefaultPromConfig("prom-alone", 0, "", "", e2ethanos.LocalPrometheusTarget, "tenant-metrics-querier-1:8080"), "", e2ethanos.DefaultPrometheusImage(), "")
+
+	q := e2ethanos.NewQuerierBuilder(e, "1", sidecar1.InternalEndpoint("grpc")).Init()
+	testutil.Ok(t, e2e.StartAndWaitReady(q))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
+	t.Cleanup(cancel)
+
+	testutil.Ok(t, e2e.StartAndWaitReady(prom1, sidecar1))
+
+	// Query once with default-tenant to ensure everything is ready
+	// for the following requests
+	instantQuery(t, ctx, q.Endpoint("http"), func() string {
+		return "prometheus_api_remote_read_queries"
+	}, time.Now, promclient.QueryOptions{
+		Deduplicate: true,
+	}, 1)
+	testutil.Ok(t, err)
+
+	// Query a few times with tenant 1
+	instantQuery(t, ctx, q.Endpoint("http"), func() string {
+		return "prometheus_api_remote_read_queries"
+	}, time.Now, promclient.QueryOptions{
+		Deduplicate: true,
+		HTTPHeaders: map[string][]string{"thanos-tenant": {"test-tenant-1"}},
+	}, 1)
+	testutil.Ok(t, err)
+
+	instantQuery(t, ctx, q.Endpoint("http"), func() string {
+		return "go_goroutines"
+	}, time.Now, promclient.QueryOptions{
+		Deduplicate: true,
+		HTTPHeaders: map[string][]string{"thanos-tenant": {"test-tenant-1"}},
+	}, 2)
+	testutil.Ok(t, err)
+
+	instantQuery(t, ctx, q.Endpoint("http"), func() string {
+		return "go_memstats_frees_total"
+	}, time.Now, promclient.QueryOptions{
+		Deduplicate: true,
+		HTTPHeaders: map[string][]string{"thanos-tenant": {"test-tenant-1"}},
+	}, 2)
+	testutil.Ok(t, err)
+
+	// query just once with tenant-2
+	instantQuery(t, ctx, q.Endpoint("http"), func() string {
+		return "go_memstats_heap_alloc_bytes"
+	}, time.Now, promclient.QueryOptions{
+		Deduplicate: true,
+		HTTPHeaders: map[string][]string{"thanos-tenant": {"test-tenant-2"}},
+	}, 2)
+	testutil.Ok(t, err)
+
+	// tenant-1 made 3 requests
+	tenant1Matcher, err := matchers.NewMatcher(matchers.MatchEqual, "tenant", "test-tenant-1")
+	testutil.Ok(t, err)
+	testutil.Ok(t, q.WaitSumMetricsWithOptions(
+		e2emon.Equals(3),
+		[]string{"http_requests_total"}, e2emon.WithLabelMatchers(
+			tenant1Matcher,
+		),
+		e2emon.WaitMissingMetrics(),
+	))
+
+	// tenant 2 just made one request
+	tenant2Matcher, err := matchers.NewMatcher(matchers.MatchEqual, "tenant", "test-tenant-2")
+	testutil.Ok(t, err)
+	testutil.Ok(t, q.WaitSumMetricsWithOptions(
+		e2emon.Equals(1),
+		[]string{"http_requests_total"}, e2emon.WithLabelMatchers(
+			tenant2Matcher,
+		),
+		e2emon.WaitMissingMetrics(),
+	))
+}