Enhance ext_proc filter to support MXN streaming

api/envoy/extensions/filters/http/ext_proc/v3/processing_mode.proto

@@ -36,11 +36,12 @@ message ProcessingMode {
 
   // Control how the request and response bodies are handled
   // When body mutation by external processor is enabled, ext_proc filter will always remove
-  // the content length header in three cases below because content length can not be guaranteed
+  // the content length header in four cases below because content length can not be guaranteed
   // to be set correctly:
   // 1) STREAMED BodySendMode: header processing completes before body mutation comes back.
   // 2) BUFFERED_PARTIAL BodySendMode: body is buffered and could be injected in different phases.
   // 3) BUFFERED BodySendMode + SKIP HeaderSendMode: header processing (e.g., update content-length) is skipped.
+  // 4) FULL_DUPLEX_STREAMED BodySendMode: header processing completes before body mutation comes back.
   //
   // In Envoy's http1 codec implementation, removing content length will enable chunked transfer
   // encoding whenever feasible. The recipient (either client or server) must be able
@@ -68,6 +69,37 @@ message ProcessingMode {
     // chunk. If the body exceeds the configured buffer limit, then the body contents
     // up to the buffer limit will be sent.
     BUFFERED_PARTIAL = 3;
+
+    // [#not-implemented-hide:]
+    // Envoy streams the body to the server in pieces as they arrive.
+    //
+    // 1) The server may choose to buffer any number chunks of data before processing them.
+    // After it finishes buffering, the server processes the buffered data. Then it splits the processed
+    // data into any number of chunks, and streams them back to Envoy one by one.
+    // The server may continuously do so until the complete body is processed.
+    // The individual response chunk size is recommended to be no greater than 64K bytes, or
+    // :ref:`max_receive_message_length <envoy_v3_api_field_config.core.v3.GrpcService.EnvoyGrpc.max_receive_message_length>`
+    // if EnvoyGrpc is used.
+    //
+    // 2) The server may also choose to buffer the entire message, including the headers (if header mode is
+    // ``SEND``), the entire body, and the trailers (if present), before sending back any response.
+    // The server response has to maintain the headers-body-trailers ordering.
+    //
+    // 3) Note that the server might also choose not to buffer data. That is, upon receiving a
+    // body request, it could process the data and send back a body response immediately.
+    //
+    // In this body mode:
+    // * The corresponding trailer mode has to be set to ``SEND``.
+    // * Envoy will send body and trailers (if present) to the server as they arrive.
+    //   Sending the trailers (if present) is to inform the server the complete body arrives.
+    //   In case there are no trailers, then Envoy will set
+    //   :ref:`end_of_stream <envoy_v3_api_field_service.ext_proc.v3.HttpBody.end_of_stream>`
+    //   to true as part of the last body chunk request to notify the server that no other data is to be sent.
+    // * The server needs to send
+    //   :ref:`StreamedBodyResponse <envoy_v3_api_msg_service.ext_proc.v3.StreamedBodyResponse>`
+    //   to Envoy in the body response.
+    // * Envoy will stream the body chunks in the responses from the server to the upstream/downstream as they arrive.
+    FULL_DUPLEX_STREAMED = 4;
   }
 
   // How to handle the request header. Default is "SEND".

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

@@ -362,14 +362,37 @@ message HeaderMutation {
   repeated string remove_headers = 2;
 }
 
-// Replace the entire message body chunk received in the corresponding
-// HttpBody message with this new body, or clear the body.
+// [#not-implemented-hide:]
+// The body response message corresponding to FULL_DUPLEX_STREAMED body mode.
+message StreamedBodyResponse {
+  // The body response chunk that will be passed to the upstream/downstream by Envoy.
+  bytes body = 1;
+
+  // The server sets this flag to true if it has received a body request with
+  // :ref:`end_of_stream <envoy_v3_api_field_service.ext_proc.v3.HttpBody.end_of_stream>` set to true,
+  // and this is the last chunk of body responses.
+  bool end_of_stream = 2;
+}
+
+// This message specifies the body mutation the server sends to Envoy.
 message BodyMutation {
   oneof mutation {
-    // The entire body to replace
+    // The entire body to replace.
+    // Should only be used when the corresponding ``BodySendMode`` in the
+    // :ref:`processing_mode <envoy_v3_api_field_extensions.filters.http.ext_proc.v3.ExternalProcessor.processing_mode>`
+    // is not set to ``FULL_DUPLEX_STREAMED``.
     bytes body = 1;
 
-    // Clear the corresponding body chunk
+    // Clear the corresponding body chunk.
+    // Should only be used when the corresponding ``BodySendMode`` in the
+    // :ref:`processing_mode <envoy_v3_api_field_extensions.filters.http.ext_proc.v3.ExternalProcessor.processing_mode>`
+    // is not set to ``FULL_DUPLEX_STREAMED``.
     bool clear_body = 2;
+
+    // [#not-implemented-hide:]
+    // Must be used when the corresponding ``BodySendMode`` in the
+    // :ref:`processing_mode <envoy_v3_api_field_extensions.filters.http.ext_proc.v3.ExternalProcessor.processing_mode>`
+    // is set to ``FULL_DUPLEX_STREAMED``.
+    StreamedBodyResponse streamed_response = 3;
   }
 }

source/extensions/filters/http/ext_proc/ext_proc.cc

@@ -72,6 +72,35 @@ initGrpcService(const ExtProcPerRoute& config) {
   return absl::nullopt;
 }
 
+void verifyProcessingModeConfig(const ExternalProcessor& config) {
+  const ProcessingMode& processing_mode = config.processing_mode();
+  if (config.has_http_service()) {
+    // In case http_service configured, the processing mode can only support sending headers.
+    if (processing_mode.request_body_mode() != ProcessingMode::NONE ||
+        processing_mode.response_body_mode() != ProcessingMode::NONE ||
+        processing_mode.request_trailer_mode() == ProcessingMode::SEND ||
+        processing_mode.response_trailer_mode() == ProcessingMode::SEND) {
+      throw EnvoyException(
+          "If the ext_proc filter is configured with http_service instead of gRPC service, "
+          "then the processing modes of this filter can not be configured to send body or "
+          "trailer.");
+    }
+  }
+
+  if ((processing_mode.request_body_mode() == ProcessingMode::FULL_DUPLEX_STREAMED) &&
+      (processing_mode.request_trailer_mode() != ProcessingMode::SEND)) {
+    throw EnvoyException(
+        "If the ext_proc filter has the request_body_mode set to FULL_DUPLEX_STREAMED, "
+        "then the request_trailer_mode has to be set to SEND");
+  }
+  if ((processing_mode.response_body_mode() == ProcessingMode::FULL_DUPLEX_STREAMED) &&
+      (processing_mode.response_trailer_mode() != ProcessingMode::SEND)) {
+    throw EnvoyException(
+        "If the ext_proc filter has the response_body_mode set to FULL_DUPLEX_STREAMED, "
+        "then the response_trailer_mode has to be set to SEND");
+  }
+}
+
 std::vector<std::string> initNamespaces(const Protobuf::RepeatedPtrField<std::string>& ns) {
   std::vector<std::string> namespaces;
   for (const auto& single_ns : ns) {
@@ -231,16 +260,8 @@ FilterConfig::FilterConfig(const ExternalProcessor& config,
                           config.response_attributes()),
       immediate_mutation_checker_(context.regexEngine()),
       thread_local_stream_manager_slot_(context.threadLocal().allocateSlot()) {
-  if (!grpc_service_.has_value()) {
-    // In case http_service configured, the processing mode can only support sending headers.
-    if (processing_mode_.request_body_mode() != ProcessingMode::NONE ||
-        processing_mode_.response_body_mode() != ProcessingMode::NONE ||
-        processing_mode_.request_trailer_mode() == ProcessingMode::SEND ||
-        processing_mode_.response_trailer_mode() == ProcessingMode::SEND) {
-      throw EnvoyException(
-          "If http_service is configured, processing modes can not send any body or trailer.");
-    }
-  }
+  // Validate processing mode configuration.
+  verifyProcessingModeConfig(config);
   if (config.disable_clear_route_cache() && (route_cache_action_ != ExternalProcessor::DEFAULT)) {
     throw EnvoyException("disable_clear_route_cache and route_cache_action can not "
                          "be set to none-default at the same time.");
@@ -562,7 +583,6 @@ FilterDataStatus Filter::handleDataBufferedMode(ProcessorState& state, Buffer::I
     case StreamOpenState::IgnoreError:
       return FilterDataStatus::Continue;
     case StreamOpenState::Ok:
-      // Fall through
       break;
     }
 
@@ -581,37 +601,24 @@ FilterDataStatus Filter::handleDataBufferedMode(ProcessorState& state, Buffer::I
   return FilterDataStatus::StopIterationAndBuffer;
 }
 
-FilterDataStatus Filter::handleDataStreamedMode(ProcessorState& state, Buffer::Instance& data,
-                                                bool end_stream) {
-  // STREAMED body mode works as follows:
-  //
-  // 1) As data callbacks come in to the filter, it "moves" the data into a new buffer, which it
-  // dispatches via gRPC message to the external processor, and then keeps in a queue. It
-  // may request a watermark if the queue is higher than the buffer limit to prevent running
-  // out of memory.
-  // 2) As a result, filters farther down the chain see empty buffers in some data callbacks.
-  // 3) When a response comes back from the external processor, it injects the processor's result
-  // into the filter chain using "inject**codedData". (The processor may respond indicating that
-  // there is no change, which means that the original buffer stored in the queue is what gets
-  // injected.)
-  //
-  // This way, we pipeline data from the proxy to the external processor, and give the processor
-  // the ability to modify each chunk, in order. Doing this any other way would have required
-  // substantial changes to the filter manager. See
-  // https://github.com/envoyproxy/envoy/issues/16760 for a discussion.
+FilterDataStatus Filter::handleDataStreamedModeBase(ProcessorState& state, Buffer::Instance& data,
+                                                    bool end_stream) {
   switch (openStream()) {
   case StreamOpenState::Error:
     return FilterDataStatus::StopIterationNoBuffer;
   case StreamOpenState::IgnoreError:
     return FilterDataStatus::Continue;
   case StreamOpenState::Ok:
-    // Fall through
     break;
   }
 
-  // Need to first enqueue the data into the chunk queue before sending.
   ProcessingRequest req = setupBodyChunk(state, data, end_stream);
-  state.enqueueStreamingChunk(data, end_stream);
+  if (state.bodyMode() != ProcessingMode::FULL_DUPLEX_STREAMED) {
+    state.enqueueStreamingChunk(data, end_stream);
+  } else {
+    // For FULL_DUPLEX_STREAMED mode, just drain the data.
+    data.drain(data.length());
+  }
   // If the current state is HeadersCallback, stays in that state.
   if (state.callbackState() == ProcessorState::CallbackState::HeadersCallback) {
     sendBodyChunk(state, ProcessorState::CallbackState::HeadersCallback, req);
@@ -626,6 +633,35 @@ FilterDataStatus Filter::handleDataStreamedMode(ProcessorState& state, Buffer::I
   }
 }
 
+FilterDataStatus Filter::handleDataStreamedMode(ProcessorState& state, Buffer::Instance& data,
+                                                bool end_stream) {
+  // STREAMED body mode works as follows:
+  //
+  // 1) As data callbacks come in to the filter, it "moves" the data into a new buffer, which it
+  // dispatches via gRPC message to the external processor, and then keeps in a queue. It
+  // may request a watermark if the queue is higher than the buffer limit to prevent running
+  // out of memory.
+  // 2) As a result, filters farther down the chain see empty buffers in some data callbacks.
+  // 3) When a response comes back from the external processor, it injects the processor's result
+  // into the filter chain using "inject**codedData". (The processor may respond indicating that
+  // there is no change, which means that the original buffer stored in the queue is what gets
+  // injected.)
+  //
+  // This way, we pipeline data from the proxy to the external processor, and give the processor
+  // the ability to modify each chunk, in order. Doing this any other way would have required
+  // substantial changes to the filter manager. See
+  // https://github.com/envoyproxy/envoy/issues/16760 for a discussion.
+  return handleDataStreamedModeBase(state, data, end_stream);
+}
+
+FilterDataStatus Filter::handleDataFullDuplexStreamedMode(ProcessorState& state,
+                                                          Buffer::Instance& data, bool end_stream) {
+  // FULL_DUPLEX_STREAMED body mode works similar to STREAMED except it does not put the data
+  // into the internal queue. And there is no internal queue based flow control. A copy of the
+  // data is dispatched to the external processor and the original data is drained.
+  return handleDataStreamedModeBase(state, data, end_stream);
+}
+
 FilterDataStatus Filter::handleDataBufferedPartialMode(ProcessorState& state,
                                                        Buffer::Instance& data, bool end_stream) {
   // BUFFERED_PARTIAL mode works as follows:
@@ -672,6 +708,8 @@ FilterDataStatus Filter::handleDataBufferedPartialMode(ProcessorState& state,
 }
 
 FilterDataStatus Filter::onData(ProcessorState& state, Buffer::Instance& data, bool end_stream) {
+  state.setBodyReceived(true);
+
   if (config_->observabilityMode()) {
     return sendDataInObservabilityMode(data, state, end_stream);
   }
@@ -690,10 +728,13 @@ FilterDataStatus Filter::onData(ProcessorState& state, Buffer::Instance& data, b
   }
 
   if (state.callbackState() == ProcessorState::CallbackState::HeadersCallback) {
-    if (state.bodyMode() == ProcessingMode::STREAMED &&
-        config_->sendBodyWithoutWaitingForHeaderResponse()) {
-      ENVOY_LOG(trace, "Sending body data even header processing is still in progress as body mode "
-                       "is STREAMED and send_body_without_waiting_for_header_response is enabled");
+    if ((state.bodyMode() == ProcessingMode::STREAMED &&
+         config_->sendBodyWithoutWaitingForHeaderResponse()) ||
+        state.bodyMode() == ProcessingMode::FULL_DUPLEX_STREAMED) {
+      ENVOY_LOG(trace,
+                "Sending body data even though header processing is still in progress as body mode "
+                "is FULL_DUPLEX_STREAMED or STREAMED and "
+                "send_body_without_waiting_for_header_response is enabled");
     } else {
       ENVOY_LOG(trace, "Header processing still in progress -- holding body data");
       // We don't know what to do with the body until the response comes back.
@@ -714,6 +755,9 @@ FilterDataStatus Filter::onData(ProcessorState& state, Buffer::Instance& data, b
   case ProcessingMode::STREAMED:
     result = handleDataStreamedMode(state, data, end_stream);
     break;
+  case ProcessingMode::FULL_DUPLEX_STREAMED:
+    result = handleDataFullDuplexStreamedMode(state, data, end_stream);
+    break;
   case ProcessingMode::BUFFERED_PARTIAL:
     result = handleDataBufferedPartialMode(state, data, end_stream);
     break;
@@ -723,7 +767,6 @@ FilterDataStatus Filter::onData(ProcessorState& state, Buffer::Instance& data, b
     result = FilterDataStatus::Continue;
     break;
   }
-
   return result;
 }
 
@@ -848,9 +891,14 @@ FilterTrailersStatus Filter::onTrailers(ProcessorState& state, Http::HeaderMap&
   state.setTrailers(&trailers);
 
   if (state.callbackState() != ProcessorState::CallbackState::Idle) {
-    ENVOY_LOG(trace, "Previous callback still executing -- holding header iteration");
-    state.setPaused(true);
-    return FilterTrailersStatus::StopIteration;
+    if (state.bodyMode() == ProcessingMode::FULL_DUPLEX_STREAMED) {
+      ENVOY_LOG(trace, "Body mode is FULL_DUPLEX_STREAMED, sending trailers even though Envoy is "
+                       "waiting for header or body response");
+    } else {
+      ENVOY_LOG(trace, "Previous callback still executing -- holding header iteration");
+      state.setPaused(true);
+      return FilterTrailersStatus::StopIteration;
+    }
   }
 
   if (!body_delivered &&
@@ -982,20 +1030,30 @@ void Filter::sendBodyChunk(ProcessorState& state, ProcessorState::CallbackState
 
 void Filter::sendTrailers(ProcessorState& state, const Http::HeaderMap& trailers,
                           bool observability_mode) {
+  // Skip if the trailers is already sent to the server.
+  if (state.trailersSentToServer()) {
+    return;
+  }
+  state.setTrailersSentToServer(true);
+
   ProcessingRequest req;
   req.set_observability_mode(observability_mode);
   addAttributes(state, req);
   addDynamicMetadata(state, req);
   auto* trailers_req = state.mutableTrailers(req);
   MutationUtils::headersToProto(trailers, config_->allowedHeaders(), config_->disallowedHeaders(),
                                 *trailers_req->mutable_trailers());
-
   if (observability_mode) {
     ENVOY_LOG(debug, "Sending trailers message in observability mode");
   } else {
-    state.onStartProcessorCall(std::bind(&Filter::onMessageTimeout, this),
-                               config_->messageTimeout(),
-                               ProcessorState::CallbackState::TrailersCallback);
+    if (state.callbackState() == ProcessorState::CallbackState::Idle) {
+      state.onStartProcessorCall(std::bind(&Filter::onMessageTimeout, this),
+                                 config_->messageTimeout(),
+                                 ProcessorState::CallbackState::TrailersCallback);
+    } else {
+      state.onStartProcessorCall(std::bind(&Filter::onMessageTimeout, this),
+                                 config_->messageTimeout(), state.callbackState());
+    }
     ENVOY_LOG(debug, "Sending trailers message");
   }
 
@@ -1181,6 +1239,8 @@ void Filter::onReceiveMessage(std::unique_ptr<ProcessingResponse>&& r) {
   // and filter is waiting for header processing response.
   // Otherwise, the response mode_override proto field is ignored.
   if (config_->allowModeOverride() && !config_->sendBodyWithoutWaitingForHeaderResponse() &&
+      (config_->processingMode().request_body_mode() != ProcessingMode::FULL_DUPLEX_STREAMED) &&
+      (config_->processingMode().response_body_mode() != ProcessingMode::FULL_DUPLEX_STREAMED) &&
       inHeaderProcessState() && response->has_mode_override()) {
     bool mode_override_allowed = true;
     const auto& mode_overide = response->mode_override();

source/extensions/filters/http/ext_proc/ext_proc.h

@@ -472,8 +472,12 @@ class Filter : public Logger::Loggable<Logger::Id::ext_proc>,
 
   Http::FilterDataStatus handleDataBufferedMode(ProcessorState& state, Buffer::Instance& data,
                                                 bool end_stream);
+  Http::FilterDataStatus handleDataStreamedModeBase(ProcessorState& state, Buffer::Instance& data,
+                                                    bool end_stream);
   Http::FilterDataStatus handleDataStreamedMode(ProcessorState& state, Buffer::Instance& data,
                                                 bool end_stream);
+  Http::FilterDataStatus handleDataFullDuplexStreamedMode(ProcessorState& state,
+                                                          Buffer::Instance& data, bool end_stream);
   Http::FilterDataStatus handleDataBufferedPartialMode(ProcessorState& state,
                                                        Buffer::Instance& data, bool end_stream);
   Http::FilterDataStatus onData(ProcessorState& state, Buffer::Instance& data, bool end_stream);