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

exporter/prometheusremotewriteexporter/config.go

@@ -5,6 +5,7 @@ package prometheusremotewriteexporter // import "github.com/open-telemetry/opent
 
 import (
 	"fmt"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheusremotewrite"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/confighttp"
@@ -52,6 +53,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
+	RemoteWriteProtoMsg prometheusremotewrite.RemoteWriteProtoMsg `mapstructure:"protobuf_message,omitempty"`
 }
 
 type CreatedMetric struct {

exporter/prometheusremotewriteexporter/exporter.go

@@ -19,6 +19,7 @@ import (
 	"github.com/gogo/protobuf/proto"
 	"github.com/golang/snappy"
 	"github.com/prometheus/prometheus/prompb"
+	writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configretry"
@@ -123,6 +124,7 @@ func newPRWExporter(cfg *Config, set exporter.Settings) (*prwExporter, error) {
 			ExportCreatedMetric: cfg.CreatedMetric.Enabled,
 			AddMetricSuffixes:   cfg.AddMetricSuffixes,
 			SendMetadata:        cfg.SendMetadata,
+			RemoteWriteProtoMsg: cfg.RemoteWriteProtoMsg,
 		},
 		telemetry:            prwTelemetry,
 		batchTimeSeriesState: newBatchTimeSericesState(),
@@ -172,22 +174,43 @@ 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
 
-		tsMap, err := prometheusremotewrite.FromMetrics(md, prwe.exporterSettings)
-		if err != nil {
-			prwe.telemetry.recordTranslationFailure(ctx)
+		switch prwe.exporterSettings.RemoteWriteProtoMsg {
+		case prometheusremotewrite.RemoteWriteProtoMsgV1:
+			// 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)
+			}
+			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)))
-		}
+			// Call export even if a conversion error, since there may be points that were successfully converted.
+			return prwe.handleExport(ctx, tsMap, m)
+		case prometheusremotewrite.RemoteWriteProtoMsgV2:
+			// RW2 case
+			tsMapv2, symbolsTable, err = prometheusremotewrite.FromMetricsV2(md, prwe.exporterSettings)
 
-		prwe.telemetry.recordTranslatedTimeSeries(ctx, len(tsMap))
+			prwe.telemetry.recordTranslatedTimeSeries(ctx, len(tsMapv2))
 
-		var m []*prompb.MetricMetadata
-		if prwe.exporterSettings.SendMetadata {
-			m = prometheusremotewrite.OtelMetricsToMetadata(md, prwe.exporterSettings.AddMetricSuffixes)
+			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(tsMapv2)))
+			// Call export even if a conversion error, since there may be points that were successfully converted.
+			return prwe.handleExportV2(ctx, symbolsTable, tsMapv2)
+		default:
+			return errors.New("unexpected rw protobuf message")
 		}
-
-		// Call export even if a conversion error, since there may be points that were successfully converted.
-		return prwe.handleExport(ctx, tsMap, m)
 	}
 }
 
@@ -208,7 +231,6 @@ func (prwe *prwExporter) handleExport(ctx context.Context, tsMap map[string]*pro
 	if len(tsMap) == 0 {
 		return nil
 	}
-
 	// 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 {
@@ -256,7 +278,12 @@ func (prwe *prwExporter) export(ctx context.Context, requests []*prompb.WriteReq
 					if !ok {
 						return
 					}
-					if errExecute := prwe.execute(ctx, request); errExecute != nil {
+					// Uses proto.Marshal to convert the WriteRequest into bytes array
+					data, errMarshal := proto.Marshal(request)
+					if errMarshal != nil {
+						errs = multierr.Append(errs, consumererror.NewPermanent(errMarshal))
+					}
+					if errExecute := prwe.execute(ctx, data); errExecute != nil {
 						mu.Lock()
 						errs = multierr.Append(errs, consumererror.NewPermanent(errExecute))
 						mu.Unlock()
@@ -270,12 +297,7 @@ func (prwe *prwExporter) export(ctx context.Context, requests []*prompb.WriteReq
 	return errs
 }
 
-func (prwe *prwExporter) execute(ctx context.Context, writeReq *prompb.WriteRequest) error {
-	// Uses proto.Marshal to convert the WriteRequest into bytes array
-	data, errMarshal := proto.Marshal(writeReq)
-	if errMarshal != nil {
-		return consumererror.NewPermanent(errMarshal)
-	}
+func (prwe *prwExporter) execute(ctx context.Context, data []byte) error {
 	// If we don't pass a buffer large enough, Snappy Encode function will not use it and instead will allocate a new buffer.
 	// Therefore we always let Snappy decide the size of the buffer.
 	compressedData := snappy.Encode(nil, data)
@@ -300,10 +322,17 @@ 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)
 
+		switch prwe.exporterSettings.RemoteWriteProtoMsg {
+		case prometheusremotewrite.RemoteWriteProtoMsgV1:
+			req.Header.Set("Content-Type", "application/x-protobuf")
+			req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
+		case prometheusremotewrite.RemoteWriteProtoMsgV2:
+			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_test.go

@@ -1169,8 +1169,8 @@ func TestRetries(t *testing.T) {
 					Enabled: true,
 				},
 			}
-
-			err = exporter.execute(tt.ctx, &prompb.WriteRequest{})
+			data, _ := proto.Marshal(&prompb.WriteRequest{})
+			err = exporter.execute(tt.ctx, data)
 			tt.assertError(t, err)
 			tt.assertErrorType(t, err)
 			assert.Equal(t, tt.expectedAttempts, totalAttempts)

exporter/prometheusremotewriteexporter/exporter_v2.go

@@ -0,0 +1,95 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package prometheusremotewriteexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusremotewriteexporter"
+
+import (
+	"context"
+	"math"
+	"sync"
+
+	"github.com/gogo/protobuf/proto"
+	writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
+	"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
+					}
+
+					// Uses proto.Marshal to convert the WriteRequest into bytes array
+					data, errMarshal := proto.Marshal(request)
+					if errMarshal != nil {
+						errs = multierr.Append(errs, consumererror.NewPermanent(errMarshal))
+					}
+
+					if errExecute := prwe.execute(ctx, data); errExecute != nil {
+						mu.Lock()
+						errs = multierr.Append(errs, consumererror.NewPermanent(errExecute))
+						mu.Unlock()
+					}
+				}
+			}
+		}()
+	}
+	wg.Wait()
+
+	return errs
+}
+
+func (prwe *prwExporter) handleExportV2(ctx context.Context, symbolsTable writev2.SymbolsTable, tsMap map[string]*writev2.TimeSeries) error {
+	// There are no metrics to export, so return.
+	if len(tsMap) == 0 {
+		return nil
+	}
+
+	// TODO implement batching
+	requests := make([]*writev2.Request, 0)
+	tsArray := make([]writev2.TimeSeries, 0, len(tsMap))
+	for _, v := range tsMap {
+		tsArray = append(tsArray, *v)
+	}
+
+	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)
+}