Adding MultiActions for composite filters. WIP

api/envoy/extensions/filters/http/composite/v3/composite.proto

@@ -37,3 +37,10 @@ message Composite {
 message ExecuteFilterAction {
   config.core.v3.TypedExtensionConfig typed_config = 1;
 }
+
+// Composite match action (see :ref:`matching docs <arch_overview_matching_api>` for more info on match actions).
+// This allows multiple delegate filters to be linked to one action, executing individual parts in parallel.
+// The final filter in the list is the filter that returns the resulting status, all other filter statuses are dropped.
+message ExecuteFilterMultiAction {
+  repeated config.core.v3.TypedExtensionConfig typed_config = 1;
+}

source/extensions/filters/http/composite/action.cc

@@ -9,6 +9,16 @@ void ExecuteFilterAction::createFilters(Http::FilterChainFactoryCallbacks& callb
 }
 REGISTER_FACTORY(ExecuteFilterActionFactory,
                  Matcher::ActionFactory<Http::Matching::HttpFilterActionContext>);
+
+void ExecuteFilterMultiAction::createFilters(
+    std::function<void(Envoy::Http::FilterFactoryCb&)> parse_wrapper) const {
+  for (auto cb : cb_list_) {
+    parse_wrapper(cb);
+  }
+}
+REGISTER_FACTORY(ExecuteFilterMultiActionFactory,
+                 Matcher::ActionFactory<Http::Matching::HttpFilterActionContext>);
+
 } // namespace Composite
 } // namespace HttpFilters
 } // namespace Extensions

source/extensions/filters/http/composite/action.h

@@ -52,6 +52,51 @@ class ExecuteFilterActionFactory
 
 DECLARE_FACTORY(ExecuteFilterActionFactory);
 
+class ExecuteFilterMultiAction
+    : public Matcher::ActionBase<
+          envoy::extensions::filters::http::composite::v3::ExecuteFilterMultiAction> {
+public:
+  explicit ExecuteFilterMultiAction(std::vector<Http::FilterFactoryCb> cb_list)
+      : cb_list_(std::move(cb_list)) {}
+
+  void createFilters(std::function<void(Http::FilterFactoryCb&)> parse_wrapper) const;
+
+private:
+  std::vector<Http::FilterFactoryCb> cb_list_;
+};
+
+class ExecuteFilterMultiActionFactory
+    : public Matcher::ActionFactory<Http::Matching::HttpFilterActionContext> {
+public:
+  std::string name() const override { return "composite-multi-action"; }
+  Matcher::ActionFactoryCb
+  createActionFactoryCb(const Protobuf::Message& config,
+                        Envoy::Http::Matching::HttpFilterActionContext& context,
+                        ProtobufMessage::ValidationVisitor& validation) override {
+    const auto& composite_action = MessageUtil::downcastAndValidate<
+        const envoy::extensions::filters::http::composite::v3::ExecuteFilterMultiAction&>(
+        config, validation);
+    std::vector<Http::FilterFactoryCb> callback_list;
+    for (auto typed_config : composite_action.typed_config()) {
+      auto& factory =
+          Config::Utility::getAndCheckFactory<Server::Configuration::NamedHttpFilterConfigFactory>(
+              typed_config);
+      ProtobufTypes::MessagePtr message = Config::Utility::translateAnyToFactoryConfig(
+          typed_config.typed_config(), validation, factory);
+      callback_list.push_back(factory.createFilterFactoryFromProto(*message, context.stat_prefix_,
+                                                                   context.factory_context_));
+    }
+    return [cb_list = std::move(callback_list)]() -> Matcher::ActionPtr {
+      return std::make_unique<ExecuteFilterMultiAction>(cb_list);
+    };
+  }
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    return std::make_unique<
+        envoy::extensions::filters::http::composite::v3::ExecuteFilterMultiAction>();
+  }
+};
+
+DECLARE_FACTORY(ExecuteFilterMultiActionFactory);
 } // namespace Composite
 } // namespace HttpFilters
 } // namespace Extensions

source/extensions/filters/http/composite/filter.cc

