Alibaba Cloud Service Mesh:Use Envoy external processing service for custom processing of requests

Prerequisites

• Add a cluster to an ASM instance.

• Automatic sidecar proxy injection is enabled.

• HTTPBin application is deployed and can be accessed. A sleep application is deployed as outlined in Related operations.

Mechanism

The following diagram illustrates how an Envoy external processing service manages requests from downstream services and responses from upstream services:

1. The downstream service ①sends a request to the upstream service. Envoy blocks the request and ②sends it to the external processing service for processing.

2. The external processing service processes the request and ③returns the result to Envoy.

3. Envoy processes the request based on the result and ④forwards the request to the upstream service.

4. The upstream service processes the request and ⑤returns a response to the downstream service. Envoy blocks the response and ⑥forwards it to the external processing server.

5. The external processing server ⑦processes the response and ⑧returns it to Envoy.

6. Envoy processes the response based on the results and ⑨forwards the processed response to the downstream service.

Step 1: Write a processing logic for the external processing service

The following code block is a snippet of the core logic code. For the complete sample code, see ext-proc-demo. For more information, see Envoy documentation.

func NewServer() *Server {
	return &Server{}
}

// Server implements the Envoy external processing server interface
// See https://www.envoyproxy.io/docs/envoy/latest/api-v3/service/ext_proc/v3/external_processor.proto
type Server struct{}

func (s *Server) Process(srv extProcPb.ExternalProcessor_ProcessServer) error {
	klog.Infof("Processing")
	ctx := srv.Context()
	for {
		select {
		case <-ctx.Done():
			klog.Infof("context done")
			return ctx.Err()
		default:
		}

		req, err := srv.Recv()
		if err == io.EOF {
			// envoy has closed the stream. Don't return anything and close this stream entirely
			return nil
		}
		if err != nil {
			return status.Errorf(codes.Unknown, "cannot receive stream request: %v", err)
		}

		resp := &extProcPb.ProcessingResponse{}
		switch v := req.Request.(type) {
        // Requests that require processing of request headers
		case *extProcPb.ProcessingRequest_RequestHeaders:
			klog.Infof("Got RequestHeaders")
			h := req.Request.(*extProcPb.ProcessingRequest_RequestHeaders)
			resp = handleRequestHeaders(h)
        // Requests that require processing of response headers
        case *extProcPb.ProcessingRequest_ResponseHeaders:
			klog.Infof("Got ResponseHeaders")
			h := req.Request.(*extProcPb.ProcessingRequest_ResponseHeaders)
			resp = handleResponseHeaders(h)
        // Requests that require processing of request headers, not currently implemented
		case *extProcPb.ProcessingRequest_RequestBody:
			klog.Infof("Got RequestBody (not currently handled)")
        // Requests that require processing of request trailers, not currently implemented
		case *extProcPb.ProcessingRequest_RequestTrailers:
			klog.Infof("Got RequestTrailers (not currently handled)")
        // Requests that require processing of response body, not currently implemented
		case *extProcPb.ProcessingRequest_ResponseBody:
			klog.Infof("Got ResponseBody (not currently handled)")
        // Requests that require processing of response trailers, not currently implemented
		case *extProcPb.ProcessingRequest_ResponseTrailers:
			klog.Infof("Got ResponseTrailers (not currently handled)")

		default:
			klog.Infof("Unknown Request type %v", v)
		}
        // Return the processing required for the request
		klog.Infof("Sending ProcessingResponse: %+v", resp)
		if err := srv.Send(resp); err != nil {
			klog.Infof("send error %v", err)
			return err
		}
	}
}

// Add x-ext-proc-header=hello-to-asm to request headers
func handleRequestHeaders(req *extProcPb.ProcessingRequest_RequestHeaders) *extProcPb.ProcessingResponse {
	klog.Infof("handle request headers: %+v\n", req)

	resp := &extProcPb.ProcessingResponse{
		Response: &extProcPb.ProcessingResponse_RequestHeaders{
			RequestHeaders: &extProcPb.HeadersResponse{
				Response: &extProcPb.CommonResponse{
					HeaderMutation: &extProcPb.HeaderMutation{
						SetHeaders: []*configPb.HeaderValueOption{
							{
								Header: &configPb.HeaderValue{
									Key:      "x-ext-proc-header",
									RawValue: []byte("hello-to-asm"),
								},
							},
						},
					},
				},
			},
		},
	}
    
	return resp
}

