Query: Trim very long external labels and add cmd flag to optionally specify metric labels to collect

cmd/thanos/query.go

@@ -109,6 +109,10 @@ func registerQuery(app *extkingpin.App) {
 	maxConcurrentSelects := cmd.Flag("query.max-concurrent-select", "Maximum number of select requests made concurrently per a query.").
 		Default("4").Int()
 
+	queryConnMetricLabels := cmd.Flag("query.conn-metric.label", "Optional selection of query connection metric labels to be collected from endpoint set").
+		Default(string(query.ExternalLabels), string(query.StoreType)).
+		Enums(string(query.ExternalLabels), string(query.StoreType))
+
 	queryReplicaLabels := cmd.Flag("query.replica-label", "Labels to treat as a replica indicator along which data is deduplicated. Still you will be able to query without deduplication using 'dedup=false' parameter. Data includes time series, recording rules, and alerting rules.").
 		Strings()
 
@@ -280,6 +284,7 @@ func registerQuery(app *extkingpin.App) {
 			*dynamicLookbackDelta,
 			time.Duration(*defaultEvaluationInterval),
 			time.Duration(*storeResponseTimeout),
+			*queryConnMetricLabels,
 			*queryReplicaLabels,
 			selectorLset,
 			getFlagsMap(cmd.Flags()),
@@ -355,6 +360,7 @@ func runQuery(
 	dynamicLookbackDelta bool,
 	defaultEvaluationInterval time.Duration,
 	storeResponseTimeout time.Duration,
+	queryConnMetricLabels []string,
 	queryReplicaLabels []string,
 	selectorLset labels.Labels,
 	flagsMap map[string]string,
@@ -497,6 +503,7 @@ func runQuery(
 			dialOpts,
 			unhealthyStoreTimeout,
 			endpointInfoTimeout,
+			queryConnMetricLabels...,
 		)
 		proxy            = store.NewProxyStore(logger, reg, endpoints.GetStoreClients, component.Query, selectorLset, storeResponseTimeout, store.RetrievalStrategy(grpcProxyStrategy))
 		rulesProxy       = rules.NewProxy(logger, endpoints.GetRulesClients)

docs/components/query.md

@@ -335,6 +335,9 @@ Flags:
       --query.auto-downsampling  Enable automatic adjustment (step / 5) to what
                                  source of data should be used in store gateways
                                  if no max_source_resolution param is specified.
+      --query.conn-metric.label=external_labels... ...
+                                 Optional selection of query connection metric
+                                 labels to be collected from endpoint set
       --query.default-evaluation-interval=1m
                                  Set default evaluation interval for sub
                                  queries.

pkg/query/endpointset.go

@@ -38,11 +38,20 @@ const (
 	noMetadataEndpointMessage = "cannot obtain metadata: neither info nor store client found"
 )
 
+type queryConnMetricLabel string
+
+const (
+	ExternalLabels queryConnMetricLabel = "external_labels"
+	StoreType      queryConnMetricLabel = "store_type"
+)
+
 type GRPCEndpointSpec struct {
 	addr           string
 	isStrictStatic bool
 }
 
+const externalLabelLimit = 1000
+
 // NewGRPCEndpointSpec creates gRPC endpoint spec.
 // It uses InfoAPI to get Metadata.
 func NewGRPCEndpointSpec(addr string, isStrictStatic bool) *GRPCEndpointSpec {
@@ -183,28 +192,41 @@ type endpointSetNodeCollector struct {
 	storePerExtLset map[string]int
 
 	connectionsDesc *prometheus.Desc
+	labels          []string
 }
 
-func newEndpointSetNodeCollector() *endpointSetNodeCollector {
+func newEndpointSetNodeCollector(labels ...string) *endpointSetNodeCollector {
+	if len(labels) == 0 {
+		labels = []string{string(ExternalLabels), string(StoreType)}
+	}
 	return &endpointSetNodeCollector{
 		storeNodes: map[component.Component]map[string]int{},
 		connectionsDesc: prometheus.NewDesc(
 			"thanos_store_nodes_grpc_connections",
 			"Number of gRPC connection to Store APIs. Opened connection means healthy store APIs available for Querier.",
-			[]string{"external_labels", "store_type"}, nil,
+			labels, nil,
 		),
+		labels: labels,
 	}
 }
 
+// truncateExtLabels truncates the stringify external labels with the format of {labels..}.
+func truncateExtLabels(s string, threshold int) string {
+	if len(s) > threshold {
+		return fmt.Sprintf("%s}", s[:threshold-1])
+	}
+	return s
+}
 func (c *endpointSetNodeCollector) Update(nodes map[component.Component]map[string]int) {
 	storeNodes := make(map[component.Component]map[string]int, len(nodes))
 	storePerExtLset := map[string]int{}
 
-	for k, v := range nodes {
-		storeNodes[k] = make(map[string]int, len(v))
-		for kk, vv := range v {
-			storePerExtLset[kk] += vv
-			storeNodes[k][kk] = vv
+	for storeType, occurrencesPerExtLset := range nodes {
+		storeNodes[storeType] = make(map[string]int, len(occurrencesPerExtLset))
+		for externalLabels, occurrences := range occurrencesPerExtLset {
+			externalLabels = truncateExtLabels(externalLabels, externalLabelLimit)
+			storePerExtLset[externalLabels] += occurrences
+			storeNodes[storeType][externalLabels] = occurrences
 		}
 	}
 
@@ -228,7 +250,17 @@ func (c *endpointSetNodeCollector) Collect(ch chan<- prometheus.Metric) {
 			if storeType != nil {
 				storeTypeStr = storeType.String()
 			}
-			ch <- prometheus.MustNewConstMetric(c.connectionsDesc, prometheus.GaugeValue, float64(occurrences), externalLabels, storeTypeStr)
+			// Select only required labels.
+			lbls := []string{}
+			for _, lbl := range c.labels {
+				switch lbl {
+				case string(ExternalLabels):
+					lbls = append(lbls, externalLabels)
+				case string(StoreType):
+					lbls = append(lbls, storeTypeStr)
+				}
+			}
+			ch <- prometheus.MustNewConstMetric(c.connectionsDesc, prometheus.GaugeValue, float64(occurrences), lbls...)
 		}
 	}
 }
@@ -268,8 +300,9 @@ func NewEndpointSet(
 	dialOpts []grpc.DialOption,
 	unhealthyEndpointTimeout time.Duration,
 	endpointInfoTimeout time.Duration,
+	endpointMetricLabels ...string,
 ) *EndpointSet {
-	endpointsMetric := newEndpointSetNodeCollector()
+	endpointsMetric := newEndpointSetNodeCollector(endpointMetricLabels...)
 	if reg != nil {
 		reg.MustRegister(endpointsMetric)
 	}

pkg/query/endpointset_test.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"math"
 	"net"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -23,6 +24,7 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/pkg/errors"
+	promtestutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/thanos-io/thanos/pkg/component"
 	"github.com/thanos-io/thanos/pkg/info/infopb"
 	"github.com/thanos-io/thanos/pkg/store/labelpb"
@@ -269,13 +271,55 @@ func (e *testEndpoints) CloseOne(addr string) {
 	delete(e.srvs, addr)
 }
 
+func TestTruncateExtLabels(t *testing.T) {
+	const testLength = 5
+
+	for _, tc := range []struct {
+		labelToTruncate string
+		expectedOutput  string
+	}{
+		{
+			labelToTruncate: "{abc}",
+			expectedOutput:  "{abc}",
+		},
+		{
+			labelToTruncate: "{abcd}",
+			expectedOutput:  "{abc}",
+		},
+		{
+			labelToTruncate: "{abcde}",
+			expectedOutput:  "{abc}",
+		},
+		{
+			labelToTruncate: "{abcdef}",
+			expectedOutput:  "{abc}",
+		},
+		{
+			labelToTruncate: "{abcdefghij}",
+			expectedOutput:  "{abc}",
+		},
+	} {
+		t.Run(tc.labelToTruncate, func(t *testing.T) {
+			got := truncateExtLabels(tc.labelToTruncate, testLength)
+			testutil.Equals(t, tc.expectedOutput, got)
+			testutil.Assert(t, len(got) <= testLength)
+		})
+	}
+}
+
 func TestEndpointSetUpdate(t *testing.T) {
+	const metricsMeta = `
+	# HELP thanos_store_nodes_grpc_connections Number of gRPC connection to Store APIs. Opened connection means healthy store APIs available for Querier.
+	# TYPE thanos_store_nodes_grpc_connections gauge
+	`
 	testCases := []struct {
-		name      string
-		endpoints []testEndpointMeta
-		strict    bool
+		name       string
+		endpoints  []testEndpointMeta
+		strict     bool
+		connLabels []string
 
-		expectedEndpoints int
+		expectedEndpoints   int
+		expectedConnMetrics string
 	}{
 		{
 			name: "available endpoint",
@@ -289,7 +333,13 @@ func TestEndpointSetUpdate(t *testing.T) {
 					},
 				},
 			},
+			connLabels: []string{"store_type"},
+
 			expectedEndpoints: 1,
+			expectedConnMetrics: metricsMeta +
+				`
+			thanos_store_nodes_grpc_connections{store_type="sidecar"} 1
+			`,
 		},
 		{
 			name: "unavailable endpoint",
@@ -304,7 +354,9 @@ func TestEndpointSetUpdate(t *testing.T) {
 					},
 				},
 			},
-			expectedEndpoints: 0,
+
+			expectedEndpoints:   0,
+			expectedConnMetrics: "",
 		},
 		{
 			name: "slow endpoint",
@@ -319,7 +371,9 @@ func TestEndpointSetUpdate(t *testing.T) {
 					},
 				},
 			},
-			expectedEndpoints: 0,
+
+			expectedEndpoints:   0,
+			expectedConnMetrics: "",
 		},
 		{
 			name: "strict endpoint",
@@ -334,7 +388,35 @@ func TestEndpointSetUpdate(t *testing.T) {
 				},
 			},
 			strict:            true,
+			connLabels:        []string{"store_type"},
+			expectedEndpoints: 1,
+			expectedConnMetrics: metricsMeta +
+				`
+			thanos_store_nodes_grpc_connections{store_type="sidecar"} 1
+			`,
+		},
+		{
+			name: "long external labels",
+			endpoints: []testEndpointMeta{
+				{
+					InfoResponse: sidecarInfo,
+					// Simulate very long external labels.
+					extlsetFn: func(addr string) []labelpb.ZLabelSet {
+						sLabel := []string{}
+						for i := 0; i < 1000; i++ {
+							sLabel = append(sLabel, "lbl")
+							sLabel = append(sLabel, "val")
+						}
+						return labelpb.ZLabelSetsFromPromLabels(
+							labels.FromStrings(sLabel...),
+						)
+					},
+				},
+			},
 			expectedEndpoints: 1,
+			expectedConnMetrics: metricsMeta + `
+			thanos_store_nodes_grpc_connections{external_labels="{lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val\", lbl=\"val}",store_type="sidecar"} 1
+			`,
 		},
 	}
 
@@ -345,12 +427,15 @@ func TestEndpointSetUpdate(t *testing.T) {
 			defer endpoints.Close()
 
 			discoveredEndpointAddr := endpoints.EndpointAddresses()
-			endpointSet := makeEndpointSet(discoveredEndpointAddr, tc.strict, time.Now)
+			// Specify only "store_type" to exclude "external_labels".
+			endpointSet := makeEndpointSet(discoveredEndpointAddr, tc.strict, time.Now, tc.connLabels...)
 			defer endpointSet.Close()
 
 			endpointSet.Update(context.Background())
 			testutil.Equals(t, tc.expectedEndpoints, len(endpointSet.GetEndpointStatus()))
 			testutil.Equals(t, tc.expectedEndpoints, len(endpointSet.GetStoreClients()))
+
+			testutil.Ok(t, promtestutil.CollectAndCompare(endpointSet.endpointsMetric, strings.NewReader(tc.expectedConnMetrics)))
 		})
 	}
 }