@@ -10,77 +10,83 @@ namespace HttpFilters {
 namespace Composite {
 
 namespace {
+// Helper that returns the last `filter->func(args...)` result in the list of filters, if the list
+// is empty, return `rval`.
+template <class FilterPtrT, class FuncT, class RValT, class... Args>
+RValT delegateAllFilterActionOr(std::vector<FilterPtrT>& filter_list, FuncT func, RValT rval,
+                                Args&&... args) {
+  RValT value = rval;
+  for (auto filter = filter_list.rbegin(); filter != filter_list.rend(); ++filter) {
+    value = ((**filter).*func)(std::forward<Args>(args)...);
+  }
+  return value;
+}
 // Helper that returns `filter->func(args...)` if the filter is not null, returning `rval`
 // otherwise.
 template <class FilterPtrT, class FuncT, class RValT, class... Args>
 RValT delegateFilterActionOr(FilterPtrT& filter, FuncT func, RValT rval, Args&&... args) {
   if (filter) {
     return ((*filter).*func)(std::forward<Args>(args)...);
   }
-
   return rval;
 }
 } // namespace
 Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers, bool end_stream) {
   decoded_headers_ = true;
 
-  return delegateFilterActionOr(delegated_filter_, &StreamDecoderFilter::decodeHeaders,
-                                Http::FilterHeadersStatus::Continue, headers, end_stream);
+  return delegateAllFilterActionOr(delegated_filter_list_, &StreamDecoderFilter::decodeHeaders,
+                                   Http::FilterHeadersStatus::Continue, headers, end_stream);
 }
 
 Http::FilterDataStatus Filter::decodeData(Buffer::Instance& data, bool end_stream) {
-  return delegateFilterActionOr(delegated_filter_, &StreamDecoderFilter::decodeData,
-                                Http::FilterDataStatus::Continue, data, end_stream);
+  return delegateAllFilterActionOr(delegated_filter_list_, &StreamDecoderFilter::decodeData,
+                                   Http::FilterDataStatus::Continue, data, end_stream);
 }
 
 Http::FilterTrailersStatus Filter::decodeTrailers(Http::RequestTrailerMap& trailers) {
-  return delegateFilterActionOr(delegated_filter_, &StreamDecoderFilter::decodeTrailers,
-                                Http::FilterTrailersStatus::Continue, trailers);
+  return delegateAllFilterActionOr(delegated_filter_list_, &StreamDecoderFilter::decodeTrailers,
+                                   Http::FilterTrailersStatus::Continue, trailers);
 }
 
 Http::FilterMetadataStatus Filter::decodeMetadata(Http::MetadataMap& metadata_map) {
-  return delegateFilterActionOr(delegated_filter_, &StreamDecoderFilter::decodeMetadata,
-                                Http::FilterMetadataStatus::Continue, metadata_map);
+  return delegateAllFilterActionOr(delegated_filter_list_, &StreamDecoderFilter::decodeMetadata,
+                                   Http::FilterMetadataStatus::Continue, metadata_map);
 }
 
 void Filter::decodeComplete() {
-  if (delegated_filter_) {
-    delegated_filter_->decodeComplete();
+  for (auto delegated_filter : delegated_filter_list_) {
+    delegated_filter->decodeComplete();
   }
 }
 
 Http::Filter1xxHeadersStatus Filter::encode1xxHeaders(Http::ResponseHeaderMap& headers) {
-  return delegateFilterActionOr(delegated_filter_, &StreamEncoderFilter::encode1xxHeaders,
-                                Http::Filter1xxHeadersStatus::Continue, headers);
+  return delegateAllFilterActionOr(delegated_filter_list_, &StreamEncoderFilter::encode1xxHeaders,
+                                   Http::Filter1xxHeadersStatus::Continue, headers);
 }
 
 Http::FilterHeadersStatus Filter::encodeHeaders(Http::ResponseHeaderMap& headers, bool end_stream) {
-  return delegateFilterActionOr(delegated_filter_, &StreamEncoderFilter::encodeHeaders,
-                                Http::FilterHeadersStatus::Continue, headers, end_stream);
+  return delegateAllFilterActionOr(delegated_filter_list_, &StreamEncoderFilter::encodeHeaders,
+                                   Http::FilterHeadersStatus::Continue, headers, end_stream);
 }
 Http::FilterDataStatus Filter::encodeData(Buffer::Instance& data, bool end_stream) {
-  return delegateFilterActionOr(delegated_filter_, &StreamEncoderFilter::encodeData,
-                                Http::FilterDataStatus::Continue, data, end_stream);
+  return delegateAllFilterActionOr(delegated_filter_list_, &StreamEncoderFilter::encodeData,
+                                   Http::FilterDataStatus::Continue, data, end_stream);
 }
 Http::FilterTrailersStatus Filter::encodeTrailers(Http::ResponseTrailerMap& trailers) {
-  return delegateFilterActionOr(delegated_filter_, &StreamEncoderFilter::encodeTrailers,
-                                Http::FilterTrailersStatus::Continue, trailers);
+  return delegateAllFilterActionOr(delegated_filter_list_, &StreamEncoderFilter::encodeTrailers,
+                                   Http::FilterTrailersStatus::Continue, trailers);
 }
 Http::FilterMetadataStatus Filter::encodeMetadata(Http::MetadataMap& metadata_map) {
-  return delegateFilterActionOr(delegated_filter_, &StreamEncoderFilter::encodeMetadata,
-                                Http::FilterMetadataStatus::Continue, metadata_map);
+  return delegateAllFilterActionOr(delegated_filter_list_, &StreamEncoderFilter::encodeMetadata,
+                                   Http::FilterMetadataStatus::Continue, metadata_map);
 }
 void Filter::encodeComplete() {
-  if (delegated_filter_) {
-    delegated_filter_->encodeComplete();
+  for (auto delegated_filter : delegated_filter_list_) {
+    delegated_filter->encodeComplete();
   }
 }
-void Filter::onMatchCallback(const Matcher::Action& action) {
-  const auto& composite_action = action.getTyped<ExecuteFilterAction>();
-
-  FactoryCallbacksWrapper wrapper(*this, dispatcher_);
-  composite_action.createFilters(wrapper);
 
+void Filter::ApplyWrapper(FactoryCallbacksWrapper& wrapper) {
   if (!wrapper.errors_.empty()) {
     stats_.filter_delegation_error_.inc();
     ENVOY_LOG(debug, "failed to create delegated filter {}",
@@ -91,19 +97,21 @@ void Filter::onMatchCallback(const Matcher::Action& action) {
 
   if (wrapper.filter_to_inject_) {
     stats_.filter_delegation_success_.inc();
+    Http::StreamFilterSharedPtr delegated_filter;
     if (absl::holds_alternative<Http::StreamDecoderFilterSharedPtr>(*wrapper.filter_to_inject_)) {
-      delegated_filter_ = std::make_shared<StreamFilterWrapper>(
+      delegated_filter = std::make_shared<StreamFilterWrapper>(
           absl::get<Http::StreamDecoderFilterSharedPtr>(*wrapper.filter_to_inject_));
     } else if (absl::holds_alternative<Http::StreamEncoderFilterSharedPtr>(
                    *wrapper.filter_to_inject_)) {
-      delegated_filter_ = std::make_shared<StreamFilterWrapper>(
+      delegated_filter = std::make_shared<StreamFilterWrapper>(
           absl::get<Http::StreamEncoderFilterSharedPtr>(*wrapper.filter_to_inject_));
     } else {
-      delegated_filter_ = absl::get<Http::StreamFilterSharedPtr>(*wrapper.filter_to_inject_);
+      delegated_filter = absl::get<Http::StreamFilterSharedPtr>(*wrapper.filter_to_inject_);
     }
 
-    delegated_filter_->setDecoderFilterCallbacks(*decoder_callbacks_);
-    delegated_filter_->setEncoderFilterCallbacks(*encoder_callbacks_);
+    delegated_filter->setDecoderFilterCallbacks(*decoder_callbacks_);
+    delegated_filter->setEncoderFilterCallbacks(*encoder_callbacks_);
+    delegated_filter_list_.push_back(delegated_filter);
 
     // Size should be small, so a copy should be fine.
     access_loggers_.insert(access_loggers_.end(), wrapper.access_loggers_.begin(),
@@ -114,6 +122,23 @@ void Filter::onMatchCallback(const Matcher::Action& action) {
   // either directly or via some return status.
 }
 
+void Filter::onMatchCallback(const Matcher::Action& action) {
+  if (action.typeUrl() == ExecuteFilterAction::staticTypeUrl()) {
+    const auto& composite_action = action.getTyped<ExecuteFilterAction>();
+    FactoryCallbacksWrapper wrapper(*this, dispatcher_);
+    composite_action.createFilters(wrapper);
+    ApplyWrapper(wrapper);
+  } else if (action.typeUrl() == ExecuteFilterMultiAction::staticTypeUrl()) {
+    const auto& composite_multi_action = action.getTyped<ExecuteFilterMultiAction>();
+    std::function<void(Http::FilterFactoryCb&)> parse_wrapper = [this](Http::FilterFactoryCb& cb) {
+      FactoryCallbacksWrapper wrapper(*this, dispatcher_);
+      cb(wrapper);
+      ApplyWrapper(wrapper);
+    };
+    composite_multi_action.createFilters(parse_wrapper);
+  }
+}
+
 Http::FilterHeadersStatus
 Filter::StreamFilterWrapper::decodeHeaders(Http::RequestHeaderMap& headers, bool end_stream) {
   return delegateFilterActionOr(decoder_filter_, &StreamDecoderFilter::decodeHeaders,

source/extensions/filters/http/composite/filter.h

@@ -57,10 +57,10 @@ class Filter : public Http::StreamFilter,
 
   // Http::StreamFilterBase
   void onDestroy() override {
-    if (delegated_filter_) {
+    for (auto delegated_filter : delegated_filter_list_) {
       // We need to explicitly specify which base class to the conversion via due
       // to the diamond inheritance between StreamFilter and StreamFilterBase.
-      static_cast<Http::StreamDecoderFilter&>(*delegated_filter_).onDestroy();
+      static_cast<Http::StreamDecoderFilter&>(*delegated_filter).onDestroy();
     }
   }
 
@@ -87,6 +87,8 @@ class Filter : public Http::StreamFilter,
   // headers, this just provides some additional sanity checking.
   bool decoded_headers_ : 1;
 
+  void ApplyWrapper(FactoryCallbacksWrapper& wrapper);
+
   // Wraps a stream encoder OR a stream decoder filter into a stream filter, making it easier to
   // delegate calls.
   struct StreamFilterWrapper : public Http::StreamFilter {
@@ -124,9 +126,9 @@ class Filter : public Http::StreamFilter,
   };
   std::vector<AccessLog::InstanceSharedPtr> access_loggers_;
 
-  Http::StreamFilterSharedPtr delegated_filter_;
-  Http::StreamEncoderFilterCallbacks* encoder_callbacks_{};
-  Http::StreamDecoderFilterCallbacks* decoder_callbacks_{};
+  std::vector<Http::StreamFilterSharedPtr> delegated_filter_list_;
+  Http::StreamEncoderFilterCallbacks* encoder_callbacks_;
+  Http::StreamDecoderFilterCallbacks* decoder_callbacks_;
   FilterStats& stats_;
 };
 

test/extensions/filters/http/composite/BUILD

@@ -34,6 +34,7 @@ envoy_extension_cc_test(
         "//source/extensions/filters/http/composite:filter_lib",
         "//test/integration:http_integration_lib",
         "//test/integration/filters:set_response_code_filter_lib",
+        "@envoy_api//envoy/extensions/common/matching/v3:pkg_cc_proto",
         "@envoy_api//envoy/extensions/filters/network/http_connection_manager/v3:pkg_cc_proto",
     ],
 )

test/extensions/filters/http/composite/composite_filter_integration_test.cc

@@ -1,3 +1,4 @@
+#include "envoy/extensions/common/matching/v3/extension_matcher.pb.h"
 #include "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.pb.h"
 
 #include "test/integration/http_integration.h"
@@ -41,6 +42,20 @@ class CompositeFilterIntegrationTest : public testing::TestWithParam<Network::Ad
                     typed_config:
                       "@type": type.googleapis.com/test.integration.filters.SetResponseCodeFilterConfig
                       code: 403
+            multi-match:
+              action:
+                name: composite-multi-action
+                typed_config:
+                  "@type": type.googleapis.com/envoy.extensions.filters.http.composite.v3.ExecuteFilterMultiAction
+                  typed_config:
+                  - name: set-response-code
+                    typed_config:
+                      "@type": type.googleapis.com/test.integration.filters.SetResponseCodeFilterConfig
+                      code: 403
+                  - name: set-response-code2
+                    typed_config:
+                      "@type": type.googleapis.com/test.integration.filters.SetResponseCodeFilterConfig
+                      code: 404
     )EOF");
     HttpIntegrationTest::initialize();
   }
@@ -76,5 +91,19 @@ TEST_P(CompositeFilterIntegrationTest, TestBasic) {
     EXPECT_THAT(response->headers(), Http::HttpStatusIs("403"));
   }
 }
-
+// Verifies that multi-actions are preformed in order of definition.
+TEST_P(CompositeFilterIntegrationTest, TestBasicMultiAction) {
+  initialize();
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  {
+    const Http::TestRequestHeaderMapImpl request_headers = {{":method", "GET"},
+                                                            {":path", "/somepath"},
+                                                            {":scheme", "http"},
+                                                            {"match-header", "multi-match"},
+                                                            {":authority", "blah"}};
+    auto response = codec_client_->makeRequestWithBody(request_headers, 1024);
+    ASSERT_TRUE(response->waitForEndStream());
+    EXPECT_THAT(response->headers(), Http::HttpStatusIs("404"));
+  }
+}
 } // namespace Envoy