[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_byte_size` 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

@@ -61,6 +61,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_byte_size` (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
+	MaxBatchByteSize int `mapstructure:"max_batch_byte_size"`
+
 	// 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.
@@ -97,5 +100,13 @@ func (cfg *Config) Validate() error {
 			Enabled: false,
 		}
 	}
+	if cfg.MaxBatchByteSize < 0 {
+		return fmt.Errorf("max_batch_byte_size must be greater than 0")
+	}
+	if cfg.MaxBatchByteSize == 0 {
+		// Defaults to ~2.81MB
+		cfg.MaxBatchByteSize = 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(),
+				MaxBatchByteSize: 3000000,
+				TimeoutSettings:  exporterhelper.NewDefaultTimeoutSettings(),
 				RetrySettings: exporterhelper.RetrySettings{
 					Enabled:             true,
 					InitialInterval:     10 * time.Second,

exporter/prometheusremotewriteexporter/exporter.go

@@ -29,18 +29,17 @@ 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
+	endpointURL      *url.URL
+	client           *http.Client
+	wg               *sync.WaitGroup
+	closeChan        chan struct{}
+	concurrency      int
+	userAgentHeader  string
+	maxBatchByteSize int
+	clientSettings   *confighttp.HTTPClientSettings
+	settings         component.TelemetrySettings
 
 	wal              *prweWAL
 	exporterSettings prometheusremotewrite.Settings
@@ -61,13 +60,14 @@ 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,
+		endpointURL:      endpointURL,
+		wg:               new(sync.WaitGroup),
+		closeChan:        make(chan struct{}),
+		userAgentHeader:  userAgentHeader,
+		maxBatchByteSize: cfg.MaxBatchByteSize,
+		concurrency:      cfg.RemoteWriteQueue.NumConsumers,
+		clientSettings:   &cfg.HTTPClientSettings,
+		settings:         set.TelemetrySettings,
 		exporterSettings: prometheusremotewrite.Settings{
 			Namespace:           cfg.Namespace,
 			ExternalLabels:      sanitizedLabels,
@@ -154,7 +154,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.maxBatchByteSize)
 	if err != nil {
 		return err
 	}

exporter/prometheusremotewriteexporter/exporter_test.go

@@ -129,10 +129,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{},
+		MaxBatchByteSize: 3000000,
+		Namespace:        "",
+		ExternalLabels:   map[string]string{},
 		TargetInfo: &TargetInfo{
 			Enabled: true,
 		},
@@ -675,6 +676,7 @@ func Test_PushMetrics(t *testing.T) {
 							ReadBufferSize:  0,
 							WriteBufferSize: 512 * 1024,
 						},
+						MaxBatchByteSize: 3000000,
 						RemoteWriteQueue: RemoteWriteQueue{NumConsumers: 1},
 						TargetInfo: &TargetInfo{
 							Enabled: true,

exporter/prometheusremotewriteexporter/factory.go

@@ -69,9 +69,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{},
+		MaxBatchByteSize: 3000000,
+		TimeoutSettings:  exporterhelper.NewDefaultTimeoutSettings(),
 		RetrySettings: exporterhelper.RetrySettings{
 			Enabled:             true,
 			InitialInterval:     50 * time.Millisecond,