ext_proc: Add per-route configuration (#17691)

Today "disabled" and the processing mode are the two things supported
because that is what is currently supported in ext_proc.

Signed-off-by: Gregory Brail <[email protected]>

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

@@ -150,7 +150,6 @@ message ExternalProcessor {
   string stat_prefix = 8;
 }
 
-// [#not-implemented-hide:]
 // Extra settings that may be added to per-route configuration for a
 // virtual host or cluster.
 message ExtProcPerRoute {
@@ -161,23 +160,27 @@ message ExtProcPerRoute {
     // If disabled is specified in multiple per-filter-configs, the most specific one will be used.
     bool disabled = 1 [(validate.rules).bool = {const: true}];
 
-    // Override aspects of the configuration for this route
+    // Override aspects of the configuration for this route. A set of
+    // overrides in a more specific configuration will override a "disabled"
+    // flag set in a less-specific one.
     ExtProcOverrides overrides = 2;
   }
 }
 
-// [#not-implemented-hide:]
 // Overrides that may be set on a per-route basis
 message ExtProcOverrides {
   // Set a different processing mode for this route than the default.
   ProcessingMode processing_mode = 1;
 
+  // [#not-implemented-hide:]
   // Set a different asynchronous processing option than the default.
   bool async_mode = 2;
 
+  // [#not-implemented-hide:]
   // Set different optional properties than the default.
   repeated string request_properties = 3;
 
+  // [#not-implemented-hide:]
   // Set different optional properties than the default.
   repeated string response_properties = 4;
 }

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

@@ -1,7 +1,5 @@
 #include "source/extensions/filters/http/ext_proc/config.h"
 
-#include <string>
-
 #include "source/extensions/filters/http/ext_proc/client_impl.h"
 #include "source/extensions/filters/http/ext_proc/ext_proc.h"
 
@@ -14,7 +12,7 @@ Http::FilterFactoryCb ExternalProcessingFilterConfig::createFilterFactoryFromPro
     const envoy::extensions::filters::http::ext_proc::v3alpha::ExternalProcessor& proto_config,
     const std::string& stats_prefix, Server::Configuration::FactoryContext& context) {
   const uint32_t message_timeout_ms =
-      PROTOBUF_GET_MS_OR_DEFAULT(proto_config, message_timeout, kDefaultMessageTimeoutMs);
+      PROTOBUF_GET_MS_OR_DEFAULT(proto_config, message_timeout, DefaultMessageTimeoutMs);
   const auto filter_config = std::make_shared<FilterConfig>(
       proto_config, std::chrono::milliseconds(message_timeout_ms), context.scope(), stats_prefix);
 
@@ -28,6 +26,13 @@ Http::FilterFactoryCb ExternalProcessingFilterConfig::createFilterFactoryFromPro
   };
 }
 
+Router::RouteSpecificFilterConfigConstSharedPtr
+ExternalProcessingFilterConfig::createRouteSpecificFilterConfigTyped(
+    const envoy::extensions::filters::http::ext_proc::v3alpha::ExtProcPerRoute& proto_config,
+    Server::Configuration::ServerFactoryContext&, ProtobufMessage::ValidationVisitor&) {
+  return std::make_shared<FilterConfigPerRoute>(proto_config);
+}
+
 REGISTER_FACTORY(ExternalProcessingFilterConfig,
                  Server::Configuration::NamedHttpFilterConfigFactory){"envoy.ext_proc"};
 

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

