[exporter/prometheusremotewrite] feat: prom rw exporter add support for rw2 (#35888)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Draft PR for adding rw2 support in the prometheus remote write exporter.
Very much a draft, not full implementation of the spec with a lot of
code duplication and no tests WIP.

TODO:

- [x] changelog entry
- [x] feature flag on top of config options
- [x] rip out batching for now
- [x] enum instead of bool for RW2
- [x] update exporter readme
- [x] Validate supported enum value set
- [x] Check for TODOs 
- [x] unit tests
- [x] going over the spec and making changes to be rw2 compliant ->
follow up PR
- [x] try to reduce duplicate code

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue #33661
Fixes

<!--Describe what testing was performed and which tests were added.-->
#### Testing

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Signed-off-by: Juraj Michalek <[email protected]>
Co-authored-by: David Ashpole <[email protected]>
Co-authored-by: Arthur Silva Sens <[email protected]>
Co-authored-by: Bartlomiej Plotka <[email protected]>

Author: 4 people

.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 configuration to 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: WARNING! PRW 2.0 support for the exporter is still under development and not ready for usage.
+
+# 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

@@ -51,7 +51,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 always 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`)
@@ -64,6 +64,11 @@ The following settings can be optionally configured:
   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.
 - `max_batch_request_parallelism` (default = `5`): Maximum parallelism allowed for a single request bigger than `max_batch_size_bytes`.
+- `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, thus, not ready for usage.
+
 
 Example:
 

exporter/prometheusremotewriteexporter/config.go

@@ -5,7 +5,9 @@ package prometheusremotewriteexporter // import "github.com/open-telemetry/opent
 
 import (
 	"errors"
+	"fmt"
 
+	"github.com/prometheus/prometheus/config"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configretry"
@@ -50,8 +52,11 @@ type Config struct {
 	// AddMetricSuffixes controls whether unit and type suffixes are added to metrics on export
 	AddMetricSuffixes bool `mapstructure:"add_metric_suffixes"`
 
-	// SendMetadata controls whether prometheus metadata will be generated and sent
+	// SendMetadata controls whether prometheus metadata will be generated and sent, this option is ignored when using PRW 2.0, which always includes metadata.
 	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 {
@@ -118,5 +123,14 @@ func (cfg *Config) Validate() error {
 		return errors.New("compression type must be snappy")
 	}
 
+	err := cfg.RemoteWriteProtoMsg.Validate()
+	if err != nil {
+		return err
+	}
+
+	if !enableSendingRW2FeatureGate.IsEnabled() && cfg.RemoteWriteProtoMsg == config.RemoteWriteProtoMsgV2 {
+		return fmt.Errorf("remote write v2 is only supported with the feature gate %s", enableSendingRW2FeatureGate.ID())
+	}
+
 	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"
@@ -81,6 +82,7 @@ func TestLoadConfig(t *testing.T) {
 				TargetInfo: &TargetInfo{
 					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"
@@ -80,20 +81,21 @@ var bufferPool = sync.Pool{
 
 // prwExporter converts OTLP metrics to Prometheus remote write TimeSeries and sends them to a remote endpoint.
 type prwExporter struct {
-	endpointURL       *url.URL
-	client            *http.Client
-	wg                *sync.WaitGroup
-	closeChan         chan struct{}
-	concurrency       int
-	userAgentHeader   string
-	maxBatchSizeBytes int
-	clientSettings    *confighttp.ClientConfig
-	settings          component.TelemetrySettings
-	retrySettings     configretry.BackOffConfig
-	retryOnHTTP429    bool
-	wal               *prweWAL
-	exporterSettings  prometheusremotewrite.Settings
-	telemetry         prwTelemetry
+	endpointURL         *url.URL
+	client              *http.Client
+	wg                  *sync.WaitGroup
+	closeChan           chan struct{}
+	concurrency         int
+	userAgentHeader     string
+	maxBatchSizeBytes   int
+	clientSettings      *confighttp.ClientConfig
+	settings            component.TelemetrySettings
+	retrySettings       configretry.BackOffConfig
+	retryOnHTTP429      bool
+	wal                 *prweWAL
+	exporterSettings    prometheusremotewrite.Settings
+	telemetry           prwTelemetry
+	RemoteWriteProtoMsg config.RemoteWriteProtoMsg
 
 	// When concurrency is enabled, concurrent goroutines would potentially
 	// fight over the same batchState object. To avoid this, we use a pool
@@ -133,6 +135,10 @@ func newPRWExporter(cfg *Config, set exporter.Settings) (*prwExporter, error) {
 		return nil, err
 	}
 
+	if err := config.RemoteWriteProtoMsg.Validate(cfg.RemoteWriteProtoMsg); err != nil {
+		return nil, err
+	}
+
 	userAgentHeader := fmt.Sprintf("%s/%s", strings.ReplaceAll(strings.ToLower(set.BuildInfo.Description), " ", "-"), set.BuildInfo.Version)
 
 	concurrency := 5
@@ -147,16 +153,17 @@ func newPRWExporter(cfg *Config, set exporter.Settings) (*prwExporter, error) {
 	telemetry.setNumberConsumer(context.Background(), int64(concurrency))
 
 	prwe := &prwExporter{
-		endpointURL:       endpointURL,
-		wg:                new(sync.WaitGroup),
-		closeChan:         make(chan struct{}),
-		userAgentHeader:   userAgentHeader,
-		maxBatchSizeBytes: cfg.MaxBatchSizeBytes,
-		concurrency:       concurrency,
-		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,
@@ -168,6 +175,8 @@ func newPRWExporter(cfg *Config, set exporter.Settings) (*prwExporter, error) {
 		batchStatePool: sync.Pool{New: func() any { return newBatchTimeServicesState() }},
 	}
 
+	prwe.settings.Logger.Info("starting prometheus remote write exporter", zap.Any("ProtoMsg", cfg.RemoteWriteProtoMsg))
+
 	prwe.wal = newWAL(cfg.WAL, prwe.export)
 	return prwe, nil
 }
@@ -201,6 +210,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.
@@ -213,21 +239,21 @@ 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 was enabled check if we want to send RW1 or RW2.
+		switch prwe.RemoteWriteProtoMsg {
+		case config.RemoteWriteProtoMsgV1:
+			return prwe.pushMetricsV1(ctx, md)
+		case config.RemoteWriteProtoMsgV2:
+			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)
 	}
 }
 
@@ -298,7 +324,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)
+
+					if errMarshal := buf.protobuf.Marshal(request); errMarshal != nil {
+						mu.Lock()
+						errs = multierr.Append(errs, consumererror.NewPermanent(errMarshal))
+						mu.Unlock()
+						return
+					}
+
+					if errExecute := prwe.execute(ctx, buf); errExecute != nil {
 						mu.Lock()
 						errs = multierr.Append(errs, consumererror.NewPermanent(errExecute))
 						mu.Unlock()
@@ -312,16 +350,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()))
@@ -354,10 +383,20 @@ 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 {
+		// If feature flag not enabled support only RW1
+		case !enableSendingRW2FeatureGate.IsEnabled(), prwe.RemoteWriteProtoMsg == config.RemoteWriteProtoMsgV1:
+			req.Header.Set("Content-Type", "application/x-protobuf")
+			req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
+		case prwe.RemoteWriteProtoMsg == 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")
+		default:
+			return fmt.Errorf("unsupported remote-write protobuf message: %v (should be validated earlier)", prwe.RemoteWriteProtoMsg)
+		}
+
 		resp, err := prwe.client.Do(req)
 		prwe.telemetry.recordRemoteWriteSentBatch(ctx)
 		if err != nil {

exporter/prometheusremotewriteexporter/exporter_concurrency_test.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/gogo/protobuf/proto"
 	"github.com/golang/snappy"
+	"github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/prompb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -109,7 +110,8 @@ func Test_PushMetricsConcurrent(t *testing.T) {
 		TargetInfo: &TargetInfo{
 			Enabled: true,
 		},
-		BackOffConfig: retrySettings,
+		BackOffConfig:       retrySettings,
+		RemoteWriteProtoMsg: config.RemoteWriteProtoMsgV1,
 	}
 
 	assert.NotNil(t, cfg)