[exporter/prometheusremotewrite] feat: prom rw exporter add support for rw2

exporter/prometheusremotewriteexporter/config.go

@@ -52,6 +52,9 @@ type Config struct {
 
 	// SendMetadata controls whether prometheus metadata will be generated and sent
 	SendMetadata bool `mapstructure:"send_metadata"`
+
+	// SendRW2 controls whether prometheus remote write v1 or v2 is sent
+	SendRW2 bool `mapstructure:"send_remote_write_v2"`
 }
 
 type CreatedMetric struct {

exporter/prometheusremotewriteexporter/exporter.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
 	"io"
 	"math"
 	"net/http"
@@ -123,6 +124,7 @@ func newPRWExporter(cfg *Config, set exporter.Settings) (*prwExporter, error) {
 			ExportCreatedMetric: cfg.CreatedMetric.Enabled,
 			AddMetricSuffixes:   cfg.AddMetricSuffixes,
 			SendMetadata:        cfg.SendMetadata,
+			SendRW2:             cfg.SendRW2,
 		},
 		telemetry:            prwTelemetry,
 		batchTimeSeriesState: newBatchTimeSericesState(),
@@ -172,22 +174,36 @@ func (prwe *prwExporter) PushMetrics(ctx context.Context, md pmetric.Metrics) er
 	case <-prwe.closeChan:
 		return errors.New("shutdown has been called")
 	default:
+		var tsMap map[string]*prompb.TimeSeries
+		var tsMapv2 map[string]*writev2.TimeSeries
+		var symbolsTable writev2.SymbolsTable
+		var m []*prompb.MetricMetadata
+		var err error
+		if !prwe.exporterSettings.SendRW2 {
+			// RW1 case
+			tsMap, err = prometheusremotewrite.FromMetrics(md, prwe.exporterSettings)
+
+			prwe.telemetry.recordTranslatedTimeSeries(ctx, len(tsMap))
+
+			if prwe.exporterSettings.SendMetadata {
+				m = prometheusremotewrite.OtelMetricsToMetadata(md, prwe.exporterSettings.AddMetricSuffixes)
+			}
+
+		} else {
+			// RW2 case
+			tsMapv2, symbolsTable, err = prometheusremotewrite.FromMetricsV2(md, prwe.exporterSettings)
+
+			prwe.telemetry.recordTranslatedTimeSeries(ctx, len(tsMapv2))
+		}
 
-		tsMap, err := prometheusremotewrite.FromMetrics(md, prwe.exporterSettings)
 		if err != nil {
 			prwe.telemetry.recordTranslationFailure(ctx)
 			prwe.settings.Logger.Debug("failed to translate metrics, exporting remaining metrics", zap.Error(err), zap.Int("translated", len(tsMap)))
 		}
 
-		prwe.telemetry.recordTranslatedTimeSeries(ctx, len(tsMap))
-
-		var m []*prompb.MetricMetadata
-		if prwe.exporterSettings.SendMetadata {
-			m = prometheusremotewrite.OtelMetricsToMetadata(md, prwe.exporterSettings.AddMetricSuffixes)
-		}
-
 		// Call export even if a conversion error, since there may be points that were successfully converted.
-		return prwe.handleExport(ctx, tsMap, m)
+		return prwe.handleExport(ctx, tsMap, m, symbolsTable, tsMapv2)
+
 	}
 }
 
@@ -203,28 +219,53 @@ func validateAndSanitizeExternalLabels(cfg *Config) (map[string]string, error) {
 	return sanitizedLabels, nil
 }
 
