Incorporate review comments add the middleware interface

Author: raghu999

.chloggen/controller-interface.yaml

@@ -1,13 +1,11 @@
 change_type: enhancement
 component: exporterhelper
 note: >
-  Add `ConcurrencyController` interface support to the queue sender. This allows
-  extensions to dynamically control the number of concurrent requests in the
-  export pipeline.
+  Add `RequestMiddleware` interface support to the queue sender. This allows
+  extensions to intercept and manage export requests (e.g., for dynamic concurrency control).
 issues: [14080]
 subtext: |
-  The `ConcurrencyControllerFactory` interface has been added to a new
+  The `RequestMiddlewareFactory` interface has been added to the
   `extension/extensioncapabilities` module. The `exporterhelper`'s sending queue
-  can now be configured with a `concurrency_controller` ID to delegate flow
-  control logic to an extension.
+  can now be configured with a `request_middleware` ID to delegate request execution logic to an extension.
 change_logs: [user, api]

exporter/exporterhelper/README.md

@@ -21,15 +21,16 @@ The following configuration options can be modified:
 - `sending_queue`
   - `enabled` (default = true)
   - `num_consumers` (default = 10): Number of consumers that dequeue batches; ignored if `enabled` is `false`
-  - `wait_for_result` (default = false): determines if incoming requests are blocked until the request is processed or not.
+  - `wait_for_result` (default = false): Determines if incoming requests are blocked until the request is processed or not.
   - `block_on_overflow` (default = false): If true, blocks the request until the queue has space otherwise rejects the data immediately; ignored if `enabled` is `false`
   - `sizer` (default = requests): How the queue and batching is measured. Available options: 
     - `requests`: number of incoming batches of metrics, logs, traces (the most performant option);
     - `items`: number of the smallest parts of each signal (spans, metric data points, log records);
     - `bytes`: the size of serialized data in bytes (the least performant option).
   - `queue_size` (default = 1000): Maximum size the queue can accept. Measured in units defined by `sizer`
   - `batch`: see below.
-  - `concurrency_controller` (default = none): The ID of an extension implementing the `ConcurrencyController` interface (e.g., `adaptive_concurrency`). When configured, this extension dynamically manages the number of concurrent requests sent to the backend based on real-time signals like latency and error rates, providing adaptive backpressure to prevent downstream overload.
+  - `concurrency_controller` (default = none): The ID of an extension implementing the `RequestMiddlewareFactory` interface (e.g., `adaptive_concurrency`). When configured, exporterhelper executes export requests through the middleware, enabling logic such as adaptive concurrency, rate limiting, or circuit breaking.
+
 
 #### Sending queue batch settings
 
@@ -143,62 +144,42 @@ service:
 [filestorage]: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/extension/storage/filestorage
 
 
-### Concurrency Controller (Adaptive Request Concurrency)
-
-To use dynamic concurrency control, the following setting needs to be set:
-
-- `sending_queue`
-  - `concurrency_controller` (default = none): When set, enables adaptive backpressure by using the specified extension to dynamically manage the number of concurrent requests.
-
-#### How it works
-
-Traditionally, exporters use a static `num_consumers` to determine how many concurrent requests can be sent to a backend. However, static limits are difficult to tune:
-- **Too high:** You risk overwhelming the downstream backend, leading to increased latency, 429 (Too Many Requests) errors, and "death spirals."
-- **Too low:** You underutilize the available network and backend capacity, causing the collector's internal queue to fill up unnecessarily.
-
-The Concurrency Controller implementation (e.g., `adaptive_concurrency`) replaces the fixed worker pool with a dynamic permit system based on the **AIMD (Additive Increase / Multiplicative Decrease)** algorithm.
-
-#### The Control Loop
+### Request Middleware (e.g. Adaptive Concurrency)
 
