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

.chloggen/jm-prom-rw-exporter-add-support-for-rw2.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: prometheusremotewriteexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `exporter.prometheusremotewritexporter.enableSendingRW2` feature gate and configure the exporter to send Prometheus remote write 2.0 version.
+
+# 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: PRW 2.0 support for the exporter is still under development.
+
+# 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, api]

exporter/prometheusremotewriteexporter/README.md

@@ -50,7 +50,7 @@ The following settings can be optionally configured:
   - *Note the following headers cannot be changed: `Content-Encoding`, `Content-Type`, `X-Prometheus-Remote-Write-Version`, and `User-Agent`.*
 - `namespace`: prefix attached to each exported metric name.
 - `add_metric_suffixes`: If set to false, type and unit suffixes will not be added to metrics. Default: true.
-- `send_metadata`: If set to true, prometheus metadata will be generated and sent. Default: false.
+- `send_metadata`: If set to true, prometheus metadata will be generated and sent. Default: false. This option is ignored when using PRW 2.0, which already includes metadata.
 - `remote_write_queue`: fine tuning for queueing and sending of the outgoing remote writes.
   - `enabled`: enable the sending queue (default: `true`)
   - `queue_size`: number of OTLP metrics that can be queued. Ignored if `enabled` is `false` (default: `10000`)
@@ -66,6 +66,11 @@ The following settings can be optionally configured:
 - `max_batch_size_bytes` (default = `3000000` -> `~2.861 mb`): Maximum size of a batch of
   samples to be sent to the remote write endpoint. If the batch size is larger
   than this value, it will be split into multiple batches.