// Add x-ext-proc-header=hello-from-asm to response headers
func handleResponseHeaders(req *extProcPb.ProcessingRequest_ResponseHeaders) *extProcPb.ProcessingResponse {
	klog.Infof("handle response headers: %+v\n", req)

	resp := &extProcPb.ProcessingResponse{
		Response: &extProcPb.ProcessingResponse_ResponseHeaders{
			ResponseHeaders: &extProcPb.HeadersResponse{
				Response: &extProcPb.CommonResponse{
					HeaderMutation: &extProcPb.HeaderMutation{
						SetHeaders: []*configPb.HeaderValueOption{
							{
								Header: &configPb.HeaderValue{
									Key:      "x-ext-proc-header",
									RawValue: []byte("hello-from-asm"),
								},
							},
						},
					},
				},
			},
		},
	}

	return resp
}

Step 2: Deploy the external processing service

In this example, the sample image of the external processing service provided by ASM is used. The service will add the request header x-ext-proc-header: hello-to-asm to the received requests and append the response header x-ext-proc-header: hello-from-asm to the received responses.

1. Create a file named ext.yaml with the following content.

   apiVersion: v1
   kind: Service
   metadata:
     name: ext-proc
     labels:
       app: ext-proc
       service: ext-proc
   spec:
     ports:
     # The listener ports of the external processing service.
     - name: grpc
       port: 9002
       targetPort: 9002
     selector:
       app: ext-proc
   ---
   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: ext-proc
   spec:
     replicas: 1
     selector:
       matchLabels:
         app: ext-proc
         version: v1
     template:
       metadata:
         labels:
           app: ext-proc
           version: v1
       spec:
         containers:
         - image: registry-cn-hangzhou.ack.aliyuncs.com/ack-demo/ext-proc:v0.2
           imagePullPolicy: IfNotPresent
           name: ext-proc
           ports:
           - containerPort: 9002

2. Run the following command to view the log of the pod and verify whether the external processing service is operational.

   kubectl logs ext-proc-64c8xxxxx-xxxxx

   Expected output:

   I1126 06:41:25.467033       1 main.go:52] Starting gRPC server on port :9002

   The log as shown above indicates that the external processing service is running as expected.

Step 3: Configure the Envoy Filter

1. Log on to the ASM console. In the left-side navigation pane, choose Service Mesh > Mesh Management.

2. On the Mesh Management page, click the name of the ASM instance. In the left-side navigation pane, choose Plugin Extension Center > EnvoyFilter Template.

3. Create an Envoy Filter named httpbin-ext-proc with the following content.

   apiVersion: networking.istio.io/v1alpha3
   kind: EnvoyFilter
   spec:
     configPatches:
       - applyTo: HTTP_FILTER
         match:
           context: SIDECAR_INBOUND
           listener:
             portNumber: 80
             filterChain:
               filter:
                 name: envoy.filters.network.http_connection_manager
           proxy:
             proxyVersion: ^MIN_VERSION-MAX_VERSION.*
         patch:
           operation: INSERT_BEFORE
           value:
             name: envoy.filters.http.ext_proc
             typed_config:
               '@type': >-
                 type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor
               grpc_service:
                 envoy_grpc:
                   cluster_name: outbound|9002||ext-proc.default.svc.cluster.local
                   authority: ext-proc.default.svc.cluster.local
               processing_mode:
                 request_header_mode: SEND
                 response_header_mode: SEND

Step 4: Access the HTTPBin application for verification

Run the following command to access the HTTPBin application and check the response header.

kubectl exec -it deploy/sleep -- curl httpbin:8000/headers -i

Expected output:

HTTP/1.1 200 OK
server: envoy
date: Wed, 11 Dec 2024 06:47:59 GMT
content-type: application/json
content-length: 564
access-control-allow-origin: *
access-control-allow-credentials: true
x-envoy-upstream-service-time: 3
x-ext-proc-header: hello-from-asm

{
  "headers": {
    "Accept": "*/*",
    "Host": "httpbin:8000",
    "User-Agent": "curl/8.1.2",
    "X-B3-Parentspanid": "5c6dd2cc9312d6bb",
    "X-B3-Sampled": "1",
    "X-B3-Spanid": "1153a2737cee4434",
    "X-B3-Traceid": "baba86b696edc75a5c6dd2cc9312d6bb",
    "X-Envoy-Attempt-Count": "1",
    "X-Ext-Proc-Header": "hello-to-asm",
    "X-Forwarded-Client-Cert": "By=spiffe://cluster.local/ns/default/sa/httpbin;Hash=69d8f267c3c00b4396a83e12d14520acc9dadb1492d660e10f77e94dcad7cb06;Subject=\"\";URI=spiffe://cluster.local/ns/default/sa/sleep"
  }
}

The output shows that the request header x-ext-proc-header: hello-to-asm is added before the request was forwarded, and the response header x-ext-proc-header: hello-from-asm is added before the response was returned.