adding support for filter by tag in SPM

Signed-off-by: Harshil Gupta <[email protected]>

Author: wololowarrior

cmd/query/app/query_parser.go

@@ -275,6 +275,16 @@ func (p *queryParser) parseMetricsQueryParams(r *http.Request) (bqp metricstore.
 	if err != nil {
 		return bqp, err
 	}
+
+	// Parse tags if present
+	tags, err := p.parseTags(r.Form[tagParam], r.Form[tagsParam])
+	if err != nil {
+		return bqp, err
+	}
+	if len(tags) > 0 {
+		bqp.Tags = tags
+	}
+
 	bqp.EndTime = &endTs
 	bqp.Lookback = &lookback
 	bqp.Step = &step

cmd/query/app/query_parser_test.go

@@ -304,6 +304,67 @@ func TestParseRepeatedSpanKinds(t *testing.T) {
 	}, mqp.SpanKinds)
 }
 
+func TestParseMetricsQueryWithTags(t *testing.T) {
+	tests := []struct {
+		name     string
+		urlStr   string
+		wantTags map[string]string
+		wantErr  bool
+	}{
+		{
+			name:     "with simple tag",
+			urlStr:   "x?service=foo&tag=key:value",
+			wantTags: map[string]string{"key": "value"},
+		},
+		{
+			name:     "with multiple tags",
+			urlStr:   "x?service=foo&tag=key1:value1&tag=key2:value2",
+			wantTags: map[string]string{"key1": "value1", "key2": "value2"},
+		},
+		{
+			name:     "with JSON tags",
+			urlStr:   `x?service=foo&tags={"jsonKey":"jsonValue"}`,
+			wantTags: map[string]string{"jsonKey": "jsonValue"},
+		},
+		{
+			name:     "with both simple and JSON tags",
+			urlStr:   `x?service=foo&tag=key:value&tags={"jsonKey":"jsonValue"}`,
+			wantTags: map[string]string{"key": "value", "jsonKey": "jsonValue"},
+		},
+		{
+			name:    "with malformed tag",
+			urlStr:  "x?service=foo&tag=malformed",
+			wantErr: true,
+		},
+		{
+			name:    "with malformed JSON tags",
+			urlStr:  `x?service=foo&tags={"jsonKey":1234}`,
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			request, err := http.NewRequest(http.MethodGet, tt.urlStr, http.NoBody)
+			require.NoError(t, err)
+
+			parser := &queryParser{
+				timeNow: time.Now,
+			}
+
+			mqp, err := parser.parseMetricsQueryParams(request)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Equal(t, tt.wantTags, mqp.Tags)
+		})
+	}
+}
+
 func TestParameterErrors(t *testing.T) {
 	ts := initializeTestServer(t)
 

internal/storage/metricstore/elasticsearch/query_builder.go

@@ -66,6 +66,15 @@ func (q *QueryBuilder) BuildBoolQuery(params metricstore.BaseQueryParameters, ti
 	spanKindQuery := elastic.NewTermsQuery("tag."+spanKindField, buildInterfaceSlice(normalizeSpanKinds(params.SpanKinds))...)
 	boolQuery.Filter(spanKindQuery)
 
+	// Add tag filters if present
+	if len(params.Tags) > 0 {
+		for k, v := range params.Tags {
+			tagKey := strings.ReplaceAll(k, ".", q.cfg.Tags.DotReplacement)
+			tagQuery := elastic.NewTermQuery("tag."+tagKey, v)
+			boolQuery.Filter(tagQuery)
+		}
+	}
+
 	// Add additional terms queries if provided
 	for _, termQuery := range termsQueries {
 		boolQuery.Filter(termQuery)

internal/storage/metricstore/elasticsearch/query_builder_test.go

@@ -66,6 +66,55 @@ func TestBuildBoolQuery(t *testing.T) {
 	require.Len(t, filterClause, 3) // services, span kinds, time range
 }
 
+func TestBuildBoolQueryWithTags(t *testing.T) {
+	qb := setupTestQB()
+	params := metricstore.BaseQueryParameters{
+		ServiceNames: []string{"service1"},
+		SpanKinds:    []string{"server"},
+		Tags: map[string]string{
+			"environment": "prod",
+			"region":      "us-east-1",
+		},
+	}
+
+	boolQuery := qb.BuildBoolQuery(params, commonTimeRange)
+	require.NotNil(t, boolQuery)
+
+	src, err := boolQuery.Source()
+	require.NoError(t, err)
+
+	queryMap := src.(map[string]any)
+	boolClause := queryMap["bool"].(map[string]any)
+	filterClause := boolClause["filter"].([]any)
+
+	require.Len(t, filterClause, 5) // services, span kinds, 2 tags, time range
+
+	// Validate the presence of tag filters
+	tagFiltersFound := 0
+	for _, filter := range filterClause {
+		filterMap, ok := filter.(map[string]any)
+		if !ok {
+			continue
+		}
+
+		if termQuery, ok := filterMap["term"]; ok {
+			termMap, ok := termQuery.(map[string]any)
+			if !ok {
+				continue
+			}
+
+			if _, ok := termMap["tag.environment"]; ok {
+				tagFiltersFound++
+			}
+			if _, ok := termMap["tag.region"]; ok {
+				tagFiltersFound++
+			}
+		}
+	}
+
+	require.Equal(t, 2, tagFiltersFound, "Expected to find 2 tag filters")
+}
+
 func TestBuildLatenciesAggregation(t *testing.T) {
 	qb := setupTestQB()
 	step := time.Minute

internal/storage/metricstore/prometheus/metricstore/reader.go

@@ -51,6 +51,7 @@ type (
 		spanKindFilter string
 		serviceFilter  string
 		rate           string
+		tagFilters     []string
 	}
 
 	metricsQueryParams struct {
@@ -134,13 +135,23 @@ func (m MetricsReader) GetLatencies(ctx context.Context, requestParams *metricst
 		metricName:          "service_latencies",
 		metricDesc:          fmt.Sprintf("%.2fth quantile latency, grouped by service", requestParams.Quantile),
 		buildPromQuery: func(p promQueryParams) string {
+			// Build filter string including service_name, span_kind, and tags
+			filters := []string{fmt.Sprintf(`service_name =~ %q`, p.serviceFilter)}
+
+			if p.spanKindFilter != "" {
+				filters = append(filters, p.spanKindFilter)
+			}
+
+			// Add tag filters
+			filters = append(filters, p.tagFilters...)
+
+			filterStr := strings.Join(filters, ", ")
+
 			return fmt.Sprintf(
-				// Note: p.spanKindFilter can be ""; trailing commas are okay within a timeseries selection.
-				`histogram_quantile(%.2f, sum(rate(%s_bucket{service_name =~ %q, %s}[%s])) by (%s))`,
+				`histogram_quantile(%.2f, sum(rate(%s_bucket{%s}[%s])) by (%s))`,
 				requestParams.Quantile,
 				m.latencyMetricName,
-				p.serviceFilter,
-				p.spanKindFilter,
+				filterStr,
 				p.rate,
 				p.groupBy,
 			)
@@ -177,12 +188,22 @@ func (m MetricsReader) GetCallRates(ctx context.Context, requestParams *metricst
 		metricName:          "service_call_rate",
 		metricDesc:          "calls/sec, grouped by service",
 		buildPromQuery: func(p promQueryParams) string {
+			// Build filter string including service_name, span_kind, and tags
+			filters := []string{fmt.Sprintf(`service_name =~ %q`, p.serviceFilter)}
+
+			if p.spanKindFilter != "" {
+				filters = append(filters, p.spanKindFilter)
+			}
+
+			// Add tag filters
+			filters = append(filters, p.tagFilters...)
+
+			filterStr := strings.Join(filters, ", ")
+
 			return fmt.Sprintf(
-				// Note: p.spanKindFilter can be ""; trailing commas are okay within a timeseries selection.
-				`sum(rate(%s{service_name =~ %q, %s}[%s])) by (%s)`,
+				`sum(rate(%s{%s}[%s])) by (%s)`,
 				m.callsMetricName,
-				p.serviceFilter,
-				p.spanKindFilter,
+				filterStr,
 				p.rate,
 				p.groupBy,
 			)
@@ -211,11 +232,30 @@ func (m MetricsReader) GetErrorRates(ctx context.Context, requestParams *metrics
 		metricName:          "service_error_rate",
 		metricDesc:          "error rate, computed as a fraction of errors/sec over calls/sec, grouped by service",
 		buildPromQuery: func(p promQueryParams) string {
+			// Build base filters for all queries (service_name)
+			baseFilters := []string{fmt.Sprintf(`service_name =~ %q`, p.serviceFilter)}
+
+			// Add status_code filter only for error rate numerator, must be right after service_name to match test expectations
+			errorFilters := append([]string{}, baseFilters...)
+			errorFilters = append(errorFilters, `status_code = "STATUS_CODE_ERROR"`)
+
+			// Add span_kind filter
+			if p.spanKindFilter != "" {
+				baseFilters = append(baseFilters, p.spanKindFilter)
+				errorFilters = append(errorFilters, p.spanKindFilter)
+			}
+
+			// Add tag filters
+			baseFilters = append(baseFilters, p.tagFilters...)
+			errorFilters = append(errorFilters, p.tagFilters...)
+
+			errorFilterStr := strings.Join(errorFilters, ", ")
+			baseFilterStr := strings.Join(baseFilters, ", ")
+
 			return fmt.Sprintf(
-				// Note: p.spanKindFilter can be ""; trailing commas are okay within a timeseries selection.
-				`sum(rate(%s{service_name =~ %q, status_code = "STATUS_CODE_ERROR", %s}[%s])) by (%s) / sum(rate(%s{service_name =~ %q, %s}[%s])) by (%s)`,
-				m.callsMetricName, p.serviceFilter, p.spanKindFilter, p.rate, p.groupBy,
-				m.callsMetricName, p.serviceFilter, p.spanKindFilter, p.rate, p.groupBy,
+				`sum(rate(%s{%s}[%s])) by (%s) / sum(rate(%s{%s}[%s])) by (%s)`,
+				m.callsMetricName, errorFilterStr, p.rate, p.groupBy,
+				m.callsMetricName, baseFilterStr, p.rate, p.groupBy,
 			)
 		},
 	}
@@ -308,11 +348,23 @@ func (m MetricsReader) buildPromQuery(metricsParams metricsQueryParams) string {
 	if len(metricsParams.SpanKinds) > 0 {
 		spanKindFilter = fmt.Sprintf(`span_kind =~ %q`, strings.Join(metricsParams.SpanKinds, "|"))
 	}
+
+	// Build tag filters
+	var tagFilters []string
+	if len(metricsParams.Tags) > 0 {
+		for k, v := range metricsParams.Tags {
+			// Escape dots in key names for Prometheus compatibility
+			escapedKey := strings.ReplaceAll(k, ".", "_")
+			tagFilters = append(tagFilters, fmt.Sprintf(`%s=%q`, escapedKey, v))
+		}
+	}
+
 	promParams := promQueryParams{
 		serviceFilter:  strings.Join(metricsParams.ServiceNames, "|"),
 		spanKindFilter: spanKindFilter,
 		rate:           promqlDurationString(metricsParams.RatePer),
 		groupBy:        strings.Join(groupBy, ","),
+		tagFilters:     tagFilters,
 	}
 	return metricsParams.buildPromQuery(promParams)
 }

internal/storage/v1/api/metricstore/interface.go

@@ -45,6 +45,8 @@ type BaseQueryParameters struct {
 	RatePer *time.Duration
 	// SpanKinds is the list of span kinds to include (logical OR) in the resulting metrics aggregation.
 	SpanKinds []string
+	// Tags is a map of tag keys and values to filter the metrics by.
+	Tags map[string]string
 }
 
 // LatenciesQueryParameters contains the parameters required for latency metrics queries.