[config/confighttp] add memorylimiterextension to HTTPServerConfig

.chloggen/timn_memorylimiterextension-HTTPServerConfig.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: confighttp
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: add an option to setup MemoryLimiter extension in to HTTPServerConfig
+
+# One or more tracking issues or pull requests related to the change
+issues: [8632]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

cmd/builder/test/core.builder.yaml

@@ -44,6 +44,7 @@ replaces:
   - go.opentelemetry.io/collector/exporter/loggingexporter => ${WORKSPACE_DIR}/exporter/loggingexporter
   - go.opentelemetry.io/collector/extension => ${WORKSPACE_DIR}/extension
   - go.opentelemetry.io/collector/extension/auth => ${WORKSPACE_DIR}/extension/auth
+  - go.opentelemetry.io/collector/extension/memorylimiterextension => ${WORKSPACE_DIR}/extension/memorylimiterextension
   - go.opentelemetry.io/collector/extension/zpagesextension => ${WORKSPACE_DIR}/extension/zpagesextension
   - go.opentelemetry.io/collector/featuregate => ${WORKSPACE_DIR}/featuregate
   - go.opentelemetry.io/collector/otelcol => ${WORKSPACE_DIR}/otelcol

config/confighttp/README.md

@@ -78,6 +78,8 @@ will not be enabled.
 - [`tls`](../configtls/README.md)
 - [`auth`](../configauth/README.md)
 
+`memory_limiter`: [Memory limiter extension](../../extension/memorylimiterextension/README.md) that will reject incoming requests once the memory utilization grows above configured limits.
+
 You can enable [`attribute processor`][attribute-processor] to append any http header to span's attribute using custom key. You also need to enable the "include_metadata"
 
 Example:

config/confighttp/confighttp.go

@@ -274,6 +274,9 @@ type ServerConfig struct {
 	// Additional headers attached to each HTTP response sent to the client.
 	// Header values are opaque since they may be sensitive.
 	ResponseHeaders map[string]configopaque.String `mapstructure:"response_headers"`
+
+	// MemoryLimiter is memory limiter this receiver will use to restrict incoming requests
+	MemoryLimiter *component.ID `mapstructure:"memory_limiter"`
 }
 
 // ToListener creates a net.Listener.
@@ -334,6 +337,15 @@ func (hss *ServerConfig) ToServer(host component.Host, settings component.Teleme
 		o(serverOpts)
 	}
 
+	if hss.MemoryLimiter != nil {
+		ml, err := getMemoryLimiterExtension(hss.MemoryLimiter, host.GetExtensions())
+		if err != nil {
+			return nil, err
+		}
+
+		handler = memoryLimiterInterceptor(handler, ml)
+	}
+
 	handler = httpContentDecompressor(handler, serverOpts.errHandler, serverOpts.decoders)
 
 	if hss.MaxRequestBodySize > 0 {
@@ -442,3 +454,26 @@ func maxRequestBodySizeInterceptor(next http.Handler, maxRecvSize int64) http.Ha
 		next.ServeHTTP(w, r)
 	})
 }
+
+func memoryLimiterInterceptor(next http.Handler, ml memoryLimiterExtension) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if ml.MustRefuse() {
+			http.Error(w, http.StatusText(http.StatusTooManyRequests), http.StatusTooManyRequests)
+			return
+		}
+		next.ServeHTTP(w, r)
+	})
+}
+
+type memoryLimiterExtension = interface{ MustRefuse() bool }
+
+func getMemoryLimiterExtension(extID *component.ID, extensions map[component.ID]component.Component) (memoryLimiterExtension, error) {
+	if ext, found := extensions[*extID]; found {
+		if server, ok := ext.(memoryLimiterExtension); ok {
+			return server, nil
+		}
+		return nil, fmt.Errorf("requested MemoryLimiter, %s, is not a memoryLimiterExtension", extID)
+	}
+
+	return nil, fmt.Errorf("failed to resolve memoryLimiterExtension %q: %s", extID, "memory limiter extension not found")
+}

config/confighttp/confighttp_test.go

@@ -31,6 +31,7 @@ import (
 	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/extension/auth"
 	"go.opentelemetry.io/collector/extension/auth/authtest"
+	"go.opentelemetry.io/collector/extension/extensiontest"
 )
 
 type customRoundTripper struct {
@@ -1302,11 +1303,84 @@ func TestServerWithDecoder(t *testing.T) {
 
 }
 
+func TestMemoryLimiterInterceptor(t *testing.T) {
+	mlID := component.NewID("memoryLimiter")
+	ml := &mockMemoryLimiterExtension{}
+	host := &mockHost{
+		ext: map[component.ID]component.Component{
+			mlID: ml,
+		},
+	}
+
+	hss := ServerConfig{
+		Endpoint:      "localhost:0",
+		MemoryLimiter: &mlID,
+	}
+
+	handlerCalled := false
+	handler := http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {
+		handlerCalled = true
+	})
+
+	srv, err := hss.ToServer(host, componenttest.NewNopTelemetrySettings(), handler)
+	require.NoError(t, err)
+
+	// test 429 status
+	ml.mustRefuse = true
+	response := &httptest.ResponseRecorder{}
+	srv.Handler.ServeHTTP(response, httptest.NewRequest("GET", "/", nil))
+	assert.False(t, handlerCalled)
+	assert.Equal(t, http.StatusTooManyRequests, response.Result().StatusCode)
+
+	// test 200 status
+	ml.mustRefuse = false
+	response = &httptest.ResponseRecorder{}
+	srv.Handler.ServeHTTP(response, httptest.NewRequest("GET", "/", nil))
+	assert.True(t, handlerCalled)
+	assert.Equal(t, http.StatusOK, response.Result().StatusCode)
+}
+
+func TestGetMemoryLimiterExtension(t *testing.T) {
+	mle := &mockMemoryLimiterExtension{}
+	nopExt, err := extensiontest.NewNopFactory().CreateExtension(context.Background(), extensiontest.NewNopCreateSettings(), CORSConfig{})
+	assert.NoError(t, err)
+
+	extList := map[component.ID]component.Component{
+		component.NewID("memoryLimiter"):     mle,
+		component.NewID("memoryLimiterFake"): nopExt,
+	}
+
+	// valid extension
+	comID := component.NewID("memoryLimiter")
+	_, err = getMemoryLimiterExtension(&comID, extList)
+	assert.NoError(t, err)
+
+	// invalid extension
+	comID = component.NewID("memoryLimiterFake")
+	_, err = getMemoryLimiterExtension(&comID, extList)
+	assert.EqualError(t, err, "requested MemoryLimiter, memoryLimiterFake, is not a memoryLimiterExtension")
+
+	// not found
+	comID = component.NewID("notfound")
+	_, err = getMemoryLimiterExtension(&comID, extList)
+	assert.EqualError(t, err, "failed to resolve memoryLimiterExtension \"notfound\": memory limiter extension not found")
+}
+
 type mockHost struct {
 	component.Host
 	ext map[component.ID]component.Component
 }
 
+type mockMemoryLimiterExtension struct {
+	mustRefuse bool
+	component.StartFunc
+	component.ShutdownFunc
+}
+
+func (mml *mockMemoryLimiterExtension) MustRefuse() bool {
+	return mml.mustRefuse
+}
+
 func (nh *mockHost) GetExtensions() map[component.ID]component.Component {
 	return nh.ext
 }