-func (prwe *prwExporter) handleExport(ctx context.Context, tsMap map[string]*prompb.TimeSeries, m []*prompb.MetricMetadata) error {
+func (prwe *prwExporter) handleExport(ctx context.Context, tsMap map[string]*prompb.TimeSeries, m []*prompb.MetricMetadata, symbolsTable writev2.SymbolsTable, tsMapv2 map[string]*writev2.TimeSeries) error {
 	// There are no metrics to export, so return.
-	if len(tsMap) == 0 {
+	if len(tsMap) == 0 || len(tsMapv2) == 0 {
 		return nil
 	}
+	if len(tsMapv2) == 0 {
+		// v1 case
+		// Calls the helper function to convert and batch the TsMap to the desired format
+		requests, err := batchTimeSeries(tsMap, prwe.maxBatchSizeBytes, m, &prwe.batchTimeSeriesState)
+		if err != nil {
+			return err
+		}
+		if !prwe.walEnabled() {
+			// Perform a direct export otherwise.
+			return prwe.export(ctx, requests)
+		}
 
-	// Calls the helper function to convert and batch the TsMap to the desired format
-	requests, err := batchTimeSeries(tsMap, prwe.maxBatchSizeBytes, m, &prwe.batchTimeSeriesState)
-	if err != nil {
-		return err
-	}
-	if !prwe.walEnabled() {
-		// Perform a direct export otherwise.
-		return prwe.export(ctx, requests)
+		// Otherwise the WAL is enabled, and just persist the requests to the WAL
+		// and they'll be exported in another goroutine to the RemoteWrite endpoint.
+		if err = prwe.wal.persistToWAL(requests); err != nil {
+			return consumererror.NewPermanent(err)
+		}
+		return nil
 	}
+	// v2 case
 
-	// Otherwise the WAL is enabled, and just persist the requests to the WAL
-	// and they'll be exported in another goroutine to the RemoteWrite endpoint.
-	if err = prwe.wal.persistToWAL(requests); err != nil {
-		return consumererror.NewPermanent(err)
+	// TODO implement batching
+	requests := make([]*writev2.Request, 0)
+	tsArray := make([]writev2.TimeSeries, 0, len(tsMap))
+	for _, v := range tsMapv2 {
+		tsArray = append(tsArray, *v)
 	}
-	return nil
+
+	requests = append(requests, &writev2.Request{
+		// TODO sort
+		// Prometheus requires time series to be sorted by Timestamp to avoid out of order problems.
+		// See:
+		// * https://github.com/open-telemetry/wg-prometheus/issues/10
+		// * https://github.com/open-telemetry/opentelemetry-collector/issues/2315
+		//Timeseries: orderBySampleTimestamp(tsArray),
+		Timeseries: tsArray,
+		Symbols:    symbolsTable.Symbols(),
+	})
+
+	// TODO implement WAl support, can be done after #15277 is fixed
+
+	return prwe.exportV2(ctx, requests)
 }
 
 // export sends a Snappy-compressed WriteRequest containing TimeSeries to a remote write endpoint in order
@@ -256,7 +297,7 @@ func (prwe *prwExporter) export(ctx context.Context, requests []*prompb.WriteReq
 					if !ok {
 						return
 					}
-					if errExecute := prwe.execute(ctx, request); errExecute != nil {
+					if errExecute := prwe.execute(ctx, request, nil, false); errExecute != nil {
 						mu.Lock()
 						errs = multierr.Append(errs, consumererror.NewPermanent(errExecute))
 						mu.Unlock()
@@ -270,9 +311,15 @@ func (prwe *prwExporter) export(ctx context.Context, requests []*prompb.WriteReq
 	return errs
 }
 
-func (prwe *prwExporter) execute(ctx context.Context, writeReq *prompb.WriteRequest) error {
+func (prwe *prwExporter) execute(ctx context.Context, writeReq *prompb.WriteRequest, writeReqv2 *writev2.Request, v2 bool) error {
 	// Uses proto.Marshal to convert the WriteRequest into bytes array
-	data, errMarshal := proto.Marshal(writeReq)
+	var data []byte
+	var errMarshal error
+	if !v2 {
+		data, errMarshal = proto.Marshal(writeReq)
+	} else {
+		data, errMarshal = proto.Marshal(writeReqv2)
+	}
 	if errMarshal != nil {
 		return consumererror.NewPermanent(errMarshal)
 	}
@@ -300,10 +347,16 @@ func (prwe *prwExporter) execute(ctx context.Context, writeReq *prompb.WriteRequ
 		// Add necessary headers specified by:
 		// https://cortexmetrics.io/docs/apis/#remote-api
 		req.Header.Add("Content-Encoding", "snappy")
-		req.Header.Set("Content-Type", "application/x-protobuf")
-		req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
 		req.Header.Set("User-Agent", prwe.userAgentHeader)
 
+		if !v2 {
+			req.Header.Set("Content-Type", "application/x-protobuf")
+			req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
+		} else {
+			req.Header.Set("Content-Type", "application/x-protobuf;proto=io.prometheus.write.v2.Request")
+			req.Header.Set("X-Prometheus-Remote-Write-Version", "2.0.0")
+		}
+
 		resp, err := prwe.client.Do(req)
 		if err != nil {
 			return err

exporter/prometheusremotewriteexporter/exporter_v2.go

@@ -0,0 +1,58 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package prometheusremotewriteexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusremotewriteexporter"
+
+import (
+	"context"
+	writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
+	"math"
+	"sync"
+
+	"go.opentelemetry.io/collector/consumer/consumererror"
+	"go.uber.org/multierr"
+)
+
+// TODO update comment
+// export sends a Snappy-compressed WriteRequest containing TimeSeries to a remote write endpoint in order
+func (prwe *prwExporter) exportV2(ctx context.Context, requests []*writev2.Request) error {
+	input := make(chan *writev2.Request, len(requests))
+	for _, request := range requests {
+		input <- request
+	}
+	close(input)
+
+	var wg sync.WaitGroup
+
+	concurrencyLimit := int(math.Min(float64(prwe.concurrency), float64(len(requests))))
+	wg.Add(concurrencyLimit) // used to wait for workers to be finished
+
+	var mu sync.Mutex
+	var errs error
+	// Run concurrencyLimit of workers until there
+	// is no more requests to execute in the input channel.
+	for i := 0; i < concurrencyLimit; i++ {
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-ctx.Done(): // Check firstly to ensure that the context wasn't cancelled.
+					return
+
+				case request, ok := <-input:
+					if !ok {
+						return
+					}
+					if errExecute := prwe.execute(ctx, nil, request, true); errExecute != nil {
+						mu.Lock()
+						errs = multierr.Append(errs, consumererror.NewPermanent(errExecute))
+						mu.Unlock()
+					}
+				}
+			}
+		}()
+	}
+	wg.Wait()
+
+	return errs
+}