Include request headers in client.Info

CHANGELOG.md

@@ -23,6 +23,7 @@
     in a future release.
   - `service.telemetry.metrics.level` and `service.telemetry.metrics.address`
     should be used to configure collector self-metrics.
+  - `confighttp` and `configgrpc`: New config option `include_metadata` to persist request metadata/headers in `client.Info.Metadata` (#4547)
 
 ## 🧰 Bug fixes 🧰
 

client/client.go

@@ -106,6 +106,9 @@ type Info struct {
 	// configauth.ServerAuthenticator implementations tied to the receiver for
 	// this connection.
 	Auth AuthData
+
+	// Metadata is the request metadata from the client connecting to this connector.
+	Metadata map[string][]string
 }
 
 // AuthData represents the authentication data as seen by authenticators tied to

config/configgrpc/configgrpc.go

@@ -169,6 +169,9 @@ type GRPCServerSettings struct {
 
 	// Auth for this receiver
 	Auth *configauth.Authentication `mapstructure:"auth,omitempty"`
+
+	// Include propagates the incoming connection's metadata to downstream consumers.
+	IncludeMetadata bool `mapstructure:"include_metadata,omitempty"`
 }
 
 // SanitizedEndpoint strips the prefix of either http:// or https:// from configgrpc.GRPCClientSettings.Endpoint.
@@ -365,8 +368,8 @@ func (gss *GRPCServerSettings) ToServerOption(host component.Host, settings comp
 		otelgrpc.WithPropagators(otel.GetTextMapPropagator()),
 	))
 
-	uInterceptors = append(uInterceptors, enhanceWithClientInformation)
-	sInterceptors = append(sInterceptors, enhanceStreamWithClientInformation)
+	uInterceptors = append(uInterceptors, enhanceWithClientInformation(gss.IncludeMetadata))
+	sInterceptors = append(sInterceptors, enhanceStreamWithClientInformation(gss.IncludeMetadata))
 
 	opts = append(opts, grpc.ChainUnaryInterceptor(uInterceptors...), grpc.ChainStreamInterceptor(sInterceptors...))
 
