ext_proc: Support trailer callbacks (envoyproxy#16102)

Existing trailers will be sent to the processing
server if the processing mode is set to enable them. If the
processing mode is set to sent trailers, but there are no trailers
present, then empty trailers will be sent to the server for
modification.

However, trailers may only be added in the end of the data callback
in Envoy, which may come in before a previous gRPC reply returns.
Filters that need to be able to consistently add trailers where none
existed should enable trailer processing in the Envoy filter
configuration instead of relying on being able to turn it on
dynamically.

Risk Level: Low. Trailers only enabled if a service called by the filter
is configured to ask for them.

Testing: New integration and unit tests added.

Docs Changes: API docs updated in .proto files.

Release Notes:

When the processing mode is changed to SEND
for request or response trailers, a corresponding message will be
sent to the server, which can respond with trailer mutations as desired.

In addition, if trailer processing is enabled in the filter
configuration, then trailer messages will be sent to the server
even if trailers are not present. This makes it possible for the server
to add trailers where none exist.

Finally, at the moment Envoy only implements trailers for the HTTP/2
protocol. Nothing will happen if trailer processing is enabled and
Envoy is using HTTP/1 until Envoy implements trailers for HTTP/1.

Signed-off-by: Gregory Brail <gregbrail@google.com>
Signed-off-by: Gokul Nair <gnair@twitter.com>

api/envoy/extensions/filters/http/ext_proc/v3alpha/ext_proc.proto

@@ -26,18 +26,19 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // The filter will send the "request_headers" and "response_headers" messages by default.
 // In addition, if the "processing mode" is set , the "request_body" and "response_body"
 // messages will be sent if the corresponding fields of the "processing_mode" are
-// set to BUFFERED. The other body processing modes, and trailer processing, are not
+// set to BUFFERED, and trailers will be sent if the corresponding fields are set
+// to SEND. The other body processing modes are not
 // implemented yet. The filter will also respond to "immediate_response" messages
 // at any point in the stream.
 
 // As designed, the filter supports up to six different processing steps, which are in the
 // process of being implemented:
 // * Request headers: IMPLEMENTED
 // * Request body: Only BUFFERED mode is implemented
-// * Request trailers: NOT IMPLEMENTED
+// * Request trailers: IMPLEMENTED
 // * Response headers: IMPLEMENTED
 // * Response body: Only BUFFERED mode is implemented
-// * Response trailers: NOT IMPLEMENTED
+// * Response trailers: IMPLEMENTED
 
 // The filter communicates with an external gRPC service that can use it to do a variety of things
 // with the request and response:
@@ -57,6 +58,8 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // * Whether it receives the message body at all, in separate chunks, or as a single buffer
 // * Whether subsequent HTTP requests are transmitted synchronously or whether they are
 //   sent asynchronously.
+// * To modify request or response trailers if they already exist
+// * To add request or response trailers where they are not present
 //
 // All of this together allows a server to process the filter traffic in fairly
 // sophisticated ways. For example:

api/envoy/service/ext_proc/v3alpha/external_processor.proto

@@ -59,12 +59,16 @@ service ExternalProcessor {
 // [#next-free-field: 8]
 message ProcessingRequest {
   // Specify whether the filter that sent this request is running in synchronous
-  // or asynchronous mode. If false, then the server must either respond
-  // with exactly one ProcessingResponse message or close the stream.
-  // If true, however, then the server must not respond with
-  // an additional message, although it may still close the stream.
-  // The choice of synchronous or asynchronous mode can be chosen in the
-  // filter configuration.
+  // or asynchronous mode. The choice of synchronous or asynchronous mode
+  // can be set in the filter configuration, and defaults to false.
+  //
+  // * A value of "false" indicates that the server must respond
+  //   to this message by either sending back a matching ProcessingResponse message,
+  //   or by closing the stream.
+  // * A value of "true" indicates that the server must not respond to this
+  //   message, although it may still close the stream to indicate that no more messages
+  //   are needed.
+  //
   bool async_mode = 1;
 
   // Each request message will include one of the following sub-messages. Which
@@ -74,29 +78,41 @@ message ProcessingRequest {
     option (validate.required) = true;
 
     // Information about the HTTP request headers, as well as peer info and additional
-    // properties. If "response_required" is set, the server must send back a
+    // properties. Unless "async_mode" is true, the server must send back a
     // HeaderResponse message, an ImmediateResponse message, or close the stream.
     HttpHeaders request_headers = 2;
 
     // Information about the HTTP response headers, as well as peer info and additional
-    // properties. If "response_required" is set, the server must send back a
+    // properties. Unless "async_mode" is true, the server must send back a
     // HeaderResponse message or close the stream.
     HttpHeaders response_headers = 3;
 
-    // A chunk of the HTTP request body. If "response_required" is set, the server must send back
+    // A chunk of the HTTP request body. Unless "async_mode" is true, the server must send back
     // a BodyResponse message, an ImmediateResponse message, or close the stream.
     HttpBody request_body = 4;
 
-    // A chunk of the HTTP request body. If "response_required" is set, the server must send back
+    // A chunk of the HTTP request body. Unless "async_mode" is true, the server must send back
     // a BodyResponse message or close the stream.
     HttpBody response_body = 5;
 
-    // The HTTP trailers for the request path. If "response_required" is set, the server
+    // The HTTP trailers for the request path. Unless "async_mode" is true, the server
     // must send back a TrailerResponse message or close the stream.
+    //
+    // This message is only sent if the trailers processing mode is set to "SEND".
+    // If there are no trailers on the original downstream request, then this message
+    // will only be sent (with empty trailers waiting to be populated) if the
+    // processing mode is set before the request headers are sent, such as
+    // in the filter configuration.
     HttpTrailers request_trailers = 6;
 
-    // The HTTP trailers for the response path. If "response_required" is set, the server
+    // The HTTP trailers for the response path. Unless "async_mode" is true, the server
     // must send back a TrailerResponse message or close the stream.
+    //
+    // This message is only sent if the trailers processing mode is set to "SEND".
+    // If there are no trailers on the original downstream request, then this message
+    // will only be sent (with empty trailers waiting to be populated) if the
+    // processing mode is set before the request headers are sent, such as
+    // in the filter configuration.
     HttpTrailers response_trailers = 7;
   }
 }
@@ -142,6 +158,7 @@ message ProcessingResponse {
     ImmediateResponse immediate_response = 7;
   }
 
+  // [#not-implemented-hide:]
   // Optional metadata that will be emitted as dynamic metadata to be consumed by the next
   // filter. This metadata will be placed in the namespace "envoy.filters.http.ext_proc".
   google.protobuf.Struct dynamic_metadata = 8;
@@ -162,6 +179,7 @@ message HttpHeaders {
   // lower-cased, because HTTP header keys are case-insensitive.
   config.core.v3.HeaderMap headers = 1;
 
+  // [#not-implemented-hide:]
   // The values of properties selected by the "request_attributes"
   // or "response_attributes" list in the configuration. Each entry
   // in the list is populated
@@ -213,6 +231,7 @@ message CommonResponse {
     // stream as normal. This is the default.
     CONTINUE = 0;
 
+    // [#not-implemented-hide:]
     // Replace the request or response with the contents
     // of this message. If header_mutation is set, apply it to the
     // headers. If body_mutation is set and contains a body, then add that
@@ -229,24 +248,23 @@ message CommonResponse {
   ResponseStatus status = 1 [(validate.rules).enum = {defined_only: true}];
 
   // Instructions on how to manipulate the headers. When responding to an
-  // HttpBody request, header mutations will only take effect if the
-  // headers were not already sent further on the filter chain, which
-  // happens only if the current processing mode for the body is BUFFERED
-  // or BUFFERED_PARTIAL.
+  // HttpBody request, header mutations will only take effect if
+  // the current processing mode for the body is BUFFERED.
   HeaderMutation header_mutation = 2;
 
   // Replace the body of the last message sent to the remote server on this
   // stream. If responding to an HttpBody request, simply replace or clear
-  // the body chunk that was sent with that request. If responding to an
-  // HttpHeaders request, then a new body may be added to the request if this
-  // message is returned along with the CONTINUE_AND_REPLACE status.
+  // the body chunk that was sent with that request. Body mutations only take
+  // effect in response to "body" messages and are ignored otherwise.
   BodyMutation body_mutation = 3;
 
+  // [#not-implemented-hide:]
   // Add new trailers to the message. This may be used when responding to either a
   // HttpHeaders or HttpBody message, but only if this message is returned
   // along with the CONTINUE_AND_REPLACE status.
   config.core.v3.HeaderMap trailers = 4;
 
+  // [#not-implemented-hide:]
   // Clear the route cache for the current request.
   // This is necessary if the remote server
   // modified headers that are used to calculate the route.

source/common/http/filter_manager.cc

@@ -206,7 +206,10 @@ bool ActiveStreamFilterBase::commonHandleAfterTrailersCallback(FilterTrailersSta
     } else {
       ASSERT(headers_continued_);
     }
-  } else {
+  } else if (status == FilterTrailersStatus::StopIteration) {
+    if (canIterate()) {
+      iteration_state_ = IterationState::StopSingleIteration;
+    }
     return false;
   }