Incorporate review suggestions

Author: raghu999

.chloggen/controller-interface.yaml

@@ -6,6 +6,6 @@ note: >
 issues: [14080]
 subtext: |
   The `RequestMiddlewareFactory` interface has been added to the
-  `extension/extensioncapabilities` module. The `exporterhelper`'s sending queue
-  can now be configured with a `request_middleware` ID to delegate request execution logic to an extension.
+  `extension/xextension/extensionmiddleware` package. The `exporterhelper`'s sending queue
+  can now be configured with a list of `request_middlewares` to delegate request execution logic to extensions.
 change_logs: [user, api]

exporter/exporterhelper/README.md

@@ -29,8 +29,6 @@ The following configuration options can be modified:
     - `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 `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
 
@@ -142,48 +140,3 @@ service:
 ```
 
 [filestorage]: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/extension/storage/filestorage
-
-
-### Request Middleware (e.g. Adaptive Concurrency)
-
-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.
-
-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 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`.
-
-`Effective_Concurrency = min(Middleware_Limit, sending_queue.num_consumers)`
-
-- **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, 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
-      num_consumers: 100 # Provide headroom for the middleware
-      # Link to the middleware extension
-      concurrency_controller: adaptive_concurrency/otlp_limit
-
-extensions:
-  # The adaptive_concurrency extension implementation would be defined here
-  adaptive_concurrency/otlp_limit:
-    # ... extension specific config ...
-
-service:
-  extensions: [adaptive_concurrency/otlp_limit]
-  pipelines:
-    traces:
-      receivers: [otlp]
-      exporters: [otlp]

exporter/exporterhelper/go.mod

@@ -17,7 +17,6 @@ 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
@@ -123,5 +122,3 @@ replace go.opentelemetry.io/collector/config/configoptional => ../../config/conf
 replace go.opentelemetry.io/collector/confmap/xconfmap => ../../confmap/xconfmap
 
 replace go.opentelemetry.io/collector/internal/testutil => ../../internal/testutil
-
-replace go.opentelemetry.io/collector/extension/extensioncapabilities => ../../extension/extensioncapabilities

exporter/exporterhelper/internal/queue_sender.go

@@ -8,14 +8,17 @@ import (
 	"fmt"
 	"time"
 
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
-	"go.opentelemetry.io/collector/extension/extensioncapabilities"
+
+	// Import the new package (aliased to avoid confusion with existing extensionmiddleware)
+	requestmiddleware "go.opentelemetry.io/collector/extension/xextension/extensionmiddleware"
 	"go.opentelemetry.io/collector/pipeline"
 )
 
@@ -43,8 +46,8 @@ func NewDefaultQueueConfig() queuebatch.Config {
 // queueSender wraps QueueBatch to add RequestMiddleware logic.
 type queueSender struct {
 	*queuebatch.QueueBatch
-	mwID *component.ID
-	mw   extensioncapabilities.RequestMiddleware
+	mwIDs []component.ID
+	mws   []requestmiddleware.RequestMiddleware
 
 	// Capture these fields during creation so we can use them in Start()
 	// because QueueBatch does not expose them publicly.
@@ -58,7 +61,7 @@ func NewQueueSender(
 	exportFailureMessage string,
 	next sender.Sender[request.Request],
 ) (sender.Sender[request.Request], error) {
-	if qCfg.RequestMiddlewareID == nil {
+	if len(qCfg.RequestMiddlewares) == 0 {
 		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.
@@ -77,19 +80,19 @@ func NewQueueSender(
 	warnIfNumConsumersMayCapMiddleware(&qCfg, qSet.Telemetry.Logger)
 
 	qs := &queueSender{
-		mwID:   qCfg.RequestMiddlewareID,
+		mwIDs:  qCfg.RequestMiddlewares,
 		id:     qSet.ID,
 		signal: qSet.Signal,
-		mw:     extensioncapabilities.NoopRequestMiddleware(), // Initialize with no-op to avoid nil checks
+		mws:    make([]requestmiddleware.RequestMiddleware, 0, len(qCfg.RequestMiddlewares)),
 	}
 
 	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()
 
-		// Wrap the actual export call in a closure for the middleware
-		nextCall := func(ctx context.Context) error {
+		// 1. Define the base exporter call
+		baseSender := 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))
@@ -98,8 +101,19 @@ func NewQueueSender(
 			return nil
 		}
 
-		// Use the middleware to execute the request
-		return qs.mw.Handle(ctx, nextCall)
+		// 2. Chain middlewares: m1(m2(base))
+		// We iterate in reverse so that the first middleware in the config is the outer-most wrapper.
+		nextCall := baseSender
+		for i := len(qs.mws) - 1; i >= 0; i-- {
+			mw := qs.mws[i]
+			currNext := nextCall
+			nextCall = func(ctx context.Context) error {
+				return mw.Handle(ctx, currNext)
+			}
+		}
+
+		// 3. Execute the chain
+		return nextCall(ctx)
 	}
 
 	qb, err := queuebatch.NewQueueBatch(qSet, qCfg, exportFunc)
@@ -111,9 +125,9 @@ func NewQueueSender(
 	return qs, nil
 }
 
-// warnIfNumConsumersMayCapMiddleware ensures sufficient worker headroom when a concurrency controller is configured.
+// warnIfNumConsumersMayCapMiddleware ensures sufficient worker headroom when request middleware is configured.
 func warnIfNumConsumersMayCapMiddleware(cfg *queuebatch.Config, logger *zap.Logger) {
-	if cfg.RequestMiddlewareID == nil {
+	if len(cfg.RequestMiddlewares) == 0 {
 		return
 	}
 
@@ -123,53 +137,65 @@ func warnIfNumConsumersMayCapMiddleware(cfg *queuebatch.Config, logger *zap.Logg
 	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.String("concurrency_controller", cfg.RequestMiddlewareID.String()),
+			zap.Strings("request_middlewares", func() []string {
+				ids := make([]string, len(cfg.RequestMiddlewares))
+				for i, id := range cfg.RequestMiddlewares {
+					ids[i] = id.String()
+				}
+				return ids
+			}()),
 		)
 	}
 }
 
-// Start overrides the default Start to resolve the extension.
+// Start overrides the default Start to resolve the extensions.
 func (qs *queueSender) Start(ctx context.Context, host component.Host) error {
-	if qs.mwID != nil {
-		ext, ok := host.GetExtensions()[*qs.mwID]
+	for _, id := range qs.mwIDs {
+		ext, ok := host.GetExtensions()[id]
 		if !ok {
-			return fmt.Errorf("request middleware extension %q not found", qs.mwID.String())
+			// Cleanup any middlewares successfully created before this failure
+			_ = qs.Shutdown(ctx)
+			return fmt.Errorf("request middleware extension %q not found", id.String())
 		}
-		factory, ok := ext.(extensioncapabilities.RequestMiddlewareFactory)
+		factory, ok := ext.(requestmiddleware.RequestMiddlewareFactory)
 		if !ok {
-			return fmt.Errorf("extension %q does not implement RequestMiddlewareFactory", qs.mwID.String())
+			// Cleanup any middlewares successfully created before this failure
+			_ = qs.Shutdown(ctx)
+			return fmt.Errorf("extension %q does not implement RequestMiddlewareFactory", id.String())
 		}
 		mw, err := factory.CreateRequestMiddleware(qs.id, qs.signal)
 		if err != nil {
-			return fmt.Errorf("failed to create request middleware: %w", err)
+			// Cleanup any middlewares successfully created before this failure
+			_ = qs.Shutdown(ctx)
+			return fmt.Errorf("failed to create request middleware for %q: %w", id.String(), err)
 		}
 
-		// Guard against nil return from factory to avoid panic in Handle()
 		if mw != nil {
-			qs.mw = mw
+			qs.mws = append(qs.mws, mw)
 		}
 	}
 
 	if err := qs.QueueBatch.Start(ctx, host); err != nil {
-		// Ensure middleware is shut down if queue fails to start.
-		// No need to check for nil because we initialize with Noop.
-		_ = qs.mw.Shutdown(ctx)
-		qs.mw = extensioncapabilities.NoopRequestMiddleware()
+		// Ensure middlewares are shut down if queue fails to start.
+		_ = qs.Shutdown(ctx)
 		return err
 	}
 	return nil
 }
 
-// Shutdown ensures the middleware is also shut down.
+// Shutdown ensures the middlewares are also shut down.
 func (qs *queueSender) Shutdown(ctx context.Context) error {
-	errQ := qs.QueueBatch.Shutdown(ctx)
+	var errs error
+	if qs.QueueBatch != nil {
+		errs = multierr.Append(errs, qs.QueueBatch.Shutdown(ctx))
+	}
 
-	// No need to check for nil because we initialize with Noop.
-	if err := qs.mw.Shutdown(ctx); err != nil && errQ == nil {
-		return err
+	// Shutdown middlewares in reverse order (LIFO)
+	for i := len(qs.mws) - 1; i >= 0; i-- {
+		errs = multierr.Append(errs, qs.mws[i].Shutdown(ctx))
 	}
-	// Reset to no-op after shutdown
-	qs.mw = extensioncapabilities.NoopRequestMiddleware()
+	// Clear middlewares after shutdown
+	qs.mws = nil
 
-	return errQ
+	return errs
 }