@@ -14,17 +14,23 @@ namespace ExternalProcessing {
 
 class ExternalProcessingFilterConfig
     : public Common::FactoryBase<
-          envoy::extensions::filters::http::ext_proc::v3alpha::ExternalProcessor> {
+          envoy::extensions::filters::http::ext_proc::v3alpha::ExternalProcessor,
+          envoy::extensions::filters::http::ext_proc::v3alpha::ExtProcPerRoute> {
 
 public:
   ExternalProcessingFilterConfig() : FactoryBase("envoy.filters.http.ext_proc") {}
 
 private:
-  static constexpr uint64_t kDefaultMessageTimeoutMs = 200;
+  static constexpr uint64_t DefaultMessageTimeoutMs = 200;
 
   Http::FilterFactoryCb createFilterFactoryFromProtoTyped(
       const envoy::extensions::filters::http::ext_proc::v3alpha::ExternalProcessor& proto_config,
       const std::string& stats_prefix, Server::Configuration::FactoryContext& context) override;
+
+  Router::RouteSpecificFilterConfigConstSharedPtr createRouteSpecificFilterConfigTyped(
+      const envoy::extensions::filters::http::ext_proc::v3alpha::ExtProcPerRoute& proto_config,
+      Server::Configuration::ServerFactoryContext& context,
+      ProtobufMessage::ValidationVisitor& validator) override;
 };
 
 } // namespace ExternalProcessing

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

@@ -1,5 +1,6 @@
 #include "source/extensions/filters/http/ext_proc/ext_proc.h"
 
+#include "source/common/http/utility.h"
 #include "source/extensions/filters/http/ext_proc/mutation_utils.h"
 
 #include "absl/strings/str_format.h"
@@ -9,6 +10,7 @@ namespace Extensions {
 namespace HttpFilters {
 namespace ExternalProcessing {
 
+using envoy::extensions::filters::http::ext_proc::v3alpha::ExtProcPerRoute;
 using envoy::extensions::filters::http::ext_proc::v3alpha::ProcessingMode;
 
 using envoy::service::ext_proc::v3alpha::ImmediateResponse;
@@ -23,8 +25,21 @@ using Http::RequestTrailerMap;
 using Http::ResponseHeaderMap;
 using Http::ResponseTrailerMap;
 
-static const std::string kErrorPrefix = "ext_proc error";
+static const std::string ErrorPrefix = "ext_proc error";
 static const int DefaultImmediateStatus = 200;
+static const std::string FilterName = "envoy.filters.http.ext_proc";
+
+FilterConfigPerRoute::FilterConfigPerRoute(const ExtProcPerRoute& config)
+    : disabled_(config.disabled()) {
+  if (config.has_overrides()) {
+    processing_mode_ = config.overrides().processing_mode();
+  }
+}
+
+void FilterConfigPerRoute::merge(const FilterConfigPerRoute& src) {
+  disabled_ = src.disabled_;
+  processing_mode_ = src.processing_mode_;
+}
 
 void Filter::setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) {
   Http::PassThroughFilter::setDecoderFilterCallbacks(callbacks);
@@ -91,6 +106,7 @@ FilterHeadersStatus Filter::onHeaders(ProcessorState& state,
 
 FilterHeadersStatus Filter::decodeHeaders(RequestHeaderMap& headers, bool end_stream) {
   ENVOY_LOG(trace, "decodeHeaders: end_stream = {}", end_stream);
+  mergePerRouteConfig();
   if (end_stream) {
     decoding_state_.setCompleteBodyAvailable(true);
   }
@@ -508,7 +524,7 @@ void Filter::onGrpcError(Grpc::Status::GrpcStatus status) {
     cleanUpTimers();
     ImmediateResponse errorResponse;
     errorResponse.mutable_status()->set_code(envoy::type::v3::StatusCode::InternalServerError);
-    errorResponse.set_details(absl::StrFormat("%s: gRPC error %i", kErrorPrefix, status));
+    errorResponse.set_details(absl::StrFormat("%s: gRPC error %i", ErrorPrefix, status));
     sendImmediateResponse(errorResponse);
   }
 }
@@ -541,7 +557,7 @@ void Filter::onMessageTimeout() {
     encoding_state_.setCallbackState(ProcessorState::CallbackState::Idle);
     ImmediateResponse errorResponse;
     errorResponse.mutable_status()->set_code(envoy::type::v3::StatusCode::InternalServerError);
-    errorResponse.set_details(absl::StrFormat("%s: per-message timeout exceeded", kErrorPrefix));
+    errorResponse.set_details(absl::StrFormat("%s: per-message timeout exceeded", ErrorPrefix));
     sendImmediateResponse(errorResponse);
   }
 }
@@ -582,6 +598,33 @@ void Filter::sendImmediateResponse(const ImmediateResponse& response) {
                                      mutate_headers, grpc_status, response.details());
 }
 
+static ProcessingMode allDisabledMode() {
+  ProcessingMode pm;
+  pm.set_request_header_mode(ProcessingMode::SKIP);
+  pm.set_response_header_mode(ProcessingMode::SKIP);
+  return pm;
+}
+
+void Filter::mergePerRouteConfig() {
+  auto&& merged_config = Http::Utility::getMergedPerFilterConfig<FilterConfigPerRoute>(
+      FilterName, decoder_callbacks_->route(),
+      [](FilterConfigPerRoute& dst, const FilterConfigPerRoute& src) { dst.merge(src); });
+  if (merged_config) {
+    if (merged_config->disabled()) {
+      // Rather than introduce yet another flag, use the processing mode
+      // structure to disable all the callbacks.
+      ENVOY_LOG(trace, "Disabling filter due to per-route configuration");
+      const auto all_disabled = allDisabledMode();
+      decoding_state_.setProcessingMode(all_disabled);
+      encoding_state_.setProcessingMode(all_disabled);
+    } else if (merged_config->processingMode()) {
+      ENVOY_LOG(trace, "Setting new processing mode from per-route configuration");
+      decoding_state_.setProcessingMode(*(merged_config->processingMode()));
+      encoding_state_.setProcessingMode(*(merged_config->processingMode()));
+    }
+  }
+}
+
 std::string responseCaseToString(const ProcessingResponse::ResponseCase response_case) {
   switch (response_case) {
   case ProcessingResponse::ResponseCase::kRequestHeaders:

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

@@ -72,6 +72,25 @@ class FilterConfig {
 
 using FilterConfigSharedPtr = std::shared_ptr<FilterConfig>;
 
+class FilterConfigPerRoute : public Router::RouteSpecificFilterConfig {
+public:
+  explicit FilterConfigPerRoute(
+      const envoy::extensions::filters::http::ext_proc::v3alpha::ExtProcPerRoute& config);
+
+  void merge(const FilterConfigPerRoute& other);
+
+  bool disabled() const { return disabled_; }
+  const absl::optional<envoy::extensions::filters::http::ext_proc::v3alpha::ProcessingMode>&
+  processingMode() const {
+    return processing_mode_;
+  }
+
+private:
+  bool disabled_;
+  absl::optional<envoy::extensions::filters::http::ext_proc::v3alpha::ProcessingMode>
+      processing_mode_;
+};
+
 class Filter : public Logger::Loggable<Logger::Id::filter>,
                public Http::PassThroughFilter,
                public ExternalProcessorCallbacks {
@@ -128,6 +147,7 @@ class Filter : public Logger::Loggable<Logger::Id::filter>,
   void sendTrailers(ProcessorState& state, const Http::HeaderMap& trailers);
 
 private:
+  void mergePerRouteConfig();
   StreamOpenState openStream();
 
   void cleanUpTimers();