Deduplicate metadata in Prometheus Remote Write 2.0 requests

CHANGELOG.md

@@ -53,6 +53,7 @@
 * [ENHANCEMENT] Compactor: Reduce object storage GET calls when updating the bucket index by skipping re-reading parquet converter markers for blocks that already have a valid-version parquet entry in the previous index. #7669
 * [ENHANCEMENT] Upgrade Thanos and promql-engine to latest. #7740
 * [ENHANCEMENT] Ruler: Adjust ruler frontend decoder to not wrap query error messages with execution prefix, this makes error responses consistent between internal and external ruler paths. #7741
+* [ENHANCEMENT] Distributor: Deduplicate metric metadata when converting Prometheus Remote Write 2.0 requests. PRW 2.0 attaches metadata to every series, so a metric family was previously expanded into one `MetricMetadata` per series. #7759
 * [BUGFIX] Querier: Fix queryWithRetry and labelsWithRetry returning (nil, nil) on cancelled context by propagating ctx.Err(). #7370
 * [BUGFIX] Metrics Helper: Fix non-deterministic bucket order in merged histograms by sorting buckets after map iteration, matching Prometheus client library behavior. #7380
 * [BUGFIX] Distributor: Return HTTP 401 Unauthorized when tenant ID resolution fails in the Prometheus Remote Write 2.0 path. #7389

pkg/util/push/push.go

@@ -214,9 +214,19 @@ func setPRW2RespHeader(w http.ResponseWriter, samples, histograms, exemplars int
 	w.Header().Set(rw20WrittenExemplarsHeader, strconv.FormatInt(exemplars, 10))
 }
 
+// v2MetadataKey identifies a unique piece of metadata within a v2 request.
+type v2MetadataKey struct {
+	metricFamilyName string
+	metricType       cortexpb.MetadataV2_MetricType
+	helpRef          uint32
+	unitRef          uint32
+}
+
 func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUnitLabels bool, enableStartTimestamp bool) (v1Req cortexpb.PreallocWriteRequest, err error) {
 	v1Timeseries := make([]cortexpb.PreallocTimeseries, 0, len(req.Timeseries))
 	var v1Metadata []*cortexpb.MetricMetadata
+	// v2 attaches metadata to every series, so a metric family repeats once per series.
+	seenMetadata := make(map[v2MetadataKey]struct{})
 
 	// Release any pulled TimeSeries back to the pool to prevent memory leaks in case of an error.
 	defer func() {
@@ -308,12 +318,21 @@ func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUni
 				return v1Req, err
 			}
 
-			var metadata *cortexpb.MetricMetadata
-			metadata, err = convertV2ToV1Metadata(metricName, symbols, v2Ts.Metadata)
-			if err != nil {
-				return v1Req, err
+			key := v2MetadataKey{
+				metricFamilyName: metricName,
+				metricType:       v2Ts.Metadata.Type,
+				helpRef:          v2Ts.Metadata.HelpRef,
+				unitRef:          v2Ts.Metadata.UnitRef,
+			}
+			if _, ok := seenMetadata[key]; !ok {
+				var metadata *cortexpb.MetricMetadata
+				metadata, err = convertV2ToV1Metadata(metricName, symbols, v2Ts.Metadata)
+				if err != nil {
+					return v1Req, err
+				}
+				seenMetadata[key] = struct{}{}
+				v1Metadata = append(v1Metadata, metadata)
 			}
-			v1Metadata = append(v1Metadata, metadata)
 		}
 	}
 

pkg/util/push/push_test.go

@@ -515,6 +515,94 @@ func Test_convertV2RequestToV1_WithEnableTypeAndUnitLabels(t *testing.T) {
 	}
 }
 
