[PrometheusRemoteWriter] Add support of MetadataKeys in remote write outbound requests (#48767)

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

Adds the support of IncludeMetadataKeys in the prometehus remote write
queue.
When set, the listed client metadata keys are read from the request
context and forwarded as HTTP headers

Related to
#33137 (comment)

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

With this feature we preserver the metadata from the prometheus exporter
to outer http requests.
The metadata can be useful information to be used to handle this egress
traffic.

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

Added TestIncludeMetadataKeys and
TestIncludeMetadataKeysAbsentWhenNotConfigured in

opentelemetry-collector-contrib/exporter/prometheusremotewriteexporter/exporter_test.go

<!--Describe the documentation added.-->
#### Documentation
Updated README.md file

---------

Signed-off-by: Andreas Gkizas <andreas.gkizas@elastic.co>
Co-authored-by: David Ashpole <dashpole@google.com>

Author: gizas

.chloggen/prw-include-metadata-keys.yaml

@@ -0,0 +1,28 @@
+# 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. receiver/filelog)
+component: exporter/prometheus_remote_write
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `include_metadata_keys` to `remote_write_queue` configuration to forward client metadata as HTTP headers.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [47317]
+
+# (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:
+
+# 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]
+

exporter/prometheusremotewriteexporter/README.md

@@ -52,6 +52,9 @@ The following settings can be optionally configured:
 - `add_metric_suffixes`: If set to false, type and unit suffixes will not be added to metrics. Default: true. **Deprecated**: Use `translation_strategy` instead.
 - `translation_strategy`: Controls how OTLP metric and attribute names are translated into Prometheus metric and label names. Options are: `UnderscoreEscapingWithSuffixes` (default), `UnderscoreEscapingWithoutSuffixes`, `NoUTF8EscapingWithSuffixes`, and `NoTranslation`. When set, this takes precedence over `add_metric_suffixes`.
 - `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.
+- `include_metadata_keys`: list of client metadata keys whose values are forwarded as HTTP headers on every outbound remote write request.
+  - **Note**: Keys that collide with headers required by the remote write protocol (`Content-Encoding`, `Content-Type`, `User-Agent`, `X-Prometheus-Remote-Write-Version`) are rejected: configuration validation fails (case-insensitive) if any such key is listed.
+  - **Note**: When WAL is enabled, then this setting has no effect. The WAL persists only the raw protobuf bytes of each write request; the originating `client.Info` context is not serialized to disk so initial metadata are lost.
 - `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`)

exporter/prometheusremotewriteexporter/config.go

@@ -6,6 +6,7 @@ package prometheusremotewriteexporter // import "github.com/open-telemetry/opent
 import (
 	"errors"
 	"fmt"
+	"net/textproto"
 
 	remoteapi "github.com/prometheus/client_golang/exp/api/remote"
 	"go.opentelemetry.io/collector/component"
@@ -72,6 +73,10 @@ type Config struct {
 
 	// RemoteWriteProtoMsg controls whether prometheus remote write v1 or v2 is sent.
 	RemoteWriteProtoMsg remoteapi.WriteMessageType `mapstructure:"protobuf_message,omitempty"`
+
+	// IncludeMetadataKeys is a list of client metadata keys whose values are
+	// forwarded as HTTP request headers on every remote write call.
+	IncludeMetadataKeys []string `mapstructure:"include_metadata_keys"`
 }
 
 type translationStrategy string
@@ -118,6 +123,16 @@ type RemoteWriteQueue struct {
 
 // TODO(jbd): Add capacity, max_samples_per_send to QueueConfig.
 
+// reservedRemoteWriteHeaders are managed by the remote write protocol itself and
+// must not be overwritten by headers forwarded from client metadata. Keys are
+// stored in canonical MIME form for case-insensitive matching.
+var reservedRemoteWriteHeaders = map[string]struct{}{
+	textproto.CanonicalMIMEHeaderKey("Content-Encoding"):                  {},
+	textproto.CanonicalMIMEHeaderKey("Content-Type"):                      {},
+	textproto.CanonicalMIMEHeaderKey("User-Agent"):                        {},
+	textproto.CanonicalMIMEHeaderKey("X-Prometheus-Remote-Write-Version"): {},
+}
+
 var _ component.Config = (*Config)(nil)
 
 // Validate checks if the exporter configuration is valid
@@ -172,5 +187,11 @@ func (cfg *Config) Validate() error {
 		}
 	}
 
+	for _, key := range cfg.IncludeMetadataKeys {
+		if _, reserved := reservedRemoteWriteHeaders[textproto.CanonicalMIMEHeaderKey(key)]; reserved {
+			return fmt.Errorf("include_metadata_keys entry %q collides with a reserved remote write header", key)
+		}
+	}
+
 	return nil
 }

exporter/prometheusremotewriteexporter/config.schema.yaml

@@ -45,6 +45,11 @@ properties:
     type: object
     additionalProperties:
       type: string
+  include_metadata_keys:
+    description: IncludeMetadataKeys is a list of client metadata keys whose values are forwarded as HTTP request headers on every remote write call.
+    type: array
+    items:
+      type: string
   max_batch_request_parallelism:
     description: maximum amount of parallel requests to do when handling large batch request
     x-pointer: true

exporter/prometheusremotewriteexporter/config_test.go

@@ -158,10 +158,49 @@ func TestLoadConfig(t *testing.T) {
 			id:           component.NewIDWithName(metadata.Type, "v1_no_translation"),
 			errorMessage: "translation strategy NoTranslation requires Prometheus Remote Write 2.0",
 		},
+		{
+			id:           component.NewIDWithName(metadata.Type, "reserved_metadata_keys"),
+			errorMessage: "include_metadata_keys entry \"content-type\" collides with a reserved remote write header",
+		},
 		{
 			id:           component.NewIDWithName(metadata.Type, "negative_max_batch_size_bytes"),
 			errorMessage: "max_batch_size_bytes must be greater than 0",
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "include_metadata_keys"),
+			expected: &Config{
+				MaxBatchSizeBytes:          3000000,
+				MaxBatchRequestParallelism: nil,
+				TimeoutSettings:            exporterhelper.NewDefaultTimeoutConfig(),
+				BackOffConfig: configretry.BackOffConfig{
+					Enabled:             true,
+					InitialInterval:     50 * time.Millisecond,
+					RandomizationFactor: 0.5,
+					Multiplier:          1.5,
+					MaxInterval:         30 * time.Second,
+					MaxElapsedTime:      5 * time.Minute,
+				},
+				RemoteWriteQueue: RemoteWriteQueue{
+					Enabled:      true,
+					QueueSize:    1000,
+					NumConsumers: 5,
+				},
+				IncludeMetadataKeys: []string{"target-id", "x-org-id"},
+				ExternalLabels:      map[string]string{},
+				AddMetricSuffixes:   true,
+				ClientConfig: func() confighttp.ClientConfig {
+					cc := confighttp.NewDefaultClientConfig()
+					cc.Endpoint = "localhost:8888"
+					cc.WriteBufferSize = 512 * 1024
+					cc.Timeout = 5 * time.Second
+					return cc
+				}(),
+				RemoteWriteProtoMsg: remoteapi.WriteV1MessageType,
+				TargetInfo: TargetInfo{
+					Enabled: true,
+				},
+			},
+		},
 	}
 
 	for _, tt := range tests {

exporter/prometheusremotewriteexporter/exporter.go

@@ -21,6 +21,7 @@ import (
 	remoteapi "github.com/prometheus/client_golang/exp/api/remote"
 	"github.com/prometheus/otlptranslator"
 	"github.com/prometheus/prometheus/prompb"
+	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configretry"
@@ -139,6 +140,9 @@ type prwExporter struct {
 	exporterSettings    prometheusremotewrite.Settings
 	telemetry           prwTelemetry
 	RemoteWriteProtoMsg remoteapi.WriteMessageType
+	// includeMetadataKeys lists client metadata keys that are forwarded as
+	// HTTP headers on every outbound remote write request.
+	includeMetadataKeys []string
 
 	// When concurrency is enabled, concurrent goroutines would potentially
 	// fight over the same batchState object. To avoid this, we use a pool
@@ -211,6 +215,7 @@ func newPRWExporter(cfg *Config, set exporter.Settings) (*prwExporter, error) {
 		retrySettings:       cfg.BackOffConfig,
 		retryOnHTTP429:      metadata.ExporterPrometheusremotewritexporterRetryOn429FeatureGate.IsEnabled(),
 		RemoteWriteProtoMsg: cfg.RemoteWriteProtoMsg,
+		includeMetadataKeys: cfg.IncludeMetadataKeys,
 		exporterSettings: prometheusremotewrite.Settings{
 			Namespace:           cfg.Namespace,
 			ExternalLabels:      sanitizedLabels,
@@ -456,6 +461,16 @@ func (prwe *prwExporter) execute(ctx context.Context, buf []byte) error {
 			return http.StatusBadRequest, fmt.Errorf("unsupported remote-write protobuf message: %v (should be validated earlier)", prwe.RemoteWriteProtoMsg)
 		}
 
+		// Forward configured client metadata keys as HTTP headers
+		if len(prwe.includeMetadataKeys) > 0 {
+			md := client.FromContext(ctx).Metadata
+			for _, key := range prwe.includeMetadataKeys {
+				for _, val := range md.Get(key) {
+					req.Header.Add(key, val)
+				}
+			}
+		}
+
 		resp, err := prwe.client.Do(req)
 		prwe.telemetry.recordRemoteWriteSentBatch(ctx)
 		if err != nil {

exporter/prometheusremotewriteexporter/exporter_test.go

@@ -25,6 +25,7 @@ import (
 	writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/confighttp"
@@ -1371,6 +1372,114 @@ func BenchmarkPushMetrics(b *testing.B) {
 	}
 }
 
+// TestIncludeMetadataKeys verifies that keys listed in
+// include_metadata_keys are read from the client metadata
+// context and forwarded as HTTP headers on every outbound remote write request.
+func TestIncludeMetadataKeys(t *testing.T) {
+	// Capture headers received by the mock remote write server.
+	var (
+		mu              sync.Mutex
+		receivedHeaders http.Header
+	)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		mu.Lock()
+		receivedHeaders = r.Header.Clone()
+		mu.Unlock()
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer server.Close()
+
+	clientConfig := confighttp.NewDefaultClientConfig()
+	clientConfig.Endpoint = server.URL
+	cfg := &Config{
+		ClientConfig:     clientConfig,
+		RemoteWriteQueue: RemoteWriteQueue{NumConsumers: 1},
+		IncludeMetadataKeys: []string{
+			"target-id",
+			"target-type",
+		},
+		TargetInfo:          TargetInfo{Enabled: true},
+		BackOffConfig:       configretry.NewDefaultBackOffConfig(),
+		RemoteWriteProtoMsg: remoteapi.WriteV1MessageType,
+	}
+
+	set := exportertest.NewNopSettings(metadata.Type)
+	prwe, err := newPRWExporter(cfg, set)
+	require.NoError(t, err)
+	require.NoError(t, prwe.Start(t.Context(), componenttest.NewNopHost()))
+	t.Cleanup(func() { require.NoError(t, prwe.Shutdown(context.Background())) }) //nolint:usetesting
+
+	// Build a context carrying the metadata keys that should be forwarded.
+	ctx := client.NewContext(t.Context(), client.Info{
+		Metadata: client.NewMetadata(map[string][]string{
+			"target-id":   {"my-project-123"},
+			"target-type": {"server-type"},
+		}),
+	})
+
+	// execute() requires a snappy-encoded protobuf body. An empty WriteRequest
+	// is valid enough for the mock server to accept.
+	emptyBuf, err := (&buffer{}).MarshalAndEncode(&prompb.WriteRequest{})
+	require.NoError(t, err)
+
+	require.NoError(t, prwe.execute(ctx, emptyBuf))
+
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, "my-project-123", receivedHeaders.Get("target-id"),
+		"target-id header should be forwarded from client metadata")
+	assert.Equal(t, "server-type", receivedHeaders.Get("target-type"),
+		"target-type header should be forwarded from client metadata")
+}
+
+// TestIncludeMetadataKeysAbsentWhenNotConfigured verifies that no extra headers
+// are added when include_metadata_keys is empty (the default).
+func TestIncludeMetadataKeysAbsentWhenNotConfigured(t *testing.T) {
+	var (
+		mu              sync.Mutex
+		receivedHeaders http.Header
+	)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		mu.Lock()
+		receivedHeaders = r.Header.Clone()
+		mu.Unlock()
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer server.Close()
+
+	clientConfig := confighttp.NewDefaultClientConfig()
+	clientConfig.Endpoint = server.URL
+	cfg := &Config{
+		ClientConfig:        clientConfig,
+		RemoteWriteQueue:    RemoteWriteQueue{NumConsumers: 1},
+		TargetInfo:          TargetInfo{Enabled: true},
+		BackOffConfig:       configretry.NewDefaultBackOffConfig(),
+		RemoteWriteProtoMsg: remoteapi.WriteV1MessageType,
+	}
+
+	set := exportertest.NewNopSettings(metadata.Type)
+	prwe, err := newPRWExporter(cfg, set)
+	require.NoError(t, err)
+	require.NoError(t, prwe.Start(t.Context(), componenttest.NewNopHost()))
+	t.Cleanup(func() { require.NoError(t, prwe.Shutdown(context.Background())) }) //nolint:usetesting
+
+	ctx := client.NewContext(t.Context(), client.Info{
+		Metadata: client.NewMetadata(map[string][]string{
+			"target-id": {"my-project-123"},
+		}),
+	})
+
+	emptyBuf, err := (&buffer{}).MarshalAndEncode(&prompb.WriteRequest{})
+	require.NoError(t, err)
+
+	require.NoError(t, prwe.execute(ctx, emptyBuf))
+
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Empty(t, receivedHeaders.Get("target-id"),
+		"target-id should not be forwarded when not listed in include_metadata_keys")
+}
+
 func BenchmarkPushMetricsVaryingMetrics(b *testing.B) {
 	benchmarkPushMetrics(b, -1, 1)
 }

exporter/prometheusremotewriteexporter/go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/prometheus/prometheus v0.313.1
 	github.com/stretchr/testify v1.11.1
 	github.com/tidwall/wal v1.2.1
+	go.opentelemetry.io/collector/client v1.63.1-0.20260723141305-52e6bf4aaaba
 	go.opentelemetry.io/collector/component v1.63.1-0.20260723141305-52e6bf4aaaba
 	go.opentelemetry.io/collector/component/componenttest v0.157.1-0.20260723141305-52e6bf4aaaba
 	go.opentelemetry.io/collector/config/confighttp v0.157.1-0.20260723141305-52e6bf4aaaba
@@ -115,7 +116,6 @@ require (
 	github.com/tidwall/pretty v1.2.0 // indirect
 	github.com/tidwall/tinylru v1.1.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
-	go.opentelemetry.io/collector/client v1.63.1-0.20260723141305-52e6bf4aaaba // indirect
 	go.opentelemetry.io/collector/config/configauth v1.63.1-0.20260723141305-52e6bf4aaaba // indirect
 	go.opentelemetry.io/collector/config/configcompression v1.63.1-0.20260723141305-52e6bf4aaaba // indirect
 	go.opentelemetry.io/collector/config/configmiddleware v1.63.1-0.20260723141305-52e6bf4aaaba // indirect

exporter/prometheusremotewriteexporter/testdata/config.yaml

@@ -78,6 +78,19 @@ prometheus_remote_write/v1_no_translation:
   protobuf_message: "prometheus.WriteRequest"
   translation_strategy: "NoTranslation"
 
+prometheus_remote_write/include_metadata_keys:
+  endpoint: "localhost:8888"
+  include_metadata_keys:
+    - "target-id"
+    - "x-org-id"
+  remote_write_queue:
+    queue_size: 1000
+    num_consumers: 5
+
+prometheus_remote_write/reserved_metadata_keys:
+  endpoint: "localhost:8888"
+  include_metadata_keys:
+    - "content-type"
 prometheus_remote_write/negative_max_batch_size_bytes:
   endpoint: "localhost:8888"
-  max_batch_size_bytes: -1
+  max_batch_size_bytes: -1