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,19 @@ 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
-	symbolTable writev2.SymbolsTable
+	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 or when different labels generate the same signature.
+	conflicts map[uint64][]*writev2.TimeSeries
+	// conflictCount is used to track the number of conflicts that were encountered.
+	conflictCount int
+	symbolTable   writev2.SymbolsTable
 }
 
 func newPrometheusConverterV2() *prometheusConverterV2 {
 	return &prometheusConverterV2{
 		unique:      map[uint64]*writev2.TimeSeries{},
+		conflicts:   map[uint64][]*writev2.TimeSeries{},
 		symbolTable: writev2.NewSymbolTable(),
 	}
 }
@@ -109,10 +114,15 @@ 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))
+	allTS := make([]writev2.TimeSeries, 0, len(c.unique)+c.conflictCount)
 	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 +141,40 @@ 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)
+			c.conflictCount++
+		} 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

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/prometheus/prompb"
 	writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -45,3 +46,95 @@ 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))
+	}
+}
+
+func TestConflictHandling(t *testing.T) {
+	// Test 1: No conflicts - different metrics should have different hashes
+	t.Run("different metrics should not conflict", func(t *testing.T) {
+		converter := newPrometheusConverterV2()
+
+		metric1 := createSample(1.0, []prompb.Label{
+			{Name: "name1", Value: "value1"},
+			{Name: "name2", Value: "value2"},
+		})
+
+		metric2 := createSample(2.0, []prompb.Label{
+			{Name: "name3", Value: "value3"},
+			{Name: "name4", Value: "value4"},
+		})
+
+		converter.addSample(metric1.sample, metric1.labels)
+		converter.addSample(metric2.sample, metric2.labels)
+
+		require.Equal(t, 0, converter.conflictCount)
+		require.Len(t, converter.unique, 2)
+	})
+
+	// Test 2: Same metric - should be merged
+	t.Run("same metric should be merged", func(t *testing.T) {
+		converter := newPrometheusConverterV2()
+
+		labels := []prompb.Label{
+			{Name: "name", Value: "value"},
+		}
+
+		sample1 := &writev2.Sample{Value: 1.0, Timestamp: 1000}
+		sample2 := &writev2.Sample{Value: 2.0, Timestamp: 2000}
+
+		converter.addSample(sample1, labels)
+		converter.addSample(sample2, labels)
+
+		require.Equal(t, 0, converter.conflictCount)
+		require.Len(t, converter.unique, 1)
+		require.Len(t, converter.unique[timeSeriesSignature(labels)].Samples, 2)
+	})
+	// TODO: Test 3 Conflict - different metrics with same hash
+}
+
+// Helper function to create a sample with labels
+type metricSample struct {
+	sample *writev2.Sample
+	labels []prompb.Label
+}
+
+func createSample(value float64, labels []prompb.Label) metricSample {
+	return metricSample{
+		sample: &writev2.Sample{
+			Value:     value,
+			Timestamp: 1000,
+		},
+		labels: labels,
+	}
+}

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)
 }