+func Test_convertV2RequestToV1_MetadataDedup(t *testing.T) {
+	symbols := []string{"", "__name__", "test_metric", "pod", "a", "b", "c", "Help text", "seconds", "Other help", "other_metric"}
+
+	// One series per pod, all sharing the same metric family and metadata.
+	sameFamily := func(nameRef uint32, podRef uint32, meta cortexpb.MetadataV2) cortexpb.PreallocTimeseriesV2 {
+		return cortexpb.PreallocTimeseriesV2{
+			TimeSeriesV2: &cortexpb.TimeSeriesV2{
+				LabelsRefs: []uint32{1, nameRef, 3, podRef},
+				Metadata:   meta,
+				Samples:    []cortexpb.Sample{{Value: 1, TimestampMs: 1}},
+			},
+		}
+	}
+
+	counterMeta := cortexpb.MetadataV2{Type: cortexpb.METRIC_TYPE_COUNTER, HelpRef: 7, UnitRef: 8}
+
+	tests := []struct {
+		name             string
+		timeseries       []cortexpb.PreallocTimeseriesV2
+		expectedMetadata []*cortexpb.MetricMetadata
+	}{
+		{
+			name: "identical metadata across series of the same family is deduped",
+			timeseries: []cortexpb.PreallocTimeseriesV2{
+				sameFamily(2, 4, counterMeta),
+				sameFamily(2, 5, counterMeta),
+				sameFamily(2, 6, counterMeta),
+			},
+			expectedMetadata: []*cortexpb.MetricMetadata{
+				{Type: cortexpb.COUNTER, MetricFamilyName: "test_metric", Help: "Help text", Unit: "seconds"},
+			},
+		},
+		{
+			name: "distinct families are kept",
+			timeseries: []cortexpb.PreallocTimeseriesV2{
+				sameFamily(2, 4, counterMeta),
+				sameFamily(10, 4, counterMeta),
+				sameFamily(2, 5, counterMeta),
+			},
+			expectedMetadata: []*cortexpb.MetricMetadata{
+				{Type: cortexpb.COUNTER, MetricFamilyName: "test_metric", Help: "Help text", Unit: "seconds"},
+				{Type: cortexpb.COUNTER, MetricFamilyName: "other_metric", Help: "Help text", Unit: "seconds"},
+			},
+		},
+		{
+			name: "same family with differing type, help or unit is kept",
+			timeseries: []cortexpb.PreallocTimeseriesV2{
+				sameFamily(2, 4, counterMeta),
+				sameFamily(2, 5, cortexpb.MetadataV2{Type: cortexpb.METRIC_TYPE_GAUGE, HelpRef: 7, UnitRef: 8}),
+				sameFamily(2, 6, cortexpb.MetadataV2{Type: cortexpb.METRIC_TYPE_COUNTER, HelpRef: 9, UnitRef: 8}),
+				sameFamily(2, 4, cortexpb.MetadataV2{Type: cortexpb.METRIC_TYPE_COUNTER, HelpRef: 7, UnitRef: 0}),
+			},
+			expectedMetadata: []*cortexpb.MetricMetadata{
+				{Type: cortexpb.COUNTER, MetricFamilyName: "test_metric", Help: "Help text", Unit: "seconds"},
+				{Type: cortexpb.GAUGE, MetricFamilyName: "test_metric", Help: "Help text", Unit: "seconds"},
+				{Type: cortexpb.COUNTER, MetricFamilyName: "test_metric", Help: "Other help", Unit: "seconds"},
+				{Type: cortexpb.COUNTER, MetricFamilyName: "test_metric", Help: "Help text", Unit: ""},
+			},
+		},
+		{
+			name: "series without metadata produce none",
+			timeseries: []cortexpb.PreallocTimeseriesV2{
+				sameFamily(2, 4, cortexpb.MetadataV2{}),
+				sameFamily(2, 5, cortexpb.MetadataV2{}),
+			},
+			expectedMetadata: nil,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			v2Req := cortexpb.PreallocWriteRequestV2{
+				WriteRequestV2: cortexpb.WriteRequestV2{
+					Symbols:    symbols,
+					Timeseries: test.timeseries,
+				},
+			}
+
+			v1Req, err := convertV2RequestToV1(&v2Req, false, false)
+			require.NoError(t, err)
+
+			// Dedup must not drop any series.
+			require.Len(t, v1Req.Timeseries, len(test.timeseries))
+			require.Equal(t, test.expectedMetadata, v1Req.Metadata)
+		})
+	}
+}
+
 func Test_convertV2RequestToV1(t *testing.T) {
 	var v2Req cortexpb.PreallocWriteRequestV2