refactor proxy message processor that remove returns type

Author: e-n-0

contrib/envoyproxy/go-control-plane/envoy.go

@@ -48,15 +48,15 @@ type appsecEnvoyExternalProcessorServer struct {
 	envoyextproc.ExternalProcessorServer
 	config           AppsecEnvoyConfig
 	requestCounter   atomic.Uint32
-	messageProcessor proxy.Processor[envoyextproc.ProcessingResponse]
+	messageProcessor proxy.Processor
 }
 
 // AppsecEnvoyExternalProcessorServer creates a new external processor server with AAP enabled
 func AppsecEnvoyExternalProcessorServer(userImplementation envoyextproc.ExternalProcessorServer, config AppsecEnvoyConfig) envoyextproc.ExternalProcessorServer {
 	processor := &appsecEnvoyExternalProcessorServer{
 		ExternalProcessorServer: userImplementation,
 		config:                  config,
-		messageProcessor: proxy.NewProcessor[envoyextproc.ProcessingResponse](proxy.ProcessorConfig[envoyextproc.ProcessingResponse]{
+		messageProcessor: proxy.NewProcessor(proxy.ProcessorConfig{
 			BlockingUnavailable:  config.BlockingUnavailable,
 			BodyParsingSizeLimit: config.BodyParsingSizeLimit,
 			Framework:            "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3",
@@ -76,10 +76,14 @@ func AppsecEnvoyExternalProcessorServer(userImplementation envoyextproc.External
 	return processor
 }
 
+type processServerKeyType struct{}
+
+var processServerKey processServerKeyType
+
 // Process handles the bidirectional stream that Envoy uses to control the filter
 func (s *appsecEnvoyExternalProcessorServer) Process(processServer envoyextproc.ExternalProcessor_ProcessServer) error {
 	var (
-		ctx            = processServer.Context()
+		ctx            = context.WithValue(processServer.Context(), processServerKey, processServer)
 		currentRequest proxy.RequestState
 	)
 
@@ -92,7 +96,7 @@ func (s *appsecEnvoyExternalProcessorServer) Process(processServer envoyextproc.
 	}()
 
 	for {
-		if err := s.checkContext(ctx); err != nil {
+		if err := s.checkContext(processServer.Context()); err != nil {
 			return err
 		}
 
@@ -102,18 +106,12 @@ func (s *appsecEnvoyExternalProcessorServer) Process(processServer envoyextproc.
 		}
 
 		// Process the message
-		processingResponse, err := s.processMessage(ctx, &processingRequest, &currentRequest)
+		err := s.processMessage(ctx, &processingRequest, &currentRequest)
 		if err != nil && err != io.EOF {
 			instr.Logger().Error("external_processing: error processing request: %s\n", err.Error())
 			return err
 		}
 
-		if processingResponse != nil {
-			if err := s.sendResponse(processServer, processingResponse); err != nil {
-				return err
-			}
-		}
-
 		if err == io.EOF {
 			return nil
 		}
@@ -144,11 +142,11 @@ func (s *appsecEnvoyExternalProcessorServer) handleReceiveError(err error) error
 }
 
 // processMessage processes a single message based on its type
-func (s *appsecEnvoyExternalProcessorServer) processMessage(ctx context.Context, req *envoyextproc.ProcessingRequest, currentRequest *proxy.RequestState) (action *envoyextproc.ProcessingResponse, err error) {
+func (s *appsecEnvoyExternalProcessorServer) processMessage(ctx context.Context, req *envoyextproc.ProcessingRequest, currentRequest *proxy.RequestState) (err error) {
 	switch v := req.Request.(type) {
 	case *envoyextproc.ProcessingRequest_RequestHeaders:
-		*currentRequest, action, err = s.messageProcessor.OnRequestHeaders(ctx, &messageRequestHeaders{ProcessingRequest: req, HttpHeaders: req.GetRequestHeaders(), integration: s.config.Integration})
-		return action, err
+		*currentRequest, err = s.messageProcessor.OnRequestHeaders(ctx, &messageRequestHeaders{ProcessingRequest: req, HttpHeaders: req.GetRequestHeaders(), integration: s.config.Integration})
+		return err
 
 	case *envoyextproc.ProcessingRequest_RequestBody:
 		return s.messageProcessor.OnRequestBody(&messageBody{ProcessingRequest: req, HttpBody: req.GetRequestBody()}, currentRequest)
@@ -158,7 +156,7 @@ func (s *appsecEnvoyExternalProcessorServer) processMessage(ctx context.Context,
 			// Handle case where request headers were never sent
 			instr.Logger().Warn("external_processing: can't process the response: envoy never sent the beginning of the request, this is a known issue" +
 				" and can happen when a malformed request is sent to Envoy where the header Host is missing. See link to issue https://github.com/envoyproxy/envoy/issues/38022")
-			return nil, status.Errorf(codes.InvalidArgument, "Error processing response headers from ext_proc: can't process the response")
+			return status.Errorf(codes.InvalidArgument, "Error processing response headers from ext_proc: can't process the response")
 		}
 		return s.messageProcessor.OnResponseHeaders(&responseHeadersEnvoy{ProcessingRequest: req, HttpHeaders: req.GetResponseHeaders()}, currentRequest)
 
@@ -172,18 +170,6 @@ func (s *appsecEnvoyExternalProcessorServer) processMessage(ctx context.Context,
 		return s.messageProcessor.OnResponseTrailers(currentRequest)
 
 	default:
-		return nil, status.Errorf(codes.Unknown, "Unknown request type: %T", v)
+		return status.Errorf(codes.Unknown, "Unknown request type: %T", v)
 	}
 }
-
-// sendResponse sends a processing response back to Envoy
-func (s *appsecEnvoyExternalProcessorServer) sendResponse(processServer envoyextproc.ExternalProcessor_ProcessServer, response *envoyextproc.ProcessingResponse) error {
-	instr.Logger().Debug("external_processing: sending response: %v\n", response)
-
-	if err := processServer.SendMsg(response); err != nil {
-		instr.Logger().Error("external_processing: error sending response (probably because of an Envoy timeout): %s", err.Error())
-		return status.Errorf(codes.Unknown, "Error sending response (probably because of an Envoy timeout): %s", err.Error())
-	}
-
-	return nil
-}

contrib/envoyproxy/go-control-plane/envoy_http.go

@@ -6,6 +6,7 @@
 package gocontrolplane
 
 import (
+	"context"
 	"fmt"
 	"math"
 	"net/http"
@@ -22,11 +23,20 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-func continueActionFunc(options proxy.ContinueActionOptions) (envoyextproc.ProcessingResponse, error) {
+func continueActionFunc(ctx context.Context, options proxy.ContinueActionOptions) error {
 	if len(options.HeaderMutations) > 0 || options.Body {
-		return buildHeadersResponse(options), nil
+		return buildHeadersResponse(ctx, options)
 	}
 
+	emptyResp, err := buildEmptyContinueResponse(options)
+	if err != nil {
+		return err
+	}
+
+	return sendResponse(ctx, &emptyResp)
+}
+
+func buildEmptyContinueResponse(options proxy.ContinueActionOptions) (envoyextproc.ProcessingResponse, error) {
 	common := &envoyextproc.CommonResponse{
 		Status: envoyextproc.CommonResponse_CONTINUE,
 	}
@@ -49,15 +59,15 @@ func continueActionFunc(options proxy.ContinueActionOptions) (envoyextproc.Proce
 	}
 }
 
-func blockActionFunc(data proxy.BlockActionOptions) (envoyextproc.ProcessingResponse, error) {
+func blockActionFunc(ctx context.Context, data proxy.BlockActionOptions) error {
 	blockedHeaders := convertHeadersToEnvoy(data.Headers)
 
 	var statusCode int32
 	if data.StatusCode > 0 && data.StatusCode <= math.MaxInt32 {
 		statusCode = int32(data.StatusCode)
 	}
 
-	return envoyextproc.ProcessingResponse{
+	resp := envoyextproc.ProcessingResponse{
 		Response: &envoyextproc.ProcessingResponse_ImmediateResponse{
 			ImmediateResponse: &envoyextproc.ImmediateResponse{
 				Status: &envoytypes.HttpStatus{
@@ -72,11 +82,13 @@ func blockActionFunc(data proxy.BlockActionOptions) (envoyextproc.ProcessingResp
 				},
 			},
 		},
-	}, nil
+	}
+
+	return sendResponse(ctx, &resp)
 }
 
 // buildHeadersResponse creates an Envoy HeadersResponse from provided data answering a RequestHeaders or ResponseHeaders message
-func buildHeadersResponse(data proxy.ContinueActionOptions) envoyextproc.ProcessingResponse {
+func buildHeadersResponse(ctx context.Context, data proxy.ContinueActionOptions) error {
 	var modeOverride *envoyextprocfilter.ProcessingMode
 	if data.Body {
 		modeOverride = &envoyextprocfilter.ProcessingMode{RequestBodyMode: envoyextprocfilter.ProcessingMode_STREAMED}
@@ -102,7 +114,24 @@ func buildHeadersResponse(data proxy.ContinueActionOptions) envoyextproc.Process
 		}
 	}
 
-	return processingResponse
+	return sendResponse(ctx, &processingResponse)
+}
+
+// sendResponse sends a processing response back to Envoy
+func sendResponse(ctx context.Context, response *envoyextproc.ProcessingResponse) error {
+	instr.Logger().Debug("external_processing: sending response: %v\n", response)
+
+	processServer, _ := ctx.Value(processServerKey).(envoyextproc.ExternalProcessor_ProcessServer)
+	if processServer == nil {
+		return status.Errorf(codes.Unknown, "No gRPC stream available to send the response")
+	}
+
+	if err := processServer.SendMsg(response); err != nil {
+		instr.Logger().Error("external_processing: error sending response (probably because of an Envoy timeout): %s", err.Error())
+		return status.Errorf(codes.Unknown, "Error sending response (probably because of an Envoy timeout): %s", err.Error())
+	}
+
+	return nil
 }
 
 // convertHeadersToEnvoy converts standard HTTP headers to Envoy HeaderValueOption format

contrib/envoyproxy/go-control-plane/proxy/config.go

@@ -31,15 +31,15 @@ type BlockActionOptions struct {
 }
 
 // ProcessorConfig contains configuration for the message processor
-type ProcessorConfig[O any] struct {
+type ProcessorConfig struct {
 	context.Context
 	BlockingUnavailable  bool
 	BodyParsingSizeLimit int
 	Framework            string
 
 	// ContinueMessageFunc is a function that generates a continue message of type O based on the provided ContinueActionOptions.
-	ContinueMessageFunc func(ContinueActionOptions) (O, error)
+	ContinueMessageFunc func(context.Context, ContinueActionOptions) error
 
 	// BlockMessageFunc is a function that generates a block message of type O based on the provided status code, headers, and body.
-	BlockMessageFunc func(options BlockActionOptions) (O, error)
+	BlockMessageFunc func(context.Context, BlockActionOptions) error
 }