loadbalancingexporter: batch logs after routing

.chloggen/saw-6744-loadbalancing-log-batcher.yaml

@@ -0,0 +1,8 @@
+change_type: enhancement
+component: exporter/loadbalancing
+note: Add an optional post-routing log batcher to the loadbalancing exporter with per-backend buffering and configurable flush limits.
+issues: [6744]
+subtext: |
+  The new `log_batcher` config is disabled by default for backward compatibility.
+  When enabled, logs are batched per resolved backend before downstream queue compression.
+change_logs: [user]

.codecov.yml

@@ -480,6 +480,10 @@ component_management:
     name: processor_groupbytrace
     paths:
     - processor/groupbytraceprocessor/**
+  - component_id: processor_hotreload
+    name: processor_hotreload
+    paths:
+    - processor/hotreloadprocessor/**
   - component_id: processor_interval
     name: processor_interval
     paths:
@@ -496,6 +500,10 @@ component_management:
     name: processor_logdedup
     paths:
     - processor/logdedupprocessor/**
+  - component_id: processor_logstometrics
+    name: processor_logstometrics
+    paths:
+    - processor/logstometricsprocessor/**
   - component_id: processor_logstransform
     name: processor_logstransform
     paths:

.github/ISSUE_TEMPLATE/beta_stability.yaml

@@ -198,10 +198,12 @@ body:
       - processor/geoip
       - processor/groupbyattrs
       - processor/groupbytrace
+      - processor/hotreload
       - processor/interval
       - processor/isolationforest
       - processor/k8sattributes
       - processor/logdedup
+      - processor/logstometrics
       - processor/logstransform
       - processor/metricsgeneration
       - processor/metricstarttime

.github/ISSUE_TEMPLATE/bug_report.yaml

@@ -201,10 +201,12 @@ body:
       - processor/geoip
       - processor/groupbyattrs
       - processor/groupbytrace
+      - processor/hotreload
       - processor/interval
       - processor/isolationforest
       - processor/k8sattributes
       - processor/logdedup
+      - processor/logstometrics
       - processor/logstransform
       - processor/metricsgeneration
       - processor/metricstarttime

.github/ISSUE_TEMPLATE/feature_request.yaml

@@ -195,10 +195,12 @@ body:
       - processor/geoip
       - processor/groupbyattrs
       - processor/groupbytrace
+      - processor/hotreload
       - processor/interval
       - processor/isolationforest
       - processor/k8sattributes
       - processor/logdedup
+      - processor/logstometrics
       - processor/logstransform
       - processor/metricsgeneration
       - processor/metricstarttime

.github/ISSUE_TEMPLATE/other.yaml

@@ -195,10 +195,12 @@ body:
       - processor/geoip
       - processor/groupbyattrs
       - processor/groupbytrace
+      - processor/hotreload
       - processor/interval
       - processor/isolationforest
       - processor/k8sattributes
       - processor/logdedup
+      - processor/logstometrics
       - processor/logstransform
       - processor/metricsgeneration
       - processor/metricstarttime

.github/ISSUE_TEMPLATE/unmaintained.yaml

@@ -200,10 +200,12 @@ body:
       - processor/geoip
       - processor/groupbyattrs
       - processor/groupbytrace
+      - processor/hotreload
       - processor/interval
       - processor/isolationforest
       - processor/k8sattributes
       - processor/logdedup
+      - processor/logstometrics
       - processor/logstransform
       - processor/metricsgeneration
       - processor/metricstarttime

exporter/loadbalancingexporter/README.md

@@ -124,6 +124,11 @@ Refer to [config.yaml](./testdata/config.yaml) for detailed examples on using th
   * `streamID`: Routes metrics based on their datapoint streamID. That's the unique hash of all it's attributes, plus the attributes and identifying information of its resource, scope, and metric data
 * loadbalancing exporter supports set of standard [queuing, retry and timeout settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md), but they are disable by default to maintain compatibility
 * The `routing_attributes` property is used to list the attributes that should be used if the `routing_key` is `attributes`.
+* The `log_batcher` property enables post-routing log batching per backend. It is `disabled` by default for backward compatibility.
+  * `enabled` turns post-routing log batching on or off.
+  * `max_records` flushes a backend batch when it reaches this many log records. Default: `512`.
+  * `max_bytes` flushes a backend batch when its serialized OTLP payload size before compression reaches this many bytes. Default: `1048576` (`1 MiB`).
+  * `flush_interval` flushes a backend batch after this interval even if size limits are not reached. Default: `100ms`.
 
 Simple example
 
@@ -138,6 +143,11 @@ processors:
 
 exporters:
   loadbalancing:
+    log_batcher:
+      enabled: true
+      max_records: 512
+      max_bytes: 1048576
+      flush_interval: 100ms
     routing_key: "service"
     protocol:
       otlp:
@@ -236,17 +246,18 @@ service:
         - loadbalancing
 ```
 
-To compress payloads in an in-memory sending queue, set:
+To compress queued payloads before they are written to persistent queue storage, set:
 
 ```yaml
 exporters:
   loadbalancing:
     sending_queue:
       enabled: true
       payload_compression: zstd
-      compress_in_memory: true
 ```
 
+`sending_queue.compress_in_memory` is currently not supported by this exporter helper version. If set, config validation fails instead of silently ignoring it.
+
 Kubernetes resolver example (For a more specific example: [example/k8s-resolver](./example/k8s-resolver/README.md))
 > [!IMPORTANT]
 > The k8s resolver requires proper permissions. See [the full example](./example/k8s-resolver/README.md) for more information.

exporter/loadbalancingexporter/config.go

@@ -40,7 +40,8 @@ const (
 type Config struct {
 	TimeoutSettings           exporterhelper.TimeoutConfig `mapstructure:",squash"`
 	configretry.BackOffConfig `mapstructure:"retry_on_failure"`
-	QueueSettings             QueueSettings `mapstructure:"sending_queue"`
+	QueueSettings             QueueSettings    `mapstructure:"sending_queue"`
+	LogBatcher                LogBatcherConfig `mapstructure:"log_batcher"`
 
 	Protocol Protocol         `mapstructure:"protocol"`
 	Resolver ResolverSettings `mapstructure:"resolver"`
@@ -61,6 +62,13 @@ type QueueSettings struct {
 	CompressInMemory                bool                    `mapstructure:"compress_in_memory"`
 }
 
+type LogBatcherConfig struct {
+	Enabled       bool          `mapstructure:"enabled"`
+	MaxRecords    int           `mapstructure:"max_records"`
+	MaxBytes      int           `mapstructure:"max_bytes"`
+	FlushInterval time.Duration `mapstructure:"flush_interval"`
+}
+
 func (q *QueueSettings) Unmarshal(conf *confmap.Conf) error {
 	if conf == nil {
 		return nil
@@ -117,18 +125,34 @@ func (q QueueSettings) Validate() error {
 		return fmt.Errorf("sending_queue.payload_compression must be one of [none, snappy, zstd], found %q", q.PayloadCompression)
 	}
 
-	if q.CompressInMemory && !q.Enabled {
-		return errors.New("sending_queue.compress_in_memory requires sending_queue.enabled=true")
-	}
-	if q.CompressInMemory && (q.PayloadCompression == "" || q.PayloadCompression == QueuePayloadCompressionNone) {
-		return errors.New("sending_queue.compress_in_memory requires sending_queue.payload_compression to be set to snappy or zstd")
+	if q.CompressInMemory {
+		return errors.New("sending_queue.compress_in_memory is not supported by this exporter helper version; use sending_queue.payload_compression instead")
 	}
 
 	return nil
 }
 
 func (cfg *Config) Validate() error {
-	return cfg.QueueSettings.Validate()
+	if err := cfg.QueueSettings.Validate(); err != nil {
+		return err
+	}
+	return cfg.LogBatcher.Validate()
+}
+
+func (c LogBatcherConfig) Validate() error {
+	if !c.Enabled {
+		return nil
+	}
+	if c.MaxRecords <= 0 {
+		return errors.New("log_batcher.max_records must be greater than 0 when log_batcher.enabled=true")
+	}
+	if c.MaxBytes <= 0 {
+		return errors.New("log_batcher.max_bytes must be greater than 0 when log_batcher.enabled=true")
+	}
+	if c.FlushInterval <= 0 {
+		return errors.New("log_batcher.flush_interval must be greater than 0 when log_batcher.enabled=true")
+	}
+	return nil
 }
 
 // Protocol holds the individual protocol-specific settings. Only OTLP is supported at the moment.

exporter/loadbalancingexporter/config_test.go

@@ -6,6 +6,7 @@ package loadbalancingexporter
 import (
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
@@ -43,14 +44,34 @@ func TestConfigValidatePayloadCompression(t *testing.T) {
 }
 
 func TestConfigValidateCompressInMemory(t *testing.T) {
+	// compress_in_memory is unsupported; any use must fail immediately regardless of other settings.
 	cfg := createDefaultConfig().(*Config)
 	cfg.QueueSettings.CompressInMemory = true
-	require.ErrorContains(t, cfg.Validate(), "sending_queue.compress_in_memory requires sending_queue.enabled=true")
+	require.ErrorContains(t, cfg.Validate(), "sending_queue.compress_in_memory is not supported")
 
+	// Same error even when enabled=true and payload_compression is set.
 	cfg.QueueSettings.Enabled = true
-	require.ErrorContains(t, cfg.Validate(), "sending_queue.compress_in_memory requires sending_queue.payload_compression")
-
 	cfg.QueueSettings.PayloadCompression = QueuePayloadCompressionSnappy
+	require.ErrorContains(t, cfg.Validate(), "sending_queue.compress_in_memory is not supported")
+}
+
+func TestConfigValidateLogBatcher(t *testing.T) {
+	cfg := createDefaultConfig().(*Config)
+	require.NoError(t, cfg.Validate())
+
+	cfg.LogBatcher.Enabled = true
+	cfg.LogBatcher.MaxRecords = 0
+	require.ErrorContains(t, cfg.Validate(), "log_batcher.max_records")
+
+	cfg.LogBatcher.MaxRecords = 10
+	cfg.LogBatcher.MaxBytes = 0
+	require.ErrorContains(t, cfg.Validate(), "log_batcher.max_bytes")
+
+	cfg.LogBatcher.MaxBytes = 1024
+	cfg.LogBatcher.FlushInterval = 0
+	require.ErrorContains(t, cfg.Validate(), "log_batcher.flush_interval")
+
+	cfg.LogBatcher.FlushInterval = time.Second
 	require.NoError(t, cfg.Validate())
 }
 
@@ -83,3 +104,34 @@ func TestLoadConfigWithQueueCompression(t *testing.T) {
 	require.Equal(t, QueuePayloadCompressionZstd, cfg.QueueSettings.PayloadCompression)
 	require.True(t, cfg.QueueSettings.CompressInMemory)
 }
+
+func TestLoadConfigWithLogBatcher(t *testing.T) {
+	cfg := createDefaultConfig().(*Config)
+	conf := confmap.NewFromStringMap(map[string]any{
+		"protocol": map[string]any{
+			"otlp": map[string]any{
+				"endpoint": "localhost:4317",
+				"tls": map[string]any{
+					"insecure": true,
+				},
+			},
+		},
+		"resolver": map[string]any{
+			"static": map[string]any{
+				"hostnames": []string{"localhost:4317"},
+			},
+		},
+		"log_batcher": map[string]any{
+			"enabled":        true,
+			"max_records":    1024,
+			"max_bytes":      2097152,
+			"flush_interval": "250ms",
+		},
+	})
+
+	require.NoError(t, conf.Unmarshal(cfg))
+	require.True(t, cfg.LogBatcher.Enabled)
+	require.Equal(t, 1024, cfg.LogBatcher.MaxRecords)
+	require.Equal(t, 2097152, cfg.LogBatcher.MaxBytes)
+	require.Equal(t, 250*time.Millisecond, cfg.LogBatcher.FlushInterval)
+}