Limiter extension interface for memorylimiterextension draft

extension/memorylimiterextension/go.mod

@@ -37,6 +37,7 @@ require (
 	github.com/tklauser/numcpus v0.6.1 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/extension/xextension v0.121.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.27.0 // indirect
 	go.opentelemetry.io/collector/pdata v1.27.0 // indirect
 	go.opentelemetry.io/otel v1.34.0 // indirect
@@ -68,4 +69,6 @@ replace go.opentelemetry.io/collector/internal/memorylimiter => ../../internal/m
 
 replace go.opentelemetry.io/collector/extension/extensiontest => ../../extension/extensiontest
 
+replace go.opentelemetry.io/collector/extension/xextension => ../../extension/xextension
+
 replace go.opentelemetry.io/collector/featuregate => ../../featuregate

extension/memorylimiterextension/memorylimiter.go

@@ -9,13 +9,16 @@ import (
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/extension/xextension/limit"
 	"go.opentelemetry.io/collector/internal/memorylimiter"
 )
 
 type memoryLimiterExtension struct {
 	memLimiter *memorylimiter.MemoryLimiter
 }
 
+var _ limit.Extension = &memoryLimiterExtension{}
+
 // newMemoryLimiter returns a new memorylimiter extension.
 func newMemoryLimiter(cfg *Config, logger *zap.Logger) (*memoryLimiterExtension, error) {
 	ml, err := memorylimiter.NewMemoryLimiter(cfg, logger)
@@ -35,6 +38,28 @@ func (ml *memoryLimiterExtension) Shutdown(ctx context.Context) error {
 }
 
 // MustRefuse returns if the caller should deny because memory has reached it's configured limits
+//
+// It's not clear that this is used anywhere, but as a legacy exported
+// function some component could observe it disappear, so it has to stay.
 func (ml *memoryLimiterExtension) MustRefuse() bool {
 	return ml.memLimiter.MustRefuse()
 }
+
+// client implements Client. Note that the component Kind and ID are
+// not used, so this struct is identical to *memoryLimiterExtension.
+type client struct {
+	ml *memoryLimiterExtension
+}
+
+// GetClient implements limit.Extension.
+func (ml *memoryLimiterExtension) GetClient(_ context.Context, _ component.Kind, _ component.ID) (limit.Client, error) {
+	return &client{ml}, nil
+}
+
+// Acquire implements limit.Client.
+func (c *client) Acquire(_ context.Context, _ uint64) (limit.ReleaseFunc, error) {
+	if c.ml.memLimiter.MustRefuse() {
+		return nil, memorylimiter.ErrDataRefused
+	}
+	return func() {}, nil
+}

extension/memorylimiterextension/memorylimiter_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/internal/memorylimiter"
 	"go.opentelemetry.io/collector/internal/memorylimiter/iruntime"
@@ -91,6 +92,21 @@ func TestMemoryPressureResponse(t *testing.T) {
 			} else {
 				require.NoError(t, err)
 			}
+
+			// Use the limit extension API.
+			lclient, err := ml.GetClient(ctx, component.KindReceiver, component.MustNewID("testing"))
+			assert.NoError(t, err)
+			nf, err := lclient.Acquire(ctx, 1)
+
+			if tt.expectError {
+				assert.Error(t, err)
+				assert.Nil(t, nf)
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, nf)
+				nf()
+			}
+
 			assert.NoError(t, ml.Shutdown(ctx))
 		})
 	}

extension/xextension/go.mod

@@ -3,12 +3,15 @@ module go.opentelemetry.io/collector/extension/xextension
 go 1.23.0
 
 require (
+	github.com/stretchr/testify v1.10.0
 	go.opentelemetry.io/collector/component v1.27.0
 	go.opentelemetry.io/collector/extension v1.27.0
 )
 
 require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/collector/pdata v1.27.0 // indirect
 	go.opentelemetry.io/otel v1.34.0 // indirect
 	go.opentelemetry.io/otel/metric v1.34.0 // indirect
@@ -21,6 +24,7 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f // indirect
 	google.golang.org/grpc v1.71.0 // indirect
 	google.golang.org/protobuf v1.36.5 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
 replace go.opentelemetry.io/collector/extension => ../

extension/xextension/limit/README.md

