[exporter/prometheusremotewrite] Make maxBatchByteSize configurable

.chloggen/prometheus-remote-write-max-batch-byte-size.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# 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: addition of `max_batch_size_bytes` configurable parameter, to allow users to adjust it based on the capabilities of their specific remote storage
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [21911]
+
+# (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:

exporter/prometheusremotewriteexporter/README.md

@@ -64,6 +64,9 @@ The following settings can be optionally configured:
   - `enabled` (default = false): If `enabled` is `true`, a `_created` metric is
     exported for Summary, Histogram, and Monotonic Sum metric points if
     `StartTimeUnixNano` is set.
+- `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.
 
 Example:
 

exporter/prometheusremotewriteexporter/config.go

@@ -31,6 +31,9 @@ type Config struct {
 
 	HTTPClientSettings confighttp.HTTPClientSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
 
+	// maximum size in bytes of time series batch sent to remote storage
+	MaxBatchSizeBytes int `mapstructure:"max_batch_size_bytes"`
+
 	// ResourceToTelemetrySettings is the option for converting resource attributes to telemetry attributes.
 	// "Enabled" - A boolean field to enable/disable this option. Default is `false`.
 	// If enabled, all the resource attributes will be converted to metric labels by default.
@@ -100,5 +103,13 @@ func (cfg *Config) Validate() error {
 			Enabled: false,
 		}
 	}
+	if cfg.MaxBatchSizeBytes < 0 {
+		return fmt.Errorf("max_batch_byte_size must be greater than 0")
+	}
+	if cfg.MaxBatchSizeBytes == 0 {
+		// Defaults to ~2.81MB
+		cfg.MaxBatchSizeBytes = 3000000
+	}
+
 	return nil
 }

exporter/prometheusremotewriteexporter/config_test.go

@@ -40,7 +40,8 @@ func TestLoadConfig(t *testing.T) {
 		{
 			id: component.NewIDWithName(metadata.Type, "2"),
 			expected: &Config{
-				TimeoutSettings: exporterhelper.NewDefaultTimeoutSettings(),
+				MaxBatchSizeBytes: 3000000,
+				TimeoutSettings:   exporterhelper.NewDefaultTimeoutSettings(),
 				RetrySettings: exporterhelper.RetrySettings{
 					Enabled:             true,
 					InitialInterval:     10 * time.Second,

exporter/prometheusremotewriteexporter/exporter.go

@@ -31,22 +31,20 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheusremotewrite"
 )
 
-const maxBatchByteSize = 3000000
-
 // 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
-	clientSettings  *confighttp.HTTPClientSettings
-	settings        component.TelemetrySettings
-	retrySettings   exporterhelper.RetrySettings
-
-	wal              *prweWAL
-	exporterSettings prometheusremotewrite.Settings
+	endpointURL       *url.URL
+	client            *http.Client
+	wg                *sync.WaitGroup
+	closeChan         chan struct{}
+	concurrency       int
+	userAgentHeader   string
+	maxBatchSizeBytes int
+	clientSettings    *confighttp.HTTPClientSettings
+	settings          component.TelemetrySettings
+	retrySettings     exporterhelper.RetrySettings
+	wal               *prweWAL
+	exporterSettings  prometheusremotewrite.Settings
 }
 
 // newPRWExporter initializes a new prwExporter instance and sets fields accordingly.
@@ -64,14 +62,15 @@ func newPRWExporter(cfg *Config, set exporter.CreateSettings) (*prwExporter, 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,
-		concurrency:     cfg.RemoteWriteQueue.NumConsumers,
-		clientSettings:  &cfg.HTTPClientSettings,
-		settings:        set.TelemetrySettings,
-		retrySettings:   cfg.RetrySettings,
+		endpointURL:       endpointURL,
+		wg:                new(sync.WaitGroup),
+		closeChan:         make(chan struct{}),
+		userAgentHeader:   userAgentHeader,
+		maxBatchSizeBytes: cfg.MaxBatchSizeBytes,
+		concurrency:       cfg.RemoteWriteQueue.NumConsumers,
+		clientSettings:    &cfg.HTTPClientSettings,
+		settings:          set.TelemetrySettings,
+		retrySettings:     cfg.RetrySettings,
 		exporterSettings: prometheusremotewrite.Settings{
 			Namespace:           cfg.Namespace,
 			ExternalLabels:      sanitizedLabels,
@@ -159,7 +158,7 @@ func (prwe *prwExporter) handleExport(ctx context.Context, tsMap map[string]*pro
 	}
 
 	// Calls the helper function to convert and batch the TsMap to the desired format
-	requests, err := batchTimeSeries(tsMap, maxBatchByteSize)
+	requests, err := batchTimeSeries(tsMap, prwe.maxBatchSizeBytes)
 	if err != nil {
 		return err
 	}

exporter/prometheusremotewriteexporter/exporter_test.go

@@ -131,10 +131,11 @@ func Test_NewPRWExporter(t *testing.T) {
 // Test_Start checks if the client is properly created as expected.
 func Test_Start(t *testing.T) {
 	cfg := &Config{
-		TimeoutSettings: exporterhelper.TimeoutSettings{},
-		RetrySettings:   exporterhelper.RetrySettings{},
-		Namespace:       "",
-		ExternalLabels:  map[string]string{},
+		TimeoutSettings:   exporterhelper.TimeoutSettings{},
+		RetrySettings:     exporterhelper.RetrySettings{},
+		MaxBatchSizeBytes: 3000000,
+		Namespace:         "",
+		ExternalLabels:    map[string]string{},
 		TargetInfo: &TargetInfo{
 			Enabled: true,
 		},
@@ -684,7 +685,8 @@ func Test_PushMetrics(t *testing.T) {
 							ReadBufferSize:  0,
 							WriteBufferSize: 512 * 1024,
 						},
-						RemoteWriteQueue: RemoteWriteQueue{NumConsumers: 1},
+						MaxBatchSizeBytes: 3000000,
+						RemoteWriteQueue:  RemoteWriteQueue{NumConsumers: 1},
 						TargetInfo: &TargetInfo{
 							Enabled: true,
 						},

exporter/prometheusremotewriteexporter/factory.go

@@ -68,9 +68,10 @@ func createMetricsExporter(ctx context.Context, set exporter.CreateSettings,
 
 func createDefaultConfig() component.Config {
 	return &Config{
-		Namespace:       "",
-		ExternalLabels:  map[string]string{},
-		TimeoutSettings: exporterhelper.NewDefaultTimeoutSettings(),
+		Namespace:         "",
+		ExternalLabels:    map[string]string{},
+		MaxBatchSizeBytes: 3000000,
+		TimeoutSettings:   exporterhelper.NewDefaultTimeoutSettings(),
 		RetrySettings: exporterhelper.RetrySettings{
 			Enabled:             true,
 			InitialInterval:     50 * time.Millisecond,