feat: prometheus translation add support for rw2

pkg/translator/prometheusremotewrite/helper_v2.go

@@ -0,0 +1,144 @@
+// 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"
+)
+
+// TODO implement this fully
+// 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[0] = 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]
+	}
+
+	// TODO what was this for??
+	//labels = labels[:0]
+	//for k, v := range l {
+	//	labels = append(labels, prompb.Label{Name: k, Value: v})
+	//}
+
+	return serieslabels
+}

pkg/translator/prometheusremotewrite/metrics_to_prw_v2.go

@@ -0,0 +1,151 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package prometheusremotewrite // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheusremotewrite"
+
+import (
+	"errors"
+	"fmt"
+	"github.com/prometheus/prometheus/model/labels"
+	writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
+	"strconv"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/multierr"
+
+	prometheustranslator "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheus"
+)
+
+// FromMetricsV2 converts pmetric.Metrics to Prometheus remote write format 2.0.
+func FromMetricsV2(md pmetric.Metrics, settings Settings) (map[string]*writev2.TimeSeries, error) {
+	c := newPrometheusConverterV2()
+	errs := c.fromMetrics(md, settings)
+	tss := c.timeSeries()
+	out := make(map[string]*writev2.TimeSeries, len(tss))
+	for i := range tss {
+		out[strconv.Itoa(i)] = &tss[i]
+	}
+
+	return out, errs
+}
+
+// prometheusConverter converts from OTel write format to Prometheus write format.
+type prometheusConverterV2 struct {
+	unique      map[uint64]*writev2.TimeSeries
+	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(),
+	}
+}
+
+// fromMetrics converts pmetric.Metrics to Prometheus remote write format.
+func (c *prometheusConverterV2) fromMetrics(md pmetric.Metrics, settings Settings) (errs error) {
+	resourceMetricsSlice := md.ResourceMetrics()
+	for i := 0; i < resourceMetricsSlice.Len(); i++ {
+		resourceMetrics := resourceMetricsSlice.At(i)
+		resource := resourceMetrics.Resource()
+		scopeMetricsSlice := resourceMetrics.ScopeMetrics()
+		// keep track of the most recent timestamp in the ResourceMetrics for
+		// use with the "target" info metric
+		var mostRecentTimestamp pcommon.Timestamp
+		for j := 0; j < scopeMetricsSlice.Len(); j++ {
+			metricSlice := scopeMetricsSlice.At(j).Metrics()
+
+			// TODO: decide if instrumentation library information should be exported as labels
+			for k := 0; k < metricSlice.Len(); k++ {
+				metric := metricSlice.At(k)
+				mostRecentTimestamp = maxTimestamp(mostRecentTimestamp, mostRecentTimestampInMetric(metric))
+
+				if !isValidAggregationTemporality(metric) {
+					errs = multierr.Append(errs, fmt.Errorf("invalid temporality and type combination for metric %q", metric.Name()))
+					continue
+				}
+
+				promName := prometheustranslator.BuildCompliantName(metric, settings.Namespace, settings.AddMetricSuffixes)
+
+				// handle individual metrics based on type
+				//exhaustive:enforce
+				switch metric.Type() {
+				case pmetric.MetricTypeGauge:
+					dataPoints := metric.Gauge().DataPoints()
+					if dataPoints.Len() == 0 {
+						errs = multierr.Append(errs, fmt.Errorf("empty data points. %s is dropped", metric.Name()))
+						break
+					}
+					c.addGaugeNumberDataPoints(dataPoints, resource, settings, promName)
+				case pmetric.MetricTypeSum:
+					// TODO implement
+				case pmetric.MetricTypeHistogram:
+					// TODO implement
+				case pmetric.MetricTypeExponentialHistogram:
+					// TODO implement
+				case pmetric.MetricTypeSummary:
+					// TODO implement
+				default:
+					errs = multierr.Append(errs, errors.New("unsupported metric type"))
+				}
+			}
+		}
+		// TODO implement
+		//addResourceTargetInfov2(resource, settings, mostRecentTimestamp, c)
+	}
+
+	return
+}
+
+// timeSeries returns a slice of the writev2.TimeSeries that were converted from OTel format.
+func (c *prometheusConverterV2) timeSeries() []writev2.TimeSeries {
+	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
+}
+
+func (c *prometheusConverterV2) addExemplars(dataPoint pmetric.HistogramDataPoint, bucketBounds []bucketBoundsData) {
+	// TODO implement
+	return
+}
+
+func (c *prometheusConverterV2) addSample(sample *writev2.Sample, lbls labels.Labels) *writev2.TimeSeries {
+	if sample == nil || len(lbls) == 0 {
+		// This shouldn't happen
+		return nil
+	}
+
+	ts, _ := c.getOrCreateTimeSeries(lbls)
+	ts.Samples = append(ts.Samples, *sample)
+	return ts
+}
+
+func (c *prometheusConverterV2) isSameMetricV2(ts *writev2.TimeSeries, lbls labels.Labels) bool {
+	b := labels.NewScratchBuilder(0)
+	seriesLabels := ts.ToLabels(&b, c.symbolTable.Symbols())
+
+	if len(seriesLabels) != len(lbls) {
+		return false
+	}
+	for i, l := range seriesLabels {
+		if l.Name != seriesLabels[i].Name || l.Value != seriesLabels[i].Value {
+			return false
+		}
+	}
+	return true
+}

pkg/translator/prometheusremotewrite/metrics_to_prw_v2_test.go

@@ -0,0 +1,26 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package prometheusremotewrite
+
+import (
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestFromMetricsV2(t *testing.T) {
+	settings := Settings{
+		Namespace:           "",
+		ExternalLabels:      nil,
+		DisableTargetInfo:   false,
+		ExportCreatedMetric: false,
+		AddMetricSuffixes:   false,
+		SendMetadata:        false,
+	}
+
+	payload := createExportRequest(5, 0, 1, 3, 0)
+
+	tsMap, err := FromMetricsV2(payload.Metrics(), settings)
+	require.NoError(t, err)
+	require.NotNil(t, tsMap)
+}

pkg/translator/prometheusremotewrite/number_data_points_v2.go

@@ -0,0 +1,44 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package prometheusremotewrite // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheusremotewrite"
+
+import (
+	writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
+	"math"
+
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/value"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+func (c *prometheusConverterV2) addGaugeNumberDataPoints(dataPoints pmetric.NumberDataPointSlice,
+	resource pcommon.Resource, settings Settings, name string) {
+	for x := 0; x < dataPoints.Len(); x++ {
+		pt := dataPoints.At(x)
+		labels := createAttributesV2(
+			resource,
+			pt.Attributes(),
+			settings.ExternalLabels,
+			nil,
+			true,
+			model.MetricNameLabel,
+			name,
+		)
+		sample := &writev2.Sample{
+			// convert ns to ms
+			Timestamp: convertTimeStamp(pt.Timestamp()),
+		}
+		switch pt.ValueType() {
+		case pmetric.NumberDataPointValueTypeInt:
+			sample.Value = float64(pt.IntValue())
+		case pmetric.NumberDataPointValueTypeDouble:
+			sample.Value = pt.DoubleValue()
+		}
+		if pt.Flags().NoRecordedValue() {
+			sample.Value = math.Float64frombits(value.StaleNaN)
+		}
+		c.addSample(sample, labels)
+	}
+}