@@ -0,0 +1,48 @@
+# Storage
+
+**Status: under development**
+
+A limit extension supports flexible methods for admission and rate
+control.  Other components, typically receivers, can request a limit
+client from the limit extension and use it to admit requests.
+
+This interface is byte-weight oriented, enabling limiting for
+individual payloads within a stream-oriented RPC.  For request-level
+limiting, consider using the Auth extension instead.
+
+The `limit.Extension` interface extends `component.Extension` by
+adding the following method:
+
+```
+GetClient(context.Context, component.Kind, component.ID, string) (Client, error)
+```
+
+After the context argument are two component-level identifiers and a
+string, which is the name of the extension.  Typically, components
+will support a list of named limiters to apply in sequence, e.g.,
+
+```
+receivers:
+  someprotocol:
+    limiters:
+	- rate
+	- memory
+```
+
+The `limit.Client` interface contains the following method:
+```
+Acquire(ctx context.Context, weight uint64) (ReleaseFunc, error)
+```
+
+The weight parameter specifies how large of an request to consider in
+the limiter, which should estimate the amount of real memory used by
+to represent the data after it is uncompressed.
+
+The result is either a non-nil release function and nil error, or a
+nil release function and non-nil error.  The limiter may block or fail
+fast, at its discretion.  When a non-nil release function is returned,
+the component is responsible for calling the release function after
+the memory is no longer used.
+
+Note: It is the responsibility of each component to `Close` a storage
+client that it has requested.

extension/xextension/limit/doc.go

@@ -0,0 +1,6 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+// Package limit implements an extension that can limit requests by
+// blocking or failing them through calls to acquire and release.
+package limit // import "go.opentelemetry.io/collector/extension/xextension/limit"

extension/xextension/limit/limit.go

@@ -0,0 +1,70 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package limit // import "go.opentelemetry.io/collector/extension/xextension/limit"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/extension"
+)
+
+// Extension is the interface that storage extensions must implement
+type Extension interface {
+	extension.Extension
+
+	// GetClient will create a client for use by the specified
+	// component.  The component can use the client to limit
+	// admission to the pipeline.
+	GetClient(ctx context.Context, kind component.Kind, id component.ID) (Client, error)
+}
+
+// Client implements a limiter for byte-weighted request admission.
+type Client interface {
+	// Acquire asks the controller to admit the caller.
+	//
+	// The weight parameter specifies how large of an request to
+	// consider in the limiter.  This should be a measure of the
+	// size of the request after it is uncompressed.  One way to
+	// compute the weight of a request is byte its OTLP encoding
+	// size using the appropriate pdata `Sizer` type.  However,
+	// components are encouraged to use an approximation, not
+	// necessarily based on the OTLP encoding size, if there is an
+	// efficient substitute.
+	//
+	// The goal is for the estimated weight to approximate the
+	// amount of real memory that will be occupied while the
+	// request is in flight.  If the component uses a protocol
+	// with less repetition than OTLP, it is possible for the
+	// Sizer to overestimate the amount of real memory used,
+	// because Golang's immutable string value allows
+	// it. Therefore, components should prefer an inexpensive,
+	// approximate method to determine weight in bytes.
+	//
+	// The limiter is permitted to block the request. After making
+	// its decision, the return value is exclusively a function
+	// value or an error. Admit will return when one of the
+	// following events occurs:
+	//
+	//   (1) admission is allowed, or
+	//   (2) the provided ctx becomes canceled, or
+	//   (3) the limiter determines that the request should fail.
+	//
+	// In case (1), the return value will be a non-nil
+	// ReleaseFunc. The caller must invoke it after it is finished
+	// with the resource being guarded by the admission
+	// controller.
+	//
+	// In case (2), the return value should be similar to:
+	// - gRPC: Cancelled or DeadlineExceeded error
+	// - HTTP: Status 408.
+	//
+	// In case (3), the return value should be similar to:
+	// - gRPC: ResourceExhausted error
+	// - HTTP: Status 429.
+	Acquire(ctx context.Context, weight uint64) (ReleaseFunc, error)
+}
+
+// ReleaseFunc is returned by Acquire when the Acquire() was admitted.
+type ReleaseFunc func()

extension/xextension/limit/limit_test.go

@@ -0,0 +1,19 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package limit
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNopLimiter(t *testing.T) {
+	ctx := context.Background()
+	nop := NewNopClient()
+	rel, err := nop.Acquire(ctx, 1000)
+	require.NoError(t, err)
+	rel()
+}

extension/xextension/limit/metadata.yaml

@@ -0,0 +1,10 @@
+type: xextension
+github_project: open-telemetry/opentelemetry-collector
+
+status:
+  class: extension
+  codeowners:
+    active:
+      - jmacd
+  stability:
+    development: [traces, metrcs, logs, profiles]

extension/xextension/limit/nop_client.go

@@ -0,0 +1,20 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package limit // import "go.opentelemetry.io/collector/extension/xextension/limit"
+
+import "context"
+
+type nopClient struct{}
+
+var nopClientInstance Client = &nopClient{}
+
+// NewNopClient returns a nop client
+func NewNopClient() Client {
+	return nopClientInstance
+}
+
+// Acquire implements Client.
+func (nopClient) Acquire(_ context.Context, _ uint64) (ReleaseFunc, error) {
+	return func() {}, nil
+}