Skip to content

Commit a330338

Browse files
SungJin1212SungJin1212
authored
Deduplicate metadata in Prometheus Remote Write 2.0 requests (#7760)
* deduplicate metric metadata in Prometheus Remote Write 2.0 requests Signed-off-by: SungJin1212 <tjdwls1201@gmail.com> * fix changelog Signed-off-by: SungJin1212 <tjdwls1201@gmail.com> --------- Signed-off-by: SungJin1212 <tjdwls1201@gmail.com>
1 parent 0363900 commit a330338

3 files changed

Lines changed: 113 additions & 5 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
* [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
5555
* [ENHANCEMENT] Upgrade Thanos and promql-engine to latest. #7740
5656
* [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
57+
* [ENHANCEMENT] Distributor: Deduplicate metric metadata when converting PRW 2.0 requests. PRW 2.0 attaches metadata to every series, so a metric family was previously expanded into one `MetricMetadata` per series. #7760
5758
* [BUGFIX] Querier: Fix queryWithRetry and labelsWithRetry returning (nil, nil) on cancelled context by propagating ctx.Err(). #7370
5859
* [BUGFIX] Metrics Helper: Fix non-deterministic bucket order in merged histograms by sorting buckets after map iteration, matching Prometheus client library behavior. #7380
5960
* [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‎

Lines changed: 24 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -214,9 +214,19 @@ func setPRW2RespHeader(w http.ResponseWriter, samples, histograms, exemplars int
214214
w.Header().Set(rw20WrittenExemplarsHeader, strconv.FormatInt(exemplars, 10))
215215
}
216216

217+
// v2MetadataKey identifies a unique piece of metadata within a v2 request.
218+
type v2MetadataKey struct {
219+
metricFamilyName string
220+
metricType cortexpb.MetadataV2_MetricType
221+
helpRef uint32
222+
unitRef uint32
223+
}
224+
217225
func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUnitLabels bool, enableStartTimestamp bool) (v1Req cortexpb.PreallocWriteRequest, err error) {
218226
v1Timeseries := make([]cortexpb.PreallocTimeseries, 0, len(req.Timeseries))
219227
var v1Metadata []*cortexpb.MetricMetadata
228+
// v2 attaches metadata to every series, so a metric family repeats once per series.
229+
seenMetadata := make(map[v2MetadataKey]struct{})
220230

221231
// Release any pulled TimeSeries back to the pool to prevent memory leaks in case of an error.
222232
defer func() {
@@ -308,12 +318,21 @@ func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUni
308318
return v1Req, err
309319
}
310320

311-
var metadata *cortexpb.MetricMetadata
312-
metadata, err = convertV2ToV1Metadata(metricName, symbols, v2Ts.Metadata)
313-
if err != nil {
314-
return v1Req, err
321+
key := v2MetadataKey{
322+
metricFamilyName: metricName,
323+
metricType: v2Ts.Metadata.Type,
324+
helpRef: v2Ts.Metadata.HelpRef,
325+
unitRef: v2Ts.Metadata.UnitRef,
326+
}
327+
if _, ok := seenMetadata[key]; !ok {
328+
var metadata *cortexpb.MetricMetadata
329+
metadata, err = convertV2ToV1Metadata(metricName, symbols, v2Ts.Metadata)
330+
if err != nil {
331+
return v1Req, err
332+
}
333+
seenMetadata[key] = struct{}{}
334+
v1Metadata = append(v1Metadata, metadata)
315335
}
316-
v1Metadata = append(v1Metadata, metadata)
317336
}
318337
}
319338

‎pkg/util/push/push_test.go‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -515,6 +515,94 @@ func Test_convertV2RequestToV1_WithEnableTypeAndUnitLabels(t *testing.T) {
515515
}
516516
}
517517

518+
func Test_convertV2RequestToV1_MetadataDedup(t *testing.T) {
519+
symbols := []string{"", "__name__", "test_metric", "pod", "a", "b", "c", "Help text", "seconds", "Other help", "other_metric"}
520+
521+
// One series per pod, all sharing the same metric family and metadata.
522+
sameFamily := func(nameRef uint32, podRef uint32, meta cortexpb.MetadataV2) cortexpb.PreallocTimeseriesV2 {
523+
return cortexpb.PreallocTimeseriesV2{
524+
TimeSeriesV2: &cortexpb.TimeSeriesV2{
525+
LabelsRefs: []uint32{1, nameRef, 3, podRef},
526+
Metadata: meta,
527+
Samples: []cortexpb.Sample{{Value: 1, TimestampMs: 1}},
528+
},
529+
}
530+
}
531+
532+
counterMeta := cortexpb.MetadataV2{Type: cortexpb.METRIC_TYPE_COUNTER, HelpRef: 7, UnitRef: 8}
533+
534+
tests := []struct {
535+
name string
536+
timeseries []cortexpb.PreallocTimeseriesV2
537+
expectedMetadata []*cortexpb.MetricMetadata
538+
}{
539+
{
540+
name: "identical metadata across series of the same family is deduped",
541+
timeseries: []cortexpb.PreallocTimeseriesV2{
542+
sameFamily(2, 4, counterMeta),
543+
sameFamily(2, 5, counterMeta),
544+
sameFamily(2, 6, counterMeta),
545+
},
546+
expectedMetadata: []*cortexpb.MetricMetadata{
547+
{Type: cortexpb.COUNTER, MetricFamilyName: "test_metric", Help: "Help text", Unit: "seconds"},
548+
},
549+
},
550+
{
551+
name: "distinct families are kept",
552+
timeseries: []cortexpb.PreallocTimeseriesV2{
553+
sameFamily(2, 4, counterMeta),
554+
sameFamily(10, 4, counterMeta),
555+
sameFamily(2, 5, counterMeta),
556+
},
557+
expectedMetadata: []*cortexpb.MetricMetadata{
558+
{Type: cortexpb.COUNTER, MetricFamilyName: "test_metric", Help: "Help text", Unit: "seconds"},
559+
{Type: cortexpb.COUNTER, MetricFamilyName: "other_metric", Help: "Help text", Unit: "seconds"},
560+
},
561+
},
562+
{
563+
name: "same family with differing type, help or unit is kept",
564+
timeseries: []cortexpb.PreallocTimeseriesV2{
565+
sameFamily(2, 4, counterMeta),
566+
sameFamily(2, 5, cortexpb.MetadataV2{Type: cortexpb.METRIC_TYPE_GAUGE, HelpRef: 7, UnitRef: 8}),
567+
sameFamily(2, 6, cortexpb.MetadataV2{Type: cortexpb.METRIC_TYPE_COUNTER, HelpRef: 9, UnitRef: 8}),
568+
sameFamily(2, 4, cortexpb.MetadataV2{Type: cortexpb.METRIC_TYPE_COUNTER, HelpRef: 7, UnitRef: 0}),
569+
},
570+
expectedMetadata: []*cortexpb.MetricMetadata{
571+
{Type: cortexpb.COUNTER, MetricFamilyName: "test_metric", Help: "Help text", Unit: "seconds"},
572+
{Type: cortexpb.GAUGE, MetricFamilyName: "test_metric", Help: "Help text", Unit: "seconds"},
573+
{Type: cortexpb.COUNTER, MetricFamilyName: "test_metric", Help: "Other help", Unit: "seconds"},
574+
{Type: cortexpb.COUNTER, MetricFamilyName: "test_metric", Help: "Help text", Unit: ""},
575+
},
576+
},
577+
{
578+
name: "series without metadata produce none",
579+
timeseries: []cortexpb.PreallocTimeseriesV2{
580+
sameFamily(2, 4, cortexpb.MetadataV2{}),
581+
sameFamily(2, 5, cortexpb.MetadataV2{}),
582+
},
583+
expectedMetadata: nil,
584+
},
585+
}
586+
587+
for _, test := range tests {
588+
t.Run(test.name, func(t *testing.T) {
589+
v2Req := cortexpb.PreallocWriteRequestV2{
590+
WriteRequestV2: cortexpb.WriteRequestV2{
591+
Symbols: symbols,
592+
Timeseries: test.timeseries,
593+
},
594+
}
595+
596+
v1Req, err := convertV2RequestToV1(&v2Req, false, false)
597+
require.NoError(t, err)
598+
599+
// Dedup must not drop any series.
600+
require.Len(t, v1Req.Timeseries, len(test.timeseries))
601+
require.Equal(t, test.expectedMetadata, v1Req.Metadata)
602+
})
603+
}
604+
}
605+
518606
func Test_convertV2RequestToV1(t *testing.T) {
519607
var v2Req cortexpb.PreallocWriteRequestV2
520608

0 commit comments

Comments
 (0)