Add attributes to the RequestContext

filter/request.go

@@ -7,14 +7,20 @@ import (
 	"strconv"
 	"strings"
 	"time"
+
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
+// From https://github.com/envoyproxy/envoy/blob/e8d5030589fd725f879b17536315460eec302f07/source/extensions/filters/http/ext_proc/ext_proc.cc#L51
+const extProcFilterName = "envoy.filters.http.ext_proc"
+
 // RequestContext stores the context between the different gRPC messages received from Envoy and it is used to store the request headers, response headers, and other information about the request.
 // The Process method should be called on every message received from Envoy in order to update the request object.
 // Note that the request object is not thread-safe and should not be shared between goroutines.
 type RequestContext struct {
 	RequestHeaders  http.Header
 	ResponseHeaders http.Header
+	Attributes      map[string]*structpb.Struct
 	url             *url.URL
 	cookies         []*http.Cookie
 	status          int
@@ -65,6 +71,23 @@ func (r *RequestContext) ResponseHeaderValues(key string) []string {
 	return r.ResponseHeaders.Values(key)
 }
 
+// Attribute returns the value of the given Envoy attribute, e.g. Attribute("source.address").
+// It is only populated when the Envoy ext_proc filter is configured with a matching entry in
+// request_attributes or response_attributes.
+// See https://www.envoyproxy.io/docs/envoy/latest/intro/arch_overview/advanced/attributes#arch-overview-attributes for the list of available attributes.
+func (r *RequestContext) Attribute(key string) (*structpb.Value, bool) {
+	// Envoy uses the hardcoded filter name for the attributes map it sends to the ext_proc filter.
+	// This key isn't documented. Behavior and value taken from the codebase.
+	// This is not pretty but envoy does not give use a better option: we could flatten the map
+	// or take a "first" element - but that's seems to be even less future-proof.
+	ns, ok := r.Attributes[extProcFilterName]
+	if !ok {
+		return nil, false
+	}
+	v, ok := ns.GetFields()[key]
+	return v, ok
+}
+
 // Scheme returns the scheme of the request (http or https)
 func (r *RequestContext) Scheme() string {
 	return r.RequestHeader(":scheme")

filter/request_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/getyourguide/extproc-go/filter"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 const (
@@ -105,6 +106,34 @@ func TestMetadata(t *testing.T) {
 	})
 }
 
+func TestAttributes(t *testing.T) {
+	t.Run("read existing attribute", func(t *testing.T) {
+		req := filter.NewRequestContext()
+		req.Attributes = map[string]*structpb.Struct{
+			"envoy.filters.http.ext_proc": {
+				Fields: map[string]*structpb.Value{
+					"source.address": structpb.NewStringValue("10.0.0.1"),
+				},
+			},
+		}
+		v, ok := req.Attribute("source.address")
+		require.True(t, ok)
+		require.Equal(t, "10.0.0.1", v.GetStringValue())
+	})
+
+	t.Run("missing namespace or key", func(t *testing.T) {
+		req := filter.NewRequestContext()
+		_, ok := req.Attribute("source.address")
+		require.False(t, ok)
+
+		req.Attributes = map[string]*structpb.Struct{
+			"envoy.filters.http.ext_proc": {Fields: map[string]*structpb.Value{"source.address": structpb.NewStringValue("10.0.0.1")}},
+		}
+		_, ok = req.Attribute("source.port")
+		require.False(t, ok)
+	})
+}
+
 func TestProcessRequestHeaders(t *testing.T) {
 	for _, tt := range []struct {
 		name    string

service/attributes_test.go

@@ -0,0 +1,46 @@
+package service
+
+import (
+	"testing"
+
+	"github.com/getyourguide/extproc-go/filter"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+func TestMergeAttributesIntoReq(t *testing.T) {
+	t.Run("populates empty req.Attributes", func(t *testing.T) {
+		req := filter.NewRequestContext()
+		mergeAttributesIntoReq(req, map[string]*structpb.Struct{
+			"envoy.filters.http.ext_proc": {Fields: map[string]*structpb.Value{"source.address": structpb.NewStringValue("10.0.0.1")}},
+		})
+		v, ok := req.Attribute("source.address")
+		require.True(t, ok)
+		require.Equal(t, "10.0.0.1", v.GetStringValue())
+	})
+
+	t.Run("merges fields across calls instead of overwriting the namespace", func(t *testing.T) {
+		req := filter.NewRequestContext()
+		mergeAttributesIntoReq(req, map[string]*structpb.Struct{
+			"envoy.filters.http.ext_proc": {Fields: map[string]*structpb.Value{"request.path": structpb.NewStringValue("/foo")}},
+		})
+		mergeAttributesIntoReq(req, map[string]*structpb.Struct{
+			"envoy.filters.http.ext_proc": {Fields: map[string]*structpb.Value{"request.method": structpb.NewStringValue("GET")}},
+		})
+
+		path, ok := req.Attribute("request.path")
+		require.True(t, ok)
+		require.Equal(t, "/foo", path.GetStringValue())
+
+		method, ok := req.Attribute("request.method")
+		require.True(t, ok)
+		require.Equal(t, "GET", method.GetStringValue())
+	})
+
+	t.Run("empty merge is a no-op", func(t *testing.T) {
+		req := filter.NewRequestContext()
+		mergeAttributesIntoReq(req, nil)
+		_, ok := req.Attribute("request.path")
+		require.False(t, ok)
+	})
+}

service/service.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
+	"maps"
 
 	extproc "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/getyourguide/extproc-go/filter"
@@ -17,6 +18,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	"go.opentelemetry.io/otel/trace/noop"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 const (
@@ -85,7 +87,7 @@ func (svc *ExtProcessor) Process(procsrv extproc.ExternalProcessor_ProcessServer
 		switch msg := procreq.Request.(type) {
 		case *extproc.ProcessingRequest_RequestHeaders:
 			ctx, span := svc.tracer.Start(ctx, RequestHeadersResourceName)
-			if err := svc.requestHeadersMessage(ctx, req, msg, procsrv); err != nil {
+			if err := svc.requestHeadersMessage(ctx, req, msg, procreq.GetAttributes(), procsrv); err != nil {
 				span.End()
 				return IgnoreCanceled(err)
 			}
@@ -106,7 +108,7 @@ func (svc *ExtProcessor) Process(procsrv extproc.ExternalProcessor_ProcessServer
 			span.End()
 		case *extproc.ProcessingRequest_ResponseHeaders:
 			ctx, span := svc.tracer.Start(ctx, ResponseHeadersResourceName)
-			if err := svc.responseHeadersMessage(ctx, req, msg, procsrv); err != nil {
+			if err := svc.responseHeadersMessage(ctx, req, msg, procreq.GetAttributes(), procsrv); err != nil {
 				span.End()
 				return IgnoreCanceled(err)
 			}
@@ -131,12 +133,36 @@ func (svc *ExtProcessor) Process(procsrv extproc.ExternalProcessor_ProcessServer
 	}
 }
 
+// mergeAttributesIntoReq merges Envoy-provided attributes into req.Attributes.
+// Fields are merged per namespace rather than the namespace being overwritten,
+// since request and response stage attributes may share a namespace.
+func mergeAttributesIntoReq(req *filter.RequestContext, attrs map[string]*structpb.Struct) {
+	if len(attrs) == 0 {
+		return
+	}
+	if req.Attributes == nil {
+		req.Attributes = make(map[string]*structpb.Struct, len(attrs))
+	}
+	for namespace, s := range attrs {
+		existing, ok := req.Attributes[namespace]
+		if !ok || existing == nil {
+			req.Attributes[namespace] = s
+			continue
+		}
+		if existing.Fields == nil {
+			existing.Fields = make(map[string]*structpb.Value, len(s.GetFields()))
+		}
+		maps.Copy(existing.Fields, s.GetFields())
+	}
+}
+
 // Step 1. Request headers: Contains the headers from the original HTTP request.
-func (svc *ExtProcessor) requestHeadersMessage(ctx context.Context, req *filter.RequestContext, msg *extproc.ProcessingRequest_RequestHeaders, procsrv extproc.ExternalProcessor_ProcessServer) error {
+func (svc *ExtProcessor) requestHeadersMessage(ctx context.Context, req *filter.RequestContext, msg *extproc.ProcessingRequest_RequestHeaders, attrs map[string]*structpb.Struct, procsrv extproc.ExternalProcessor_ProcessServer) error {
 	for _, header := range msg.RequestHeaders.GetHeaders().GetHeaders() {
 		headerValue := cmp.Or(string(header.GetRawValue()), header.GetValue())
 		req.RequestHeaders.Add(header.Key, headerValue)
 	}
+	mergeAttributesIntoReq(req, attrs)
 	crw := filter.NewCommonResponseWriter(req.RequestHeaders)
 
 	for _, f := range svc.filters {
@@ -207,11 +233,12 @@ func (svc *ExtProcessor) requestTrailersMessage(_ context.Context, _ *filter.Req
 }
 
 // Step 4. Response headers: Contains the headers from the HTTP response. Keep in mind that if the upstream system sends them before processing the request body that this message may arrive before the complete body.
-func (svc *ExtProcessor) responseHeadersMessage(ctx context.Context, req *filter.RequestContext, msg *extproc.ProcessingRequest_ResponseHeaders, procsrv extproc.ExternalProcessor_ProcessServer) error {
+func (svc *ExtProcessor) responseHeadersMessage(ctx context.Context, req *filter.RequestContext, msg *extproc.ProcessingRequest_ResponseHeaders, attrs map[string]*structpb.Struct, procsrv extproc.ExternalProcessor_ProcessServer) error {
 	for _, header := range msg.ResponseHeaders.GetHeaders().GetHeaders() {
 		headerValue := cmp.Or(string(header.GetRawValue()), header.GetValue())
 		req.ResponseHeaders.Add(header.Key, headerValue)
 	}
+	mergeAttributesIntoReq(req, attrs)
 	crw := filter.NewCommonResponseWriter(req.ResponseHeaders)
 
 	for i := len(svc.filters) - 1; i >= 0; i-- {