-1. **Acquire:** Before an export attempt begins, the exporter asks the controller for a permit. If the current dynamic limit is reached, the request blocks until a slot becomes available.
-2. **Measure:** The controller tracks the **Round Trip Time (RTT)** and the outcome (success or retryable error) of every request.
-3. **Adapt:** At regular intervals, the controller compares the recent RTT baseline against current performance:
-   - **Increase:** If latency is stable and requests are succeeding, the controller increases the concurrency limit to maximize throughput.
-   - **Decrease:** If latency spikes or the backend returns "backpressure" signals (like HTTP 429 or gRPC `ResourceExhausted`), the controller immediately shrinks the limit to allow the backend to recover.
+Traditionally, exporters use a static `num_consumers` to determine how many concurrent requests can be sent to a backend. A Request Middleware implementation allows extensions to replace or augment this behavior with dynamic logic.
 
-This feedback loop ensures the Collector automatically finds the "sweet spot" of maximum throughput without requiring manual tuning as network conditions or backend capacity change.
+The middleware wraps the request execution, allowing it to:
+1. **Intercept:** Acquire permits or check conditions before the request starts (e.g., rate limiting).
+2. **Measure:** Track the duration and outcome of the request (e.g., adaptive concurrency).
+3. **Control:** Block or fail requests based on internal logic.
 
 #### Interaction with num_consumers
 
-When a concurrency_controller is configured, it acts as a gatekeeper on top of the existing queue consumers. The effective concurrency is the minimum of the controller's dynamic limit and the static num_consumers.
-
-To ensure the controller has enough headroom to operate, this component enforces a minimum of 200 consumers when a controller is active.
-
-- Automatic Adjustment: If you explicitly set num_consumers to a low value (e.g., 10), it will be automatically increased to 200 to prevent artificial bottlenecks.
+When a middleware is configured (via `concurrency_controller`), it acts as a gatekeeper on top of the existing queue consumers. The effective concurrency is the minimum of the middleware's logic and the static `num_consumers`.
 
-- High Concurrency: If you need more than 200 concurrent requests (e.g., num_consumers: 500), your configured value will be respected.
+`Effective_Concurrency = min(Middleware_Limit, sending_queue.num_consumers)`
 
-**Recommendation:** generally, you do not need to configure num_consumers when using the controller; the default headroom (200) is sufficient for most use cases. Only increase it if you expect to exceed 200 concurrent requests.
+- **Warning:** If you leave `num_consumers` at the default value (10) while using middleware that requires high concurrency (like Adaptive Request Concurrency), the queue sender will log a warning.
+- **Recommendation:** Set `num_consumers` high enough to avoid capping the middleware’s maximum intended concurrency (for example, match the middleware’s configured max).
 
 #### Example Configuration
 
-In this example, the OTLP exporter is configured to use the `adaptive_concurrency` extension. The extension will start with a small number of parallel requests and automatically scale up to 100 based on the health of the OTLP endpoint.
+In this example, an OTLP exporter is configured to use the `adaptive_concurrency` extension (which implements the Request Middleware interface).
 
 ```yaml
 exporters:
   otlp:
     endpoint: https://my-backend:4317
     sending_queue:
       enabled: true
-      # Link to the concurrency controller extension defined below
+      num_consumers: 100 # Provide headroom for the middleware
+      # Link to the middleware extension
       concurrency_controller: adaptive_concurrency/otlp_limit
 
 extensions:
-  # Define the adaptive concurrency algorithm
+  # The adaptive_concurrency extension implementation would be defined here
   adaptive_concurrency/otlp_limit:
-    initial_limit: 5      # Start with 5 concurrent requests
-    max_concurrency: 100  # Never exceed 100 concurrent requests
-    decrease_ratio: 0.5   # If backpressure is detected, cut concurrency by half
+    # ... extension specific config ...
 
 service:
   extensions: [adaptive_concurrency/otlp_limit]

exporter/exporterhelper/go.mod

@@ -17,6 +17,7 @@ require (
 	go.opentelemetry.io/collector/consumer/consumertest v0.143.0
 	go.opentelemetry.io/collector/exporter v1.49.0
 	go.opentelemetry.io/collector/exporter/exportertest v0.143.0
+	go.opentelemetry.io/collector/extension/extensioncapabilities v1.143.0
 	go.opentelemetry.io/collector/extension/extensiontest v0.143.0
 	go.opentelemetry.io/collector/extension/xextension v0.143.0
 	go.opentelemetry.io/collector/featuregate v1.49.0

exporter/exporterhelper/internal/queue_sender.go

@@ -19,8 +19,6 @@ import (
 	"go.opentelemetry.io/collector/pipeline"
 )
 
-const minConsumersWithController = 200
-
 // NewDefaultQueueConfig returns the default config for queuebatch.Config.
 // By default:
 //
@@ -42,11 +40,11 @@ func NewDefaultQueueConfig() queuebatch.Config {
 	}
 }
 
-// queueSender wraps QueueBatch to add ConcurrencyController logic.
+// queueSender wraps QueueBatch to add RequestMiddleware logic.
 type queueSender struct {
 	*queuebatch.QueueBatch
-	ctrlID *component.ID
-	ctrl   extensioncapabilities.ConcurrencyController
+	mwID *component.ID
+	mw   extensioncapabilities.RequestMiddleware
 
 	// Capture these fields during creation so we can use them in Start()
 	// because QueueBatch does not expose them publicly.
@@ -60,7 +58,7 @@ func NewQueueSender(
 	exportFailureMessage string,
 	next sender.Sender[request.Request],
 ) (sender.Sender[request.Request], error) {
-	if qCfg.ConcurrencyControllerID == nil {
+	if qCfg.RequestMiddlewareID == nil {
 		exportFunc := func(ctx context.Context, req request.Request) error {
 			// Have to read the number of items before sending the request since the request can
 			// be modified by the downstream components like the batcher.
@@ -76,44 +74,32 @@ func NewQueueSender(
 		return queuebatch.NewQueueBatch(qSet, qCfg, exportFunc)
 	}
 
-	enforceMinConsumers(&qCfg, qSet.Telemetry.Logger)
+	warnIfNumConsumersMayCapMiddleware(&qCfg, qSet.Telemetry.Logger)
 
 	qs := &queueSender{
-		ctrlID: qCfg.ConcurrencyControllerID,
+		mwID:   qCfg.RequestMiddlewareID,
 		id:     qSet.ID,
 		signal: qSet.Signal,
+		mw:     extensioncapabilities.NoopRequestMiddleware(), // Initialize with no-op to avoid nil checks
 	}
 
 	exportFunc := func(ctx context.Context, req request.Request) error {
 		// Have to read the number of items before sending the request since the request can
 		// be modified by the downstream components like the batcher.
 		itemsCount := req.ItemsCount()
 
-		ctrl := qs.ctrl
-		if ctrl == nil {
+		// Wrap the actual export call in a closure for the middleware
+		nextCall := func(ctx context.Context) error {
 			if errSend := next.Send(ctx, req); errSend != nil {
 				qSet.Telemetry.Logger.Error("Exporting failed. Dropping data."+exportFailureMessage,
 					zap.Error(errSend), zap.Int("dropped_items", itemsCount))
 				return errSend
 			}
 			return nil
 		}
-		// This blocks the worker if the dynamic limit is reached.
-		if err := ctrl.Acquire(ctx); err != nil {
-			return err
-		}
 
-		start := time.Now()
-		var errSend error
-		defer func() { ctrl.Record(ctx, time.Since(start), errSend) }()
-
-		errSend = next.Send(ctx, req)
-		if errSend != nil {
-			qSet.Telemetry.Logger.Error("Exporting failed. Dropping data."+exportFailureMessage,
-				zap.Error(errSend), zap.Int("dropped_items", itemsCount))
-			return errSend
-		}
-		return nil
+		// Use the middleware to execute the request
+		return qs.mw.Handle(ctx, nextCall)
 	}
 
 	qb, err := queuebatch.NewQueueBatch(qSet, qCfg, exportFunc)
@@ -125,75 +111,66 @@ func NewQueueSender(
 	return qs, nil
 }
 
-// enforceMinConsumers ensures sufficient worker headroom when a concurrency controller is configured.
-func enforceMinConsumers(cfg *queuebatch.Config, logger *zap.Logger) {
-	if cfg.ConcurrencyControllerID == nil {
+// warnIfNumConsumersMayCapMiddleware ensures sufficient worker headroom when a concurrency controller is configured.
+func warnIfNumConsumersMayCapMiddleware(cfg *queuebatch.Config, logger *zap.Logger) {
+	if cfg.RequestMiddlewareID == nil {
 		return
 	}
 
 	defaultConsumers := NewDefaultQueueConfig().NumConsumers
 
-	// If the user left the default worker count, bump it to allow the controller to be the governor.
-	if cfg.NumConsumers == defaultConsumers && cfg.NumConsumers < minConsumersWithController {
-		if logger != nil {
-			logger.Warn("sending_queue.num_consumers overridden because concurrency_controller is configured",
-				zap.Int("original", cfg.NumConsumers),
-				zap.Int("override", minConsumersWithController),
-				zap.String("controller", cfg.ConcurrencyControllerID.String()),
-			)
-		}
-		cfg.NumConsumers = minConsumersWithController
-		return
-	}
-
-	// If the user explicitly set num_consumers below the recommended headroom, don't override,
-	// but warn that the controller will be capped by the worker pool.
-	if cfg.NumConsumers > 0 && cfg.NumConsumers < minConsumersWithController && logger != nil {
-		logger.Warn("sending_queue.num_consumers caps concurrency_controller; consider increasing num_consumers for headroom",
+	// If the user left the default worker count, warn that the worker pool might artificially cap the controller.
+	if cfg.NumConsumers == defaultConsumers && logger != nil {
+		logger.Warn("sending_queue.num_consumers is at the default; request middleware may be capped by worker pool",
 			zap.Int("num_consumers", cfg.NumConsumers),
-			zap.Int("recommended_min", minConsumersWithController),
-			zap.String("controller", cfg.ConcurrencyControllerID.String()),
+			zap.String("concurrency_controller", cfg.RequestMiddlewareID.String()),
 		)
 	}
 }
 
 // Start overrides the default Start to resolve the extension.
 func (qs *queueSender) Start(ctx context.Context, host component.Host) error {
-	if qs.ctrlID != nil {
-		ext, ok := host.GetExtensions()[*qs.ctrlID]
+	if qs.mwID != nil {
+		ext, ok := host.GetExtensions()[*qs.mwID]
 		if !ok {
-			return fmt.Errorf("concurrency controller extension %q not found", qs.ctrlID.String())
+			return fmt.Errorf("request middleware extension %q not found", qs.mwID.String())
 		}
-		factory, ok := ext.(extensioncapabilities.ConcurrencyControllerFactory)
+		factory, ok := ext.(extensioncapabilities.RequestMiddlewareFactory)
 		if !ok {
-			return fmt.Errorf("extension %q does not implement ConcurrencyControllerFactory", qs.ctrlID.String())
+			return fmt.Errorf("extension %q does not implement RequestMiddlewareFactory", qs.mwID.String())
 		}
-		ctrl, err := factory.CreateConcurrencyController(qs.id, qs.signal)
+		mw, err := factory.CreateRequestMiddleware(qs.id, qs.signal)
 		if err != nil {
-			return fmt.Errorf("failed to create concurrency controller: %w", err)
+			return fmt.Errorf("failed to create request middleware: %w", err)
+		}
+
+		// Guard against nil return from factory to avoid panic in Handle()
+		if mw != nil {
+			qs.mw = mw
 		}
-		qs.ctrl = ctrl
 	}
 
 	if err := qs.QueueBatch.Start(ctx, host); err != nil {
-		if qs.ctrl != nil {
-			_ = qs.ctrl.Shutdown(ctx)
-			qs.ctrl = nil
+		if qs.mw != nil {
+			_ = qs.mw.Shutdown(ctx)
+			// Reset to no-op on failure cleanup
+			qs.mw = extensioncapabilities.NoopRequestMiddleware()
 		}
 		return err
 	}
 	return nil
 }
 
-// Shutdown ensures the controller is also shut down.
+// Shutdown ensures the middleware is also shut down.
 func (qs *queueSender) Shutdown(ctx context.Context) error {
 	errQ := qs.QueueBatch.Shutdown(ctx)
 
-	if qs.ctrl != nil {
-		if err := qs.ctrl.Shutdown(ctx); err != nil && errQ == nil {
+	if qs.mw != nil {
+		if err := qs.mw.Shutdown(ctx); err != nil && errQ == nil {
 			return err
 		}
-		qs.ctrl = nil
+		// Reset to no-op after shutdown
+		qs.mw = extensioncapabilities.NoopRequestMiddleware()
 	}
 	return errQ
 }