@@ -576,7 +661,7 @@ func TestEndpointSetUpdate_AtomicEndpointAdditions(t *testing.T) {
 	wg.Wait()
 }
 
-func TestEndpointSet_Update(t *testing.T) {
+func TestEndpointSetUpdate_AvailabilityScenarios(t *testing.T) {
 	endpoints, err := startTestEndpoints([]testEndpointMeta{
 		{
 			InfoResponse: sidecarInfo,
@@ -1493,15 +1578,15 @@ func TestUpdateEndpointStateForgetsPreviousErrors(t *testing.T) {
 	testutil.Equals(t, `null`, string(b))
 }
 
-func makeEndpointSet(discoveredEndpointAddr []string, strict bool, now nowFunc) *EndpointSet {
+func makeEndpointSet(discoveredEndpointAddr []string, strict bool, now nowFunc, metricLabels ...string) *EndpointSet {
 	endpointSet := NewEndpointSet(now, nil, nil,
 		func() (specs []*GRPCEndpointSpec) {
 			for _, addr := range discoveredEndpointAddr {
 				specs = append(specs, NewGRPCEndpointSpec(addr, strict))
 			}
 			return specs
 		},
-		testGRPCOpts, time.Minute, time.Second)
+		testGRPCOpts, time.Minute, time.Second, metricLabels...)
 	return endpointSet
 }