@@ -385,23 +388,32 @@ func GetGRPCCompressionKey(compressionType string) string {
 
 // enhanceWithClientInformation intercepts the incoming RPC, replacing the incoming context with one that includes
 // a client.Info, potentially with the peer's address.
-func enhanceWithClientInformation(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
-	return handler(contextWithClient(ctx), req)
+func enhanceWithClientInformation(includeMetadata bool) func(ctx context.Context, req interface{}, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+	return func(ctx context.Context, req interface{}, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+		return handler(contextWithClient(ctx, includeMetadata), req)
+	}
 }
 
-func enhanceStreamWithClientInformation(srv interface{}, ss grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
-	wrapped := middleware.WrapServerStream(ss)
-	wrapped.WrappedContext = contextWithClient(ss.Context())
-	return handler(srv, wrapped)
+func enhanceStreamWithClientInformation(includeMetadata bool) func(srv interface{}, ss grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+	return func(srv interface{}, ss grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		wrapped := middleware.WrapServerStream(ss)
+		wrapped.WrappedContext = contextWithClient(ss.Context(), includeMetadata)
+		return handler(srv, wrapped)
+	}
 }
 
 // contextWithClient attempts to add the peer address to the client.Info from the context. When no
 // client.Info exists in the context, one is created.
-func contextWithClient(ctx context.Context) context.Context {
+func contextWithClient(ctx context.Context, includeMetadata bool) context.Context {
 	cl := client.FromContext(ctx)
 	if p, ok := peer.FromContext(ctx); ok {
 		cl.Addr = p.Addr
 	}
+	if includeMetadata {
+		if md, ok := metadata.FromIncomingContext(ctx); ok {
+			cl.Metadata = md.Copy()
+		}
+	}
 	return client.NewContext(ctx, cl)
 }
 

config/configgrpc/configgrpc_test.go

@@ -572,9 +572,10 @@ func TestReceiveOnUnixDomainSocket(t *testing.T) {
 
 func TestContextWithClient(t *testing.T) {
 	testCases := []struct {
-		desc     string
-		input    context.Context
-		expected client.Info
+		desc       string
+		input      context.Context
+		doMetadata bool
+		expected   client.Info
 	}{
 		{
 			desc:     "no peer information, empty client",
@@ -624,10 +625,38 @@ func TestContextWithClient(t *testing.T) {
 				},
 			},
 		},
+		{
+			desc: "existing client with metadata, no metadata processing",
+			input: client.NewContext(context.Background(), client.Info{
+				Metadata: map[string][]string{"test-metadata-key": {"test-value"}},
+			}),
+			expected: client.Info{},
+		},
+		{
+			desc: "existing client with metadata",
+			input: client.NewContext(context.Background(), client.Info{
+				Metadata: map[string][]string{"test-metadata-key": {"test-value"}},
+			}),
+			doMetadata: true,
+			expected: client.Info{
+				Metadata: map[string][]string{"test-metadata-key": {"test-value"}},
+			},
+		},
+		{
+			desc: "existing client with metadata in context",
+			input: metadata.NewIncomingContext(
+				client.NewContext(context.Background(), client.Info{}),
+				metadata.Pairs("test-metadata-key", "test-value"),
+			),
+			doMetadata: true,
+			expected: client.Info{
+				Metadata: map[string][]string{"test-metadata-key": {"test-value"}},
+			},
+		},
 	}
 	for _, tC := range testCases {
 		t.Run(tC.desc, func(t *testing.T) {
-			cl := client.FromContext(contextWithClient(tC.input))
+			cl := client.FromContext(contextWithClient(tC.input, tC.doMetadata))
 			assert.Equal(t, tC.expected, cl)
 		})
 	}
@@ -650,7 +679,7 @@ func TestStreamInterceptorEnhancesClient(t *testing.T) {
 	}
 
 	// test
-	err := enhanceStreamWithClientInformation(nil, stream, nil, handler)
+	err := enhanceStreamWithClientInformation(false)(nil, stream, nil, handler)
 
 	// verify
 	assert.NoError(t, err)

config/confighttp/clientinfohandler.go

@@ -27,25 +27,32 @@ var _ http.Handler = (*clientInfoHandler)(nil)
 // clientInfoHandler is an http.Handler that enhances the incoming request context with client.Info.
 type clientInfoHandler struct {
 	next http.Handler
+
+	// include client metadata or not
+	includeMetadata bool
 }
 
 // ServeHTTP intercepts incoming HTTP requests, replacing the request's context with one that contains
 // a client.Info containing the client's IP address.
 func (h *clientInfoHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
-	req = req.WithContext(contextWithClient(req))
+	req = req.WithContext(contextWithClient(req, h.includeMetadata))
 	h.next.ServeHTTP(w, req)
 }
 
 // contextWithClient attempts to add the client IP address to the client.Info from the context. When no
 // client.Info exists in the context, one is created.
-func contextWithClient(req *http.Request) context.Context {
+func contextWithClient(req *http.Request, includeMetadata bool) context.Context {
 	cl := client.FromContext(req.Context())
 
 	ip := parseIP(req.RemoteAddr)
 	if ip != nil {
 		cl.Addr = ip
 	}
 
+	if includeMetadata {
+		cl.Metadata = req.Header.Clone()
+	}
+
 	ctx := client.NewContext(req.Context(), cl)
 	return ctx
 }

config/confighttp/confighttp.go

@@ -199,6 +199,9 @@ type HTTPServerSettings struct {
 
 	// Auth for this receiver
 	Auth *configauth.Authentication `mapstructure:"auth,omitempty"`
+
+	// Include propagates the client metadata from the incoming requests to the downstream consumers
+	IncludeMetadata bool `mapstructure:"include_metadata,omitempty"`
 }
 
 // ToListener creates a net.Listener.
@@ -284,7 +287,8 @@ func (hss *HTTPServerSettings) ToServer(host component.Host, settings component.
 
 	// wrap the current handler in an interceptor that will add client.Info to the request's context
 	handler = &clientInfoHandler{
-		next: handler,
+		next:            handler,
+		includeMetadata: hss.IncludeMetadata,
 	}
 
 	return &http.Server{

config/confighttp/confighttp_test.go

@@ -738,17 +738,18 @@ func TestHttpHeaders(t *testing.T) {
 
 func TestContextWithClient(t *testing.T) {
 	testCases := []struct {
-		desc     string
-		input    *http.Request
-		expected client.Info
+		desc       string
+		input      *http.Request
+		doMetadata bool
+		expected   client.Info
 	}{
 		{
-			desc:     "request without client IP",
+			desc:     "request without client IP or headers",
 			input:    &http.Request{},
 			expected: client.Info{},
 		},
 		{
-			desc: "request without client IP",
+			desc: "request with client IP",
 			input: &http.Request{
 				RemoteAddr: "1.2.3.4:55443",
 			},
@@ -758,10 +759,28 @@ func TestContextWithClient(t *testing.T) {
 				},
 			},
 		},
+		{
+			desc: "request with client headers, no metadata processing",
+			input: &http.Request{
+				Header: map[string][]string{"x-test-header": {"test-value"}},
+			},
+			doMetadata: false,
+			expected:   client.Info{},
+		},
+		{
+			desc: "request with client headers",
+			input: &http.Request{
+				Header: map[string][]string{"x-test-header": {"test-value"}},
+			},
+			doMetadata: true,
+			expected: client.Info{
+				Metadata: map[string][]string{"x-test-header": {"test-value"}},
+			},
+		},
 	}
 	for _, tC := range testCases {
 		t.Run(tC.desc, func(t *testing.T) {
-			ctx := contextWithClient(tC.input)
+			ctx := contextWithClient(tC.input, tC.doMetadata)
 			assert.Equal(t, tC.expected, client.FromContext(ctx))
 		})
 	}