[batchprocessor] Modernize the batch processor

.chloggen/batchprocessor-exporterhelper.yaml

@@ -0,0 +1,16 @@
+change_type: enhancement
+
+component: processor/batch
+
+note: batch processor implemented by exporterhelper behind feature flags
+
+issues: [13582]
+
+subtext: |
+  there are two feature gates:
+  - processor.batch.useExporterHelper: use the new implementation
+  - processor.batch.propagateErrors: return errors, implies backpressure
+  new features are functional however the processor produces exporter metrics,
+  see https://github.com/open-telemetry/opentelemetry-collector/issues/14038
+
+change_logs: [user]

processor/batchprocessor/README.md

@@ -112,3 +112,21 @@ metadata-key values.
 
 The number of batch processors currently in use is exported as the
 `otelcol_processor_batch_metadata_cardinality` metric.
+
+## Modernization feature gates
+
+There are two feature gates currently defined for this component,
+which we are in the processor of modernizing.
+
+- `processor.batch.useExporterHelper`: Use the new exporterhelper-based QueueBatch logic, the same logic used in all exporters
+- `processor.batch.propagateErrors`: Sets the exporterhelper's `WaitForResult` to true, making it delay in order to return errors from the pipeline
+
+This mode is currently behind a feature gate and still experimental,
+mainly because metrics emitted by the component are incorrectly named.
+At present, when the exporterhelper mode is used, the component emits
+exporter metrics. See [#14038](https://github.com/open-telemetry/opentelemetry-collector/issues/14038).
+
+As we move towards finalizing these feature gates, it would be reasonable
+to enable at least one more feature of the underlying QueueBatch logic,
+the `num_consumers` parameter which is currently set to 1 can be raised
+to enable concurrent-batching behavior.

processor/batchprocessor/batch_processor.go

@@ -111,7 +111,7 @@ type batch[T any] interface {
 }
 
 // newBatchProcessor returns a new batch processor component.
-func newBatchProcessor[T any](set processor.Settings, cfg *Config, batchFunc func() batch[T]) (*batchProcessor[T], error) {
+func newBatchProcessor[T any](_ context.Context, set processor.Settings, cfg *Config, batchFunc func() batch[T]) (*batchProcessor[T], error) {
 	// use lower-case, to be consistent with http/2 headers.
 	mks := make([]string, len(cfg.MetadataKeys))
 	for i, k := range cfg.MetadataKeys {
@@ -376,8 +376,8 @@ type tracesBatchProcessor struct {
 }
 
 // newTracesBatchProcessor creates a new batch processor that batches traces by size or with timeout
-func newTracesBatchProcessor(set processor.Settings, next consumer.Traces, cfg *Config) (processor.Traces, error) {
-	bp, err := newBatchProcessor(set, cfg, func() batch[ptrace.Traces] { return newBatchTraces(next) })
+func newTracesBatchProcessor(ctx context.Context, set processor.Settings, next consumer.Traces, cfg *Config) (processor.Traces, error) {
+	bp, err := newBatchProcessor(ctx, set, cfg, func() batch[ptrace.Traces] { return newBatchTraces(next) })
 	if err != nil {
 		return nil, err
 	}
@@ -394,8 +394,8 @@ type metricsBatchProcessor struct {
 }
 
 // newMetricsBatchProcessor creates a new batch processor that batches metrics by size or with timeout
-func newMetricsBatchProcessor(set processor.Settings, next consumer.Metrics, cfg *Config) (processor.Metrics, error) {
-	bp, err := newBatchProcessor(set, cfg, func() batch[pmetric.Metrics] { return newMetricsBatch(next) })
+func newMetricsBatchProcessor(ctx context.Context, set processor.Settings, next consumer.Metrics, cfg *Config) (processor.Metrics, error) {
+	bp, err := newBatchProcessor(ctx, set, cfg, func() batch[pmetric.Metrics] { return newMetricsBatch(next) })
 	if err != nil {
 		return nil, err
 	}
@@ -413,8 +413,8 @@ type logsBatchProcessor struct {
 }
 
 // newLogsBatchProcessor creates a new batch processor that batches logs by size or with timeout
-func newLogsBatchProcessor(set processor.Settings, next consumer.Logs, cfg *Config) (processor.Logs, error) {
-	bp, err := newBatchProcessor(set, cfg, func() batch[plog.Logs] { return newBatchLogs(next) })
+func newLogsBatchProcessor(ctx context.Context, set processor.Settings, next consumer.Logs, cfg *Config) (processor.Logs, error) {
+	bp, err := newBatchProcessor(ctx, set, cfg, func() batch[plog.Logs] { return newBatchLogs(next) })
 	if err != nil {
 		return nil, err
 	}