handle conflits in prw v2

.chloggen/handle-conflits.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: 'pkg/translator/prometheusremotewrite'
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: 'Handle conflicts in PRW v2'
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [33661]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

pkg/translator/prometheusremotewrite/metrics_to_prw_v2.go

@@ -33,14 +33,17 @@ func FromMetricsV2(md pmetric.Metrics, settings Settings) (map[string]*writev2.T
 
 // prometheusConverterV2 converts from OTLP to Prometheus write 2.0 format.
 type prometheusConverterV2 struct {
-	// TODO handle conflicts
-	unique      map[uint64]*writev2.TimeSeries
+	unique map[uint64]*writev2.TimeSeries
+	// conflicts is a map of time series signatures(an unique identifier for TS labels) to a list of TSs with the same signature
+	// this is used to handle conflicts, that occur when multiple TSs have the same labels
+	conflicts   map[uint64][]*writev2.TimeSeries
 	symbolTable writev2.SymbolsTable
 }
 
 func newPrometheusConverterV2() *prometheusConverterV2 {
 	return &prometheusConverterV2{
 		unique:      map[uint64]*writev2.TimeSeries{},
+		conflicts:   map[uint64][]*writev2.TimeSeries{},
 		symbolTable: writev2.NewSymbolTable(),
 	}
 }
@@ -109,10 +112,19 @@ func (c *prometheusConverterV2) fromMetrics(md pmetric.Metrics, settings Setting
 
 // timeSeries returns a slice of the writev2.TimeSeries that were converted from OTel format.
 func (c *prometheusConverterV2) timeSeries() []writev2.TimeSeries {
-	allTS := make([]writev2.TimeSeries, 0, len(c.unique))
+	conflicts := 0
+	for _, ts := range c.conflicts {
+		conflicts += len(ts)
+	}
+	allTS := make([]writev2.TimeSeries, 0, len(c.unique)+conflicts)
 	for _, ts := range c.unique {
 		allTS = append(allTS, *ts)
 	}
+	for _, cTS := range c.conflicts {
+		for _, ts := range cTS {
+			allTS = append(allTS, *ts)
+		}
+	}
 	return allTS
 }
 
@@ -131,9 +143,39 @@ func (c *prometheusConverterV2) addSample(sample *writev2.Sample, lbls []prompb.
 		off = c.symbolTable.Symbolize(l.Value)
 		buf = append(buf, off)
 	}
-	ts := writev2.TimeSeries{
+
+	sig := timeSeriesSignature(lbls)
+	ts := &writev2.TimeSeries{
 		LabelsRefs: buf,
 		Samples:    []writev2.Sample{*sample},
 	}
-	c.unique[timeSeriesSignature(lbls)] = &ts
+
+	// check if the time series is already in the unique map
+	if existingTS, ok := c.unique[sig]; ok {
+		// if the time series is already in the unique map, check if it is the same metric
+		if !isSameMetricV2(existingTS, ts) {
+			// if the time series is not the same metric, add it to the conflicts map
+			c.conflicts[sig] = append(c.conflicts[sig], ts)
+		} else {
+			// if the time series is the same metric, add the sample to the existing time series
+			existingTS.Samples = append(existingTS.Samples, *sample)
+		}
+	} else {
+		// if the time series is not in the unique map, add it to the unique map
+		c.unique[sig] = ts
+	}
+}
+
+// isSameMetricV2 checks if two time series are the same metric
+func isSameMetricV2(ts1, ts2 *writev2.TimeSeries) bool {
+	if len(ts1.LabelsRefs) != len(ts2.LabelsRefs) {
+		return false
+	}
+	// As the labels are sorted as name, value, name, value, ... we can compare the labels by index jumping 2 steps at a time
+	for i := 0; i < len(ts1.LabelsRefs); i += 2 {
+		if ts1.LabelsRefs[i] != ts2.LabelsRefs[i] || ts1.LabelsRefs[i+1] != ts2.LabelsRefs[i+1] {
+			return false
+		}
+	}
+	return true
 }

pkg/translator/prometheusremotewrite/metrics_to_prw_v2_test.go

@@ -45,3 +45,36 @@ func TestFromMetricsV2(t *testing.T) {
 	require.ElementsMatch(t, want, slices.Collect(maps.Values(tsMap)))
 	require.ElementsMatch(t, wantedSymbols, symbolsTable.Symbols())
 }
+
+func TestIsSameMetricV2(t *testing.T) {
+	tests := []struct {
+		name string
+		ts1  *writev2.TimeSeries
+		ts2  *writev2.TimeSeries
+		same bool
+	}{
+		{
+			name: "same",
+			same: true,
+			ts1: &writev2.TimeSeries{
+				LabelsRefs: []uint32{1, 2, 3, 4},
+			},
+			ts2: &writev2.TimeSeries{
+				LabelsRefs: []uint32{1, 2, 3, 4},
+			},
+		},
+		{
+			name: "different",
+			same: false,
+			ts1: &writev2.TimeSeries{
+				LabelsRefs: []uint32{1, 2, 3, 4},
+			},
+			ts2: &writev2.TimeSeries{
+				LabelsRefs: []uint32{1, 2, 3, 5},
+			},
+		},
+	}
+	for _, test := range tests {
+		require.Equal(t, test.same, isSameMetricV2(test.ts1, test.ts2))
+	}
+}

pkg/translator/prometheusremotewrite/number_data_points_v2_test.go

@@ -124,8 +124,6 @@ func TestPrometheusConverterV2_addGaugeNumberDataPoints(t *testing.T) {
 	}
 }
 
-// Right now we are not handling duplicates, the second one will just overwrite the first one as this test case shows
-// In follow-up PRs we plan to start handling conflicts and this test will be updated to reflect the new behavior.
 func TestPrometheusConverterV2_addGaugeNumberDataPointsDuplicate(t *testing.T) {
 	ts := uint64(time.Now().UnixNano())
 	metric1 := getIntGaugeMetric(
@@ -138,21 +136,23 @@ func TestPrometheusConverterV2_addGaugeNumberDataPointsDuplicate(t *testing.T) {
 		pcommon.NewMap(),
 		2, ts,
 	)
-	want := func() map[uint64]*writev2.TimeSeries {
-		labels := labels.Labels{
+
+	want := map[uint64]*writev2.TimeSeries{
+		labels.Labels{
 			labels.Label{
 				Name:  labels.MetricName,
 				Value: "test",
 			},
-		}
-		return map[uint64]*writev2.TimeSeries{
-			labels.Hash(): {
-				LabelsRefs: []uint32{1, 2},
-				Samples: []writev2.Sample{
-					{Timestamp: convertTimeStamp(pcommon.Timestamp(ts)), Value: 2},
-				},
-			},
-		}
+		}.Hash(): {
+			LabelsRefs: []uint32{1, 2},
+			Samples: []writev2.Sample{{
+				Timestamp: convertTimeStamp(pcommon.Timestamp(ts)),
+				Value:     1,
+			}, {
+				Timestamp: convertTimeStamp(pcommon.Timestamp(ts)),
+				Value:     2,
+			}},
+		},
 	}
 
 	settings := Settings{
@@ -166,5 +166,5 @@ func TestPrometheusConverterV2_addGaugeNumberDataPointsDuplicate(t *testing.T) {
 	converter.addGaugeNumberDataPoints(metric1.Gauge().DataPoints(), pcommon.NewResource(), settings, metric1.Name())
 	converter.addGaugeNumberDataPoints(metric2.Gauge().DataPoints(), pcommon.NewResource(), settings, metric2.Name())
 
-	assert.Equal(t, want(), converter.unique)
+	assert.Equal(t, want, converter.unique)
 }