+- `protobuf_message` (default = `prometheus.WriteRequest`): 
+  - Protobuf message to use when writing to the remote write endpoint. This option is ignored unless the `exporter.prometheusremotewritexporter.enableSendingRW2` feature gate is enabled.
+  - `prometheus.WriteRequest` is the message used in [Remote Write 1.0](https://prometheus.io/docs/specs/remote_write_spec/).
+  - `io.prometheus.write.v2.Request` is the message used in [Remote Write 2.0](https://prometheus.io/docs/specs/remote_write_spec_2_0/). It is more efficient, always includes metadata, and adds support for the created timestamp and native histograms. Your remote storage provider must support PRW 2.0 to be able to use this message. PRW 2.0 support is currently **In Development**, and is only partially implemented.
+
 
 Example:
 

exporter/prometheusremotewriteexporter/config.go

@@ -6,6 +6,7 @@ package prometheusremotewriteexporter // import "github.com/open-telemetry/opent
 import (
 	"fmt"
 
+	"github.com/prometheus/prometheus/config"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configretry"
@@ -52,6 +53,9 @@ type Config struct {
 
 	// SendMetadata controls whether prometheus metadata will be generated and sent
 	SendMetadata bool `mapstructure:"send_metadata"`
+
+	// RemoteWriteProtoMsg controls whether prometheus remote write v1 or v2 is sent
+	RemoteWriteProtoMsg config.RemoteWriteProtoMsg `mapstructure:"protobuf_message,omitempty"`
 }
 
 type CreatedMetric struct {
@@ -115,5 +119,10 @@ func (cfg *Config) Validate() error {
 		cfg.MaxBatchSizeBytes = 3000000
 	}
 
+	err := cfg.RemoteWriteProtoMsg.Validate()
+	if err != nil {
+		return err
+	}
+
 	return nil
 }

exporter/prometheusremotewriteexporter/config_test.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
+	"github.com/prometheus/prometheus/config"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
@@ -79,7 +80,8 @@ func TestLoadConfig(t *testing.T) {
 				TargetInfo: &TargetInfo{
 					Enabled: true,
 				},
-				CreatedMetric: &CreatedMetric{Enabled: true},
+				CreatedMetric:       &CreatedMetric{Enabled: true},
+				RemoteWriteProtoMsg: config.RemoteWriteProtoMsgV1,
 			},
 		},
 		{

exporter/prometheusremotewriteexporter/exporter.go

@@ -18,6 +18,7 @@ import (
 	"github.com/cenkalti/backoff/v4"
 	"github.com/gogo/protobuf/proto"
 	"github.com/golang/snappy"
+	"github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/prompb"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/confighttp"
@@ -85,6 +86,7 @@ type prwExporter struct {
 	exporterSettings     prometheusremotewrite.Settings
 	telemetry            prwTelemetry
 	batchTimeSeriesState batchTimeSeriesState
+	RemoteWriteProtoMsg  config.RemoteWriteProtoMsg
 }
 
 func newPRWTelemetry(set exporter.Settings) (prwTelemetry, error) {
@@ -118,19 +120,25 @@ func newPRWExporter(cfg *Config, set exporter.Settings) (*prwExporter, error) {
 		return nil, err
 	}
 
+	err = config.RemoteWriteProtoMsg.Validate(cfg.RemoteWriteProtoMsg)
+	if err != nil {
+		return nil, err
+	}
+
 	userAgentHeader := fmt.Sprintf("%s/%s", strings.ReplaceAll(strings.ToLower(set.BuildInfo.Description), " ", "-"), set.BuildInfo.Version)
 
 	prwe := &prwExporter{
-		endpointURL:       endpointURL,
-		wg:                new(sync.WaitGroup),
-		closeChan:         make(chan struct{}),
-		userAgentHeader:   userAgentHeader,
-		maxBatchSizeBytes: cfg.MaxBatchSizeBytes,
-		concurrency:       cfg.RemoteWriteQueue.NumConsumers,
-		clientSettings:    &cfg.ClientConfig,
-		settings:          set.TelemetrySettings,
-		retrySettings:     cfg.BackOffConfig,
-		retryOnHTTP429:    retryOn429FeatureGate.IsEnabled(),
+		endpointURL:         endpointURL,
+		wg:                  new(sync.WaitGroup),
+		closeChan:           make(chan struct{}),
+		userAgentHeader:     userAgentHeader,
+		maxBatchSizeBytes:   cfg.MaxBatchSizeBytes,
+		concurrency:         cfg.RemoteWriteQueue.NumConsumers,
+		clientSettings:      &cfg.ClientConfig,
+		settings:            set.TelemetrySettings,
+		retrySettings:       cfg.BackOffConfig,
+		retryOnHTTP429:      retryOn429FeatureGate.IsEnabled(),
+		RemoteWriteProtoMsg: cfg.RemoteWriteProtoMsg,
 		exporterSettings: prometheusremotewrite.Settings{
 			Namespace:           cfg.Namespace,
 			ExternalLabels:      sanitizedLabels,
@@ -143,6 +151,8 @@ func newPRWExporter(cfg *Config, set exporter.Settings) (*prwExporter, error) {
 		batchTimeSeriesState: newBatchTimeSericesState(),
 	}
 
+	prwe.settings.Logger.Info("Running with prometheus remote write proto message", zap.Any("ProtoMsg", cfg.RemoteWriteProtoMsg))
+
 	prwe.wal = newWAL(cfg.WAL, prwe.export)
 	return prwe, nil
 }
@@ -176,6 +186,23 @@ func (prwe *prwExporter) Shutdown(context.Context) error {
 	return err
 }
 
+func (prwe *prwExporter) pushMetricsV1(ctx context.Context, md pmetric.Metrics) error {
+	tsMap, err := prometheusremotewrite.FromMetrics(md, prwe.exporterSettings)
+
+	prwe.telemetry.recordTranslatedTimeSeries(ctx, len(tsMap))
+
+	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(tsMap)))
+	}
+	// Call export even if a conversion error, since there may be points that were successfully converted.
+	return prwe.handleExport(ctx, tsMap, m)
+}
+
 // PushMetrics converts metrics to Prometheus remote write TimeSeries and send to remote endpoint. It maintain a map of
 // TimeSeries, validates and handles each individual metric, adding the converted TimeSeries to the map, and finally
 // exports the map.
@@ -188,21 +215,23 @@ func (prwe *prwExporter) PushMetrics(ctx context.Context, md pmetric.Metrics) er
 		return errors.New("shutdown has been called")
 	default:
 
-		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)))
+		// If feature flag not enabled support only RW1
+		if !enableSendingRW2FeatureGate.IsEnabled() {
+			return prwe.pushMetricsV1(ctx, md)
 		}
 
-		prwe.telemetry.recordTranslatedTimeSeries(ctx, len(tsMap))
+		// If feature flag enabled check if we want to send RW1 or RW2
+		switch prwe.RemoteWriteProtoMsg {
+		case config.RemoteWriteProtoMsgV1:
+			// Rw1 case
+			return prwe.pushMetricsV1(ctx, md)
+		case config.RemoteWriteProtoMsgV2:
+			// RW2 case
+			return prwe.pushMetricsV2(ctx, md)
 
-		var m []*prompb.MetricMetadata
-		if prwe.exporterSettings.SendMetadata {
-			m = prometheusremotewrite.OtelMetricsToMetadata(md, prwe.exporterSettings.AddMetricSuffixes)
+		default:
+			return fmt.Errorf("unsupported remote-write protobuf message: %v", prwe.RemoteWriteProtoMsg)
 		}
-
-		// Call export even if a conversion error, since there may be points that were successfully converted.
-		return prwe.handleExport(ctx, tsMap, m)
 	}
 }
 
@@ -223,7 +252,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 {
@@ -271,7 +299,19 @@ func (prwe *prwExporter) export(ctx context.Context, requests []*prompb.WriteReq
 					if !ok {
 						return
 					}
-					if errExecute := prwe.execute(ctx, request); errExecute != nil {
+
+					buf := bufferPool.Get().(*buffer)
+					buf.protobuf.Reset()
+					defer bufferPool.Put(buf)
+
+					// Uses proto.Marshal to convert the WriteRequest into bytes array
+					errMarshal := buf.protobuf.Marshal(request)
+					if errMarshal != nil {
+						errs = multierr.Append(errs, consumererror.NewPermanent(errMarshal))
+						return
+					}
+
+					if errExecute := prwe.execute(ctx, buf); errExecute != nil {
 						mu.Lock()
 						errs = multierr.Append(errs, consumererror.NewPermanent(errExecute))
 						mu.Unlock()
@@ -285,16 +325,7 @@ func (prwe *prwExporter) export(ctx context.Context, requests []*prompb.WriteReq
 	return errs
 }
 
-func (prwe *prwExporter) execute(ctx context.Context, writeReq *prompb.WriteRequest) error {
-	buf := bufferPool.Get().(*buffer)
-	buf.protobuf.Reset()
-	defer bufferPool.Put(buf)
-
-	// Uses proto.Marshal to convert the WriteRequest into bytes array
-	errMarshal := buf.protobuf.Marshal(writeReq)
-	if errMarshal != nil {
-		return consumererror.NewPermanent(errMarshal)
-	}
+func (prwe *prwExporter) execute(ctx context.Context, buf *buffer) error {
 	// If we don't pass a buffer large enough, Snappy Encode function will not use it and instead will allocate a new buffer.
 	// Manually grow the buffer to make sure Snappy uses it and we can re-use it afterwards.
 	maxCompressedLen := snappy.MaxEncodedLen(len(buf.protobuf.Bytes()))
@@ -327,10 +358,23 @@ 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 feature flag not enabled support only RW1
+		if !enableSendingRW2FeatureGate.IsEnabled() {
+			req.Header.Set("Content-Type", "application/x-protobuf")
+			req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
+		} else {
+			switch prwe.RemoteWriteProtoMsg {
+			case config.RemoteWriteProtoMsgV1:
+				req.Header.Set("Content-Type", "application/x-protobuf")
+				req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
+			case config.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