Rough sketch (draft 3) limiter extension and middleware

cmd/otelcorecol/builder-config.yaml

@@ -46,6 +46,8 @@
   - go.opentelemetry.io/collector/config/configcompression => ../../config/configcompression
   - go.opentelemetry.io/collector/config/configgrpc => ../../config/configgrpc
   - go.opentelemetry.io/collector/config/confighttp => ../../config/confighttp
+  - go.opentelemetry.io/collector/config/configlimiter => ../../config/configlimiter
+  - go.opentelemetry.io/collector/config/configmiddleware => ../../config/configmiddleware
   - go.opentelemetry.io/collector/config/confignet => ../../config/confignet
   - go.opentelemetry.io/collector/config/configopaque => ../../config/configopaque
   - go.opentelemetry.io/collector/config/configretry => ../../config/configretry
@@ -79,6 +81,8 @@
   - go.opentelemetry.io/collector/extension/extensionauth => ../../extension/extensionauth
   - go.opentelemetry.io/collector/extension/extensionauth/extensionauthtest => ../../extension/extensionauth/extensionauthtest
   - go.opentelemetry.io/collector/extension/extensioncapabilities => ../../extension/extensioncapabilities
+  - go.opentelemetry.io/collector/extension/extensionlimiter => ../../extension/extensionlimiter
+  - go.opentelemetry.io/collector/extension/extensionmiddleware => ../../extension/extensionmiddleware
   - go.opentelemetry.io/collector/extension/extensiontest => ../../extension/extensiontest
   - go.opentelemetry.io/collector/extension/memorylimiterextension => ../../extension/memorylimiterextension
   - go.opentelemetry.io/collector/extension/xextension => ../../extension/xextension

cmd/otelcorecol/go.mod

@@ -4,7 +4,7 @@ module go.opentelemetry.io/collector/cmd/otelcorecol
 
 go 1.23.0
 
-toolchain go1.23.7
+toolchain go1.24.1
 
 require (
 	go.opentelemetry.io/collector/component v1.29.0
@@ -89,6 +89,7 @@ require (
 	go.opentelemetry.io/collector/config/configcompression v1.29.0 // indirect
 	go.opentelemetry.io/collector/config/configgrpc v0.123.0 // indirect
 	go.opentelemetry.io/collector/config/confighttp v0.123.0 // indirect
+	go.opentelemetry.io/collector/config/configmiddleware v0.0.0-00010101000000-000000000000 // indirect
 	go.opentelemetry.io/collector/config/confignet v1.29.0 // indirect
 	go.opentelemetry.io/collector/config/configopaque v1.29.0 // indirect
 	go.opentelemetry.io/collector/config/configretry v1.29.0 // indirect
@@ -107,6 +108,8 @@ require (
 	go.opentelemetry.io/collector/exporter/xexporter v0.123.0 // indirect
 	go.opentelemetry.io/collector/extension/extensionauth v1.29.0 // indirect
 	go.opentelemetry.io/collector/extension/extensioncapabilities v0.123.0 // indirect
+	go.opentelemetry.io/collector/extension/extensionlimiter v0.0.0-00010101000000-000000000000 // indirect
+	go.opentelemetry.io/collector/extension/extensionmiddleware v0.0.0-00010101000000-000000000000 // indirect
 	go.opentelemetry.io/collector/extension/extensiontest v0.123.0 // indirect
 	go.opentelemetry.io/collector/extension/xextension v0.123.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.29.0 // indirect
@@ -185,6 +188,8 @@ replace go.opentelemetry.io/collector/config/configgrpc => ../../config/configgr
 
 replace go.opentelemetry.io/collector/config/confighttp => ../../config/confighttp
 
+replace go.opentelemetry.io/collector/config/configmiddleware => ../../config/configmiddleware
+
 replace go.opentelemetry.io/collector/config/confignet => ../../config/confignet
 
 replace go.opentelemetry.io/collector/config/configopaque => ../../config/configopaque
@@ -251,6 +256,10 @@ replace go.opentelemetry.io/collector/extension/extensionauth/extensionauthtest
 
 replace go.opentelemetry.io/collector/extension/extensioncapabilities => ../../extension/extensioncapabilities
 
+replace go.opentelemetry.io/collector/extension/extensionlimiter => ../../extension/extensionlimiter
+
+replace go.opentelemetry.io/collector/extension/extensionmiddleware => ../../extension/extensionmiddleware
+
 replace go.opentelemetry.io/collector/extension/extensiontest => ../../extension/extensiontest
 
 replace go.opentelemetry.io/collector/extension/memorylimiterextension => ../../extension/memorylimiterextension

config/configgrpc/configgrpc.go

@@ -31,6 +31,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configauth"
 	"go.opentelemetry.io/collector/config/configcompression"
+	"go.opentelemetry.io/collector/config/configmiddleware"
 	"go.opentelemetry.io/collector/config/confignet"
 	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/config/configtls"
@@ -103,6 +104,9 @@ type ClientConfig struct {
 
 	// Auth configuration for outgoing RPCs.
 	Auth *configauth.Authentication `mapstructure:"auth,omitempty"`
+
+	// Middlewares for the gRPC client.
+	Middlewares []configmiddleware.Middleware `mapstructure:"middlewares,omitempty"`
 }
 
 // NewDefaultClientConfig returns a new instance of ClientConfig with default values.
@@ -189,6 +193,9 @@ type ServerConfig struct {
 
 	// Include propagates the incoming connection's metadata to downstream consumers.
 	IncludeMetadata bool `mapstructure:"include_metadata,omitempty"`
+
+	// Middlewares for the gRPC server.
+	Middlewares []configmiddleware.Middleware `mapstructure:"middlewares,omitempty"`
 }
 
 // NewDefaultServerConfig returns a new instance of ServerConfig with default values.
@@ -362,6 +369,15 @@ func (gcs *ClientConfig) getGrpcDialOptions(
 		)
 	}
 
+	// Apply middleware options. Note: OpenTelemetry could be registered as an extension.
+	for _, middleware := range gcs.Middlewares {
+		middlewareOptions, err := middleware.GetGRPCClientOptions(ctx, host.GetExtensions())
+		if err != nil {
+			return nil, fmt.Errorf("failed to get gRPC client options from middleware: %w", err)
+		}
+		opts = append(opts, middlewareOptions...)
+	}
+
 	for _, opt := range extraOpts {
 		if wrapper, ok := opt.(grpcDialOptionWrapper); ok {
 			opts = append(opts, wrapper.opt)
@@ -404,19 +420,20 @@ func (grpcServerOptionWrapper) isToServerOption() {}
 
 // ToServer returns a [grpc.Server] for the configuration.
 func (gss *ServerConfig) ToServer(
-	_ context.Context,
+	ctx context.Context,
 	host component.Host,
 	settings component.TelemetrySettings,
 	extraOpts ...ToServerOption,
 ) (*grpc.Server, error) {
-	grpcOpts, err := gss.getGrpcServerOptions(host, settings, extraOpts)
+	grpcOpts, err := gss.getGrpcServerOptions(ctx, host, settings, extraOpts)
 	if err != nil {
 		return nil, err
 	}
 	return grpc.NewServer(grpcOpts...), nil
 }
 
 func (gss *ServerConfig) getGrpcServerOptions(
+	ctx context.Context,
 	host component.Host,
 	settings component.TelemetrySettings,
 	extraOpts []ToServerOption,
@@ -505,6 +522,15 @@ func (gss *ServerConfig) getGrpcServerOptions(
 
 	opts = append(opts, grpc.StatsHandler(otelgrpc.NewServerHandler(otelOpts...)), grpc.ChainUnaryInterceptor(uInterceptors...), grpc.ChainStreamInterceptor(sInterceptors...))
 
+	// Apply middleware options. Note: OpenTelemetry could be registered as an extension.
+	for _, middleware := range gss.Middlewares {
+		middlewareOptions, err := middleware.GetGRPCServerOptions(ctx, host.GetExtensions())
+		if err != nil {
+			return nil, fmt.Errorf("failed to get gRPC server options from middleware: %w", err)
+		}
+		opts = append(opts, middlewareOptions...)
+	}
+
 	for _, opt := range extraOpts {
 		if wrapper, ok := opt.(grpcServerOptionWrapper); ok {
 			opts = append(opts, wrapper.opt)

config/configgrpc/configgrpc_test.go

@@ -299,7 +299,7 @@ func TestDefaultGrpcServerSettings(t *testing.T) {
 			Endpoint: "0.0.0.0:1234",
 		},
 	}
-	opts, err := gss.getGrpcServerOptions(componenttest.NewNopHost(), componenttest.NewNopTelemetrySettings(), []ToServerOption{})
+	opts, err := gss.getGrpcServerOptions(context.Background(), componenttest.NewNopHost(), componenttest.NewNopTelemetrySettings(), []ToServerOption{})
 	require.NoError(t, err)
 	assert.Len(t, opts, 3)
 }
@@ -312,6 +312,7 @@ func TestGrpcServerExtraOption(t *testing.T) {
 	}
 	extraOpt := grpc.ConnectionTimeout(1_000_000_000)
 	opts, err := gss.getGrpcServerOptions(
+		context.Background(),
 		componenttest.NewNopHost(),
 		componenttest.NewNopTelemetrySettings(),
 		[]ToServerOption{WithGrpcServerOption(extraOpt)},
@@ -401,7 +402,7 @@ func TestAllGrpcServerSettingsExceptAuth(t *testing.T) {
 			},
 		},
 	}
-	opts, err := gss.getGrpcServerOptions(componenttest.NewNopHost(), componenttest.NewNopTelemetrySettings(), []ToServerOption{})
+	opts, err := gss.getGrpcServerOptions(context.Background(), componenttest.NewNopHost(), componenttest.NewNopTelemetrySettings(), []ToServerOption{})
 	require.NoError(t, err)
 	assert.Len(t, opts, 10)
 }

config/configgrpc/go.mod

@@ -10,6 +10,7 @@ require (
 	go.opentelemetry.io/collector/component/componenttest v0.123.0
 	go.opentelemetry.io/collector/config/configauth v0.123.0
 	go.opentelemetry.io/collector/config/configcompression v1.29.0
+	go.opentelemetry.io/collector/config/configmiddleware v0.0.0-00010101000000-000000000000
 	go.opentelemetry.io/collector/config/confignet v1.29.0
 	go.opentelemetry.io/collector/config/configopaque v1.29.0
 	go.opentelemetry.io/collector/config/configtls v1.29.0
@@ -39,6 +40,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/extension/extensionmiddleware v0.0.0-00010101000000-000000000000 // indirect
 	go.opentelemetry.io/collector/featuregate v1.29.0 // indirect
 	go.opentelemetry.io/collector/internal/telemetry v0.123.0 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.123.0 // indirect
@@ -64,6 +66,8 @@ replace go.opentelemetry.io/collector/config/configauth => ../configauth
 
 replace go.opentelemetry.io/collector/config/configcompression => ../configcompression
 
+replace go.opentelemetry.io/collector/config/configmiddleware => ../configmiddleware
+
 replace go.opentelemetry.io/collector/config/confignet => ../confignet
 
 replace go.opentelemetry.io/collector/config/configopaque => ../configopaque
@@ -74,6 +78,8 @@ replace go.opentelemetry.io/collector/extension => ../../extension
 
 replace go.opentelemetry.io/collector/extension/extensionauth => ../../extension/extensionauth
 
+replace go.opentelemetry.io/collector/extension/extensionmiddleware => ../../extension/extensionmiddleware
+
 replace go.opentelemetry.io/collector/pdata => ../../pdata
 
 replace go.opentelemetry.io/collector/pdata/testdata => ../../pdata/testdata

config/confighttp/confighttp.go

@@ -27,6 +27,7 @@
 	"go.opentelemetry.io/collector/config/configauth"
 	"go.opentelemetry.io/collector/config/configcompression"
 	"go.opentelemetry.io/collector/config/confighttp/internal"
+	"go.opentelemetry.io/collector/config/configmiddleware"
 	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/extension/extensionauth"
@@ -111,6 +112,11 @@
 	HTTP2PingTimeout time.Duration `mapstructure:"http2_ping_timeout,omitempty"`
 	// Cookies configures the cookie management of the HTTP client.
 	Cookies *CookiesConfig `mapstructure:"cookies,omitempty"`
+
+	// Middlewares are used to add custom functionality to the HTTP client.
+	// Middleware handlers are applied in the order they appear in this list,
+	// with the first middleware becoming the outermost handler.
+	Middlewares []configmiddleware.Middleware `mapstructure:"middleware,omitempty"`
 }
 
 // CookiesConfig defines the configuration of the HTTP client regarding cookies served by the server.
@@ -194,6 +200,21 @@
 
 	clientTransport := (http.RoundTripper)(transport)
 
+	// Apply middlewares in reverse order so they are applied in
+	// order. The first middleware runs after authentication.
+	for i := len(hcs.Middlewares) - 1; i >= 0; i-- {
+		wrapper, err := hcs.Middlewares[i].GetHTTPClientRoundTripper(ctx, host.GetExtensions())
+		// If we failed to get the middleware
+		if err != nil {
+			return nil, err
+		}
+		clientTransport, err = wrapper(clientTransport)
+		// If we failed to construct a wrapper
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	// The Auth RoundTripper should always be the innermost to ensure that
 	// request signing-based auth mechanisms operate after compression
 	// and header middleware modifies the request
@@ -338,6 +359,11 @@
 	// is zero, the value of ReadTimeout is used. If both are
 	// zero, there is no timeout.
 	IdleTimeout time.Duration `mapstructure:"idle_timeout"`
+
+	// Middlewares are used to add custom functionality to the HTTP server.
+	// Middleware handlers are applied in the order they appear in this list,
+	// with the first middleware becoming the outermost handler.
+	Middlewares []configmiddleware.Middleware `mapstructure:"middleware,omitempty"`
 }
 
 // NewDefaultServerConfig returns ServerConfig type object with default values.
@@ -411,7 +437,7 @@
 }
 
 // ToServer creates an http.Server from settings object.
-func (hss *ServerConfig) ToServer(_ context.Context, host component.Host, settings component.TelemetrySettings, handler http.Handler, opts ...ToServerOption) (*http.Server, error) {
+func (hss *ServerConfig) ToServer(ctx context.Context, host component.Host, settings component.TelemetrySettings, handler http.Handler, opts ...ToServerOption) (*http.Server, error) {
 	serverOpts := &toServerOptions{}
 	serverOpts.Apply(opts...)
 
@@ -423,6 +449,22 @@
 		hss.CompressionAlgorithms = defaultCompressionAlgorithms
 	}
 
+	// Apply middlewares in reverse order so they are applied in
+	// order.  The first middleware runs after decompression,
+	// below, preceded by Auth, CORS, etc.
+	for i := len(hss.Middlewares) - 1; i >= 0; i-- {
+		wrapper, err := hss.Middlewares[i].GetHTTPServerHandler(ctx, host.GetExtensions())
+		// If we failed to get the middleware
+		if err != nil {
+			return nil, err
+		}
+		handler, err = wrapper(handler)
+		// If we failed to construct a wrapper
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	handler = httpContentDecompressor(
 		handler,
 		hss.MaxRequestBodySize,

config/confighttp/go.mod

@@ -13,6 +13,7 @@ require (
 	go.opentelemetry.io/collector/component/componenttest v0.123.0
 	go.opentelemetry.io/collector/config/configauth v0.123.0
 	go.opentelemetry.io/collector/config/configcompression v1.29.0
+	go.opentelemetry.io/collector/config/configmiddleware v0.0.0-00010101000000-000000000000
 	go.opentelemetry.io/collector/config/configopaque v1.29.0
 	go.opentelemetry.io/collector/config/configtls v1.29.0
 	go.opentelemetry.io/collector/extension v1.29.0
@@ -36,6 +37,7 @@ require (
 	github.com/hashicorp/go-version v1.7.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/extension/extensionmiddleware v0.0.0-00010101000000-000000000000 // indirect
 	go.opentelemetry.io/collector/featuregate v1.29.0 // indirect
 	go.opentelemetry.io/collector/internal/telemetry v0.123.0 // indirect
 	go.opentelemetry.io/collector/pdata v1.29.0 // indirect
@@ -58,6 +60,8 @@ replace go.opentelemetry.io/collector/config/configauth => ../configauth
 
 replace go.opentelemetry.io/collector/config/configcompression => ../configcompression
 
+replace go.opentelemetry.io/collector/config/configmiddleware => ../configmiddleware
+
 replace go.opentelemetry.io/collector/config/configopaque => ../configopaque
 
 replace go.opentelemetry.io/collector/config/configtls => ../configtls
@@ -66,6 +70,8 @@ replace go.opentelemetry.io/collector/extension => ../../extension
 
 replace go.opentelemetry.io/collector/extension/extensionauth => ../../extension/extensionauth
 
+replace go.opentelemetry.io/collector/extension/extensionmiddleware => ../../extension/extensionmiddleware
+
 replace go.opentelemetry.io/collector/pdata => ../../pdata
 
 replace go.opentelemetry.io/collector/component => ../../component

config/confighttp/xconfighttp/go.mod

@@ -30,9 +30,11 @@ require (
 	go.opentelemetry.io/collector/component v1.29.0 // indirect
 	go.opentelemetry.io/collector/config/configauth v0.123.0 // indirect
 	go.opentelemetry.io/collector/config/configcompression v1.29.0 // indirect
+	go.opentelemetry.io/collector/config/configmiddleware v0.0.0-00010101000000-000000000000 // indirect
 	go.opentelemetry.io/collector/config/configopaque v1.29.0 // indirect
 	go.opentelemetry.io/collector/config/configtls v1.29.0 // indirect
 	go.opentelemetry.io/collector/extension/extensionauth v1.29.0 // indirect
+	go.opentelemetry.io/collector/extension/extensionmiddleware v0.0.0-00010101000000-000000000000 // indirect
 	go.opentelemetry.io/collector/featuregate v1.29.0 // indirect
 	go.opentelemetry.io/collector/internal/telemetry v0.123.0 // indirect
 	go.opentelemetry.io/collector/pdata v1.29.0 // indirect
@@ -83,3 +85,7 @@ replace go.opentelemetry.io/collector/pipeline => ../../../pipeline
 replace go.opentelemetry.io/collector/internal/telemetry => ../../../internal/telemetry
 
 replace go.opentelemetry.io/collector/featuregate => ../../../featuregate
+
+replace go.opentelemetry.io/collector/extension/extensionmiddleware => ../../../extension/extensionmiddleware
+
+replace go.opentelemetry.io/collector/config/configmiddleware => ../../configmiddleware

config/configlimiter/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

config/configlimiter/configlimiter.go

@@ -0,0 +1,42 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+// Package configlimiter implements the configuration settings to
+// apply rate limiting on incoming requests, and allows
+// components to configure rate limiting behavior.
+package configlimiter // import "go.opentelemetry.io/collector/config/configlimiter"
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/extension/extensionlimiter"
+)
+
+var (
+	errLimiterNotFound = errors.New("limiter not found")
+	errNotLimiter      = errors.New("requested extension is not a limiter")
+)
+
+// Limiter defines the rate limiting settings for a component.
+type Limiter struct {
+	// LimiterID specifies the name of the extension to use in order to apply rate limiting.
+	LimiterID component.ID `mapstructure:"limiter,omitempty"`
+}
+
+// GetProvider attempts to select the appropriate extensionlimiter.Provider from the list of extensions,
+// based on the requested extension name. If a limiter is not found, an error is returned.
+// Callers will use the returned Provider to get access to the specific rate- and
+// resource-limiter weights they are capable of limiting.
+func (l Limiter) GetProvider(_ context.Context, extensions map[component.ID]component.Component) (extensionlimiter.Provider, error) {
+	if ext, found := extensions[l.LimiterID]; found {
+		if limiter, ok := ext.(extensionlimiter.Provider); ok {
+			return limiter, nil
+		}
+		return nil, errNotLimiter
+	}
+
+	return nil, fmt.Errorf("failed to resolve limiter provider %q: %w", l.LimiterID, errLimiterNotFound)
+}