feat: prometheus translation add support for rw2

pkg/translator/prometheusremotewrite/helper_v2.go

@@ -0,0 +1,142 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package prometheusremotewrite // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheusremotewrite"
+
+import (
+	"fmt"
+	prometheustranslator "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheus"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/labels"
+	writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	conventions "go.opentelemetry.io/collector/semconv/v1.25.0"
+	"log"
+	"slices"
+)
+
+// getOrCreateTimeSeries returns the time series corresponding to the label set if existent, and false.
+// Otherwise it creates a new one and returns that, and true.
+func (c *prometheusConverterV2) getOrCreateTimeSeries(lbls labels.Labels) (*writev2.TimeSeries, bool) {
+	h := lbls.Hash()
+	ts := c.unique[h]
+
+	if ts != nil {
+		if c.isSameMetricV2(ts, lbls) {
+			// We already have this metric
+			return ts, false
+		}
+
+		// Look for a matching conflict
+		for _, cTS := range c.conflicts[h] {
+			if c.isSameMetricV2(cTS, lbls) {
+				// We already have this metric
+				return cTS, false
+			}
+		}
+
+		// New conflict
+		ts = &writev2.TimeSeries{}
+		ts.LabelsRefs = c.symbolTable.SymbolizeLabels(lbls, ts.LabelsRefs)
+		c.conflicts[h] = append(c.conflicts[h], ts)
+		return ts, true
+
+	}
+
+	// This metric is new
+	ts = &writev2.TimeSeries{}
+	ts.LabelsRefs = c.symbolTable.SymbolizeLabels(lbls, ts.LabelsRefs)
+	c.unique[h] = ts
+	return ts, true
+}
+
+// createAttributes creates a slice of Prometheus Labels with OTLP attributes and pairs of string values.
+// Unpaired string values are ignored. String pairs overwrite OTLP labels if collisions happen and
+// if logOnOverwrite is true, the overwrite is logged. Resulting label names are sanitized.
+func createAttributesV2(resource pcommon.Resource, attributes pcommon.Map, externalLabels map[string]string,
+	ignoreAttrs []string, logOnOverwrite bool, extras ...string) labels.Labels {
+	resourceAttrs := resource.Attributes()
+	serviceName, haveServiceName := resourceAttrs.Get(conventions.AttributeServiceName)
+	instance, haveInstanceID := resourceAttrs.Get(conventions.AttributeServiceInstanceID)
+
+	// Calculate the maximum possible number of labels we could return so we can preallocate l
+	maxLabelCount := attributes.Len() + len(externalLabels) + len(extras)/2
+
+	if haveServiceName {
+		maxLabelCount++
+	}
+
+	if haveInstanceID {
+		maxLabelCount++
+	}
+
+	// map ensures no duplicate label name
+	l := make(map[string]string, maxLabelCount)
+
+	// Ensure attributes are sorted by key for consistent merging of keys which
+	// collide when sanitized.
+	serieslabels := labels.Labels{}
+	// XXX: Should we always drop service namespace/service name/service instance ID from the labels
+	// (as they get mapped to other Prometheus labels)?
+	attributes.Range(func(key string, value pcommon.Value) bool {
+		if !slices.Contains(ignoreAttrs, key) {
+			serieslabels = append(serieslabels, labels.Label{Name: key, Value: value.AsString()})
+		}
+		return true
+	})
+	// Afaik not needed
+	//sort.Stable(ByLabelName(labels))
+
+	for _, label := range serieslabels {
+		var finalKey = prometheustranslator.NormalizeLabel(label.Name)
+		if existingValue, alreadyExists := l[finalKey]; alreadyExists {
+			l[finalKey] = existingValue + ";" + label.Value
+		} else {
+			l[finalKey] = label.Value
+		}
+	}
+
+	// Map service.name + service.namespace to job
+	if haveServiceName {
+		val := serviceName.AsString()
+		if serviceNamespace, ok := resourceAttrs.Get(conventions.AttributeServiceNamespace); ok {
+			val = fmt.Sprintf("%s/%s", serviceNamespace.AsString(), val)
+		}
+		l[model.JobLabel] = val
+	}
+	// Map service.instance.id to instance
+	if haveInstanceID {
+		l[model.InstanceLabel] = instance.AsString()
+	}
+	for key, value := range externalLabels {
+		// External labels have already been sanitized
+		if _, alreadyExists := l[key]; alreadyExists {
+			// Skip external labels if they are overridden by metric attributes
+			continue
+		}
+		l[key] = value
+	}
+
+	for i := 0; i < len(extras); i += 2 {
+		if i+1 >= len(extras) {
+			break
+		}
+		_, found := l[extras[i]]
+		if found && logOnOverwrite {
+			log.Println("label " + extras[i] + " is overwritten. Check if Prometheus reserved labels are used.")
+		}
+		// internal labels should be maintained
+		name := extras[i]
+		if !(len(name) > 4 && name[:2] == "__" && name[len(name)-2:] == "__") {
+			name = prometheustranslator.NormalizeLabel(name)
+		}
+		l[name] = extras[i+1]
+	}
+
+	serieslabels = serieslabels[:0]
+	for k, v := range l {
+		serieslabels = append(serieslabels, labels.Label{Name: k, Value: v})
+	}
+
+	return serieslabels
+}

pkg/translator/prometheusremotewrite/helper_v2_test.go

@@ -0,0 +1,212 @@
+package prometheusremotewrite
+
+import (
+	"github.com/prometheus/prometheus/model/labels"
+	writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"testing"
+)
+
+func TestPrometheusConverter_getOrCreateTimeSeriesV2(t *testing.T) {
+	converter := newPrometheusConverterV2()
+	lbls := labels.Labels{
+		labels.Label{
+			Name:  "key1",
+			Value: "value1",
+		},
+		labels.Label{
+			Name:  "key2",
+			Value: "value2",
+		},
+	}
+	ts, created := converter.getOrCreateTimeSeries(lbls)
+	require.NotNil(t, ts)
+	require.True(t, created)
+
+	var b labels.ScratchBuilder
+	createdLabels := ts.ToLabels(&b, converter.symbolTable.Symbols())
+
+	// Now, get (not create) the unique time series
+	gotTS, created := converter.getOrCreateTimeSeries(createdLabels)
+	require.Same(t, ts, gotTS)
+	require.False(t, created)
+
+	var keys []uint64
+	for k := range converter.unique {
+		keys = append(keys, k)
+	}
+	require.Len(t, keys, 1)
+	h := keys[0]
+
+	// Make sure that state is correctly set
+	require.Equal(t, map[uint64]*writev2.TimeSeries{
+		h: ts,
+	}, converter.unique)
+	require.Empty(t, converter.conflicts)
+
+	// Fake a hash collision, by making this not equal to the next series with the same hash
+	createdLabels = append(createdLabels, labels.Label{Name: "key3", Value: "value3"})
+	ts.LabelsRefs = converter.symbolTable.SymbolizeLabels(createdLabels, ts.LabelsRefs)
+
+	// Make the first hash collision
+	cTS1, created := converter.getOrCreateTimeSeries(lbls)
+	require.NotNil(t, cTS1)
+	require.True(t, created)
+	require.Equal(t, map[uint64][]*writev2.TimeSeries{
+		h: {cTS1},
+	}, converter.conflicts)
+
+	// Fake a hash collision, by making this not equal to the next series with the same hash
+	createdLabels1 := cTS1.ToLabels(&b, converter.symbolTable.Symbols())
+	createdLabels1 = append(createdLabels1, labels.Label{Name: "key3", Value: "value3"})
+	cTS1.LabelsRefs = converter.symbolTable.SymbolizeLabels(createdLabels1, ts.LabelsRefs)
+
+	// Make the second hash collision
+	cTS2, created := converter.getOrCreateTimeSeries(lbls)
+	require.NotNil(t, cTS2)
+	require.True(t, created)
+	require.Equal(t, map[uint64][]*writev2.TimeSeries{
+		h: {cTS1, cTS2},
+	}, converter.conflicts)
+
+	// Now, get (not create) the second colliding time series
+	gotCTS2, created := converter.getOrCreateTimeSeries(lbls)
+	require.Same(t, cTS2, gotCTS2)
+	require.False(t, created)
+	require.Equal(t, map[uint64][]*writev2.TimeSeries{
+		h: {cTS1, cTS2},
+	}, converter.conflicts)
+
+	require.Equal(t, map[uint64]*writev2.TimeSeries{
+		h: ts,
+	}, converter.unique)
+}
+
+// Test_createLabelSet checks resultant label names are sanitized and label in extra overrides label in labels if
+// collision happens. It does not check whether labels are not sorted
+func Test_createLabelSetV2(t *testing.T) {
+	tests := []struct {
+		name           string
+		resource       pcommon.Resource
+		orig           pcommon.Map
+		externalLabels map[string]string
+		extras         []string
+		want           labels.Labels
+	}{
+		{
+			"labels_clean",
+			pcommon.NewResource(),
+			lbs1,
+			map[string]string{},
+			[]string{label31, value31, label32, value32},
+			getPromLabelsV2(label11, value11, label12, value12, label31, value31, label32, value32),
+		},
+		{
+			"labels_with_resource",
+			func() pcommon.Resource {
+				res := pcommon.NewResource()
+				res.Attributes().PutStr("service.name", "prometheus")
+				res.Attributes().PutStr("service.instance.id", "127.0.0.1:8080")
+				return res
+			}(),
+			lbs1,
+			map[string]string{},
+			[]string{label31, value31, label32, value32},
+			getPromLabelsV2(label11, value11, label12, value12, label31, value31, label32, value32, "job", "prometheus", "instance", "127.0.0.1:8080"),
+		},
+		{
+			"labels_with_nonstring_resource",
+			func() pcommon.Resource {
+				res := pcommon.NewResource()
+				res.Attributes().PutInt("service.name", 12345)
+				res.Attributes().PutBool("service.instance.id", true)
+				return res
+			}(),
+			lbs1,
+			map[string]string{},
+			[]string{label31, value31, label32, value32},
+			getPromLabelsV2(label11, value11, label12, value12, label31, value31, label32, value32, "job", "12345", "instance", "true"),
+		},
+		{
+			"labels_duplicate_in_extras",
+			pcommon.NewResource(),
+			lbs1,
+			map[string]string{},
+			[]string{label11, value31},
+			getPromLabelsV2(label11, value31, label12, value12),
+		},
+		{
+			"labels_dirty",
+			pcommon.NewResource(),
+			lbs1Dirty,
+			map[string]string{},
+			[]string{label31 + dirty1, value31, label32, value32},
+			getPromLabelsV2(label11+"_", value11, "key_"+label12, value12, label31+"_", value31, label32, value32),
+		},
+		{
+			"no_original_case",
+			pcommon.NewResource(),
+			pcommon.NewMap(),
+			nil,
+			[]string{label31, value31, label32, value32},
+			getPromLabelsV2(label31, value31, label32, value32),
+		},
+		{
+			"empty_extra_case",
+			pcommon.NewResource(),
+			lbs1,
+			map[string]string{},
+			[]string{"", ""},
+			getPromLabelsV2(label11, value11, label12, value12, "", ""),
+		},
+		{
+			"single_left_over_case",
+			pcommon.NewResource(),
+			lbs1,
+			map[string]string{},
+			[]string{label31, value31, label32},
+			getPromLabelsV2(label11, value11, label12, value12, label31, value31),
+		},
+		{
+			"valid_external_labels",
+			pcommon.NewResource(),
+			lbs1,
+			exlbs1,
+			[]string{label31, value31, label32, value32},
+			getPromLabelsV2(label11, value11, label12, value12, label41, value41, label31, value31, label32, value32),
+		},
+		{
+			"overwritten_external_labels",
+			pcommon.NewResource(),
+			lbs1,
+			exlbs2,
+			[]string{label31, value31, label32, value32},
+			getPromLabelsV2(label11, value11, label12, value12, label31, value31, label32, value32),
+		},
+		{
+			"colliding attributes",
+			pcommon.NewResource(),
+			lbsColliding,
+			nil,
+			[]string{label31, value31, label32, value32},
+			getPromLabelsV2(collidingSanitized, value11+";"+value12, label31, value31, label32, value32),
+		},
+		{
+			"sanitize_labels_starts_with_underscore",
+			pcommon.NewResource(),
+			lbs3,
+			exlbs1,
+			[]string{label31, value31, label32, value32},
+			getPromLabelsV2(label11, value11, label12, value12, "key"+label51, value51, label41, value41, label31, value31, label32, value32),
+		},
+	}
+	// run tests
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			res := createAttributesV2(tt.resource, tt.orig, tt.externalLabels, nil, true, tt.extras...)
+			assert.ElementsMatch(t, tt.want, res)
+		})
+	}
+}