http: adding an interface to inform filters of local replies

include/envoy/http/filter.h

@@ -163,6 +163,18 @@ enum class FilterMetadataStatus {
   Continue,
 };
 
+/**
+ * Return codes for onLocalReply filter invocations.
+ */
+enum class LocalErrorStatus {
+  // Continue sending the local reply after onLocalError has been sent to all filters.
+  Continue,
+
+  // Continue sending onLocalReply to all filters, but reset the stream once all filters have been
+  // informed rather than sending the local reply.
+  ContinueAndResetStream,
+};
+
 /**
  * The stream filter callbacks are passed to all filters to use for writing response data and
  * interacting with the underlying stream in general.
@@ -596,6 +608,27 @@ class StreamFilterBase {
    * @param action the resulting match action
    */
   virtual void onMatchCallback(const Matcher::Action&) {}
+
+  struct LocalReplyData {
+    Http::Code code_;
+    absl::string_view details_;
+  };
+
+  /**
+   * Called after sendLocalReply is called, and before any local reply is
+   * serialized either to filters, or downstream.
+   *
+   * Note that in rare circumstances, onLocalReply may be called more than once
+   * for a given stream, because it is possible that a filter call
+   * sendLocalReply while processing the original local reply response.
+   *
+   * Filters implementing onLocalReply are responsible for never calling sendLocalReply
+   * from onLocalReply, as that has the potential for looping.
+   *
+   * @param data data associated with the sendLocalReply call.
+   * @param LocalErrorStatus the action to take after onLocalError completes.
+   */
+  virtual LocalErrorStatus onLocalReply(LocalReplyData&) { return LocalErrorStatus::Continue; }
 };
 
 /**

source/common/http/filter_manager.cc

@@ -812,6 +812,18 @@ FilterManager::commonDecodePrefix(ActiveStreamDecoderFilter* filter,
   return std::next(filter->entry());
 }
 
+LocalErrorStatus FilterManager::onLocalReply(StreamFilterBase::LocalReplyData& data) {
+  filter_manager_callbacks_.onLocalReply(data.code_);
+
+  LocalErrorStatus status = LocalErrorStatus::Continue;
+  for (auto entry : filters_) {
+    if (entry->onLocalReply(data) == LocalErrorStatus::ContinueAndResetStream) {
+      status = LocalErrorStatus::ContinueAndResetStream;
+    }
+  }
+  return status;
+}
+
 void FilterManager::sendLocalReply(
     bool old_was_grpc_request, Code code, absl::string_view body,
     const std::function<void(ResponseHeaderMap& headers)>& modify_headers,
@@ -824,7 +836,13 @@ void FilterManager::sendLocalReply(
 
   stream_info_.setResponseCodeDetails(details);
 
-  filter_manager_callbacks_.onLocalReply(code);
+  StreamFilterBase::LocalReplyData data{code, details};
+  if (FilterManager::onLocalReply(data) == LocalErrorStatus::ContinueAndResetStream) {
+    ENVOY_STREAM_LOG(debug, "Resetting stream due to {}. onLocalReply requested reset.", *this,
+                     details);
+    filter_manager_callbacks_.resetStream();
+    return;
+  }
 
   if (!filter_manager_callbacks_.responseHeaders().has_value()) {
     // If the response has not started at all, send the response through the filter chain.

source/common/http/filter_manager.h

@@ -760,6 +760,7 @@ class FilterManager : public ScopeTrackedObject,
   // Http::FilterChainFactoryCallbacks
   void addStreamDecoderFilter(StreamDecoderFilterSharedPtr filter) override {
     addStreamDecoderFilterWorker(filter, nullptr, false);
+    filters_.push_back(filter.get());
   }
   void addStreamDecoderFilter(StreamDecoderFilterSharedPtr filter,
                               Matcher::MatchTreeSharedPtr<HttpMatchingData> match_tree) override {
@@ -776,6 +777,7 @@ class FilterManager : public ScopeTrackedObject,
   }
   void addStreamEncoderFilter(StreamEncoderFilterSharedPtr filter) override {
     addStreamEncoderFilterWorker(filter, nullptr, false);
+    filters_.push_back(filter.get());
   }
   void addStreamEncoderFilter(StreamEncoderFilterSharedPtr filter,
                               Matcher::MatchTreeSharedPtr<HttpMatchingData> match_tree) override {
@@ -793,6 +795,8 @@ class FilterManager : public ScopeTrackedObject,
   void addStreamFilter(StreamFilterSharedPtr filter) override {
     addStreamDecoderFilterWorker(filter, nullptr, true);
     addStreamEncoderFilterWorker(filter, nullptr, true);
+    StreamDecoderFilter* decoder_filter = filter.get();
+    filters_.push_back(decoder_filter);
   }
   void addStreamFilter(StreamFilterSharedPtr filter,
                        Matcher::MatchTreeSharedPtr<HttpMatchingData> match_tree) override {
@@ -910,6 +914,13 @@ class FilterManager : public ScopeTrackedObject,
    */
   void maybeEndEncode(bool end_stream);
 
+  /**
+   * Called before local reply is made by the filter manager.
+   * @param data the data associated with the local reply.
+   * @param LocalErrorStatus the status from the filter chain.
+   */
+  LocalErrorStatus onLocalReply(StreamFilterBase::LocalReplyData& data);
+
   void sendLocalReply(bool is_grpc_request, Code code, absl::string_view body,
                       const std::function<void(ResponseHeaderMap& headers)>& modify_headers,
                       const absl::optional<Grpc::Status::GrpcStatus> grpc_status,
@@ -1061,6 +1072,7 @@ class FilterManager : public ScopeTrackedObject,
 
   std::list<ActiveStreamDecoderFilterPtr> decoder_filters_;
   std::list<ActiveStreamEncoderFilterPtr> encoder_filters_;
+  std::list<StreamFilterBase*> filters_;
   std::list<AccessLog::InstanceSharedPtr> access_log_handlers_;
 
   // Stores metadata added in the decoding filter that is being processed. Will be cleared before

test/integration/BUILD

@@ -357,6 +357,7 @@ envoy_cc_test(
         "//source/extensions/filters/http/buffer:config",
         "//source/extensions/filters/http/health_check:config",
         "//test/integration/filters:metadata_stop_all_filter_config_lib",
+        "//test/integration/filters:on_local_reply_filter_config_lib",
         "//test/integration/filters:request_metadata_filter_config_lib",
         "//test/integration/filters:response_metadata_filter_config_lib",
         "//test/integration/filters:set_response_code_filter_config_proto_cc_proto",

test/integration/filters/BUILD

@@ -157,6 +157,20 @@ envoy_cc_test_library(
     ],
 )
 
+envoy_cc_test_library(
+    name = "on_local_reply_filter_config_lib",
+    srcs = [
+        "on_local_reply_filter.cc",
+    ],
+    deps = [
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/registry",
+        "//include/envoy/server:filter_config_interface",
+        "//source/extensions/filters/http/common:pass_through_filter_lib",
+        "//test/extensions/filters/http/common:empty_http_filter_config_lib",
+    ],
+)
+
 envoy_cc_test_library(
     name = "passthrough_filter_config_lib",
     srcs = [

test/integration/filters/on_local_reply_filter.cc

@@ -0,0 +1,49 @@
+#include <string>
+
+#include "envoy/http/filter.h"
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+
+#include "extensions/filters/http/common/pass_through_filter.h"
+
+#include "test/extensions/filters/http/common/empty_http_filter_config.h"
+
+namespace Envoy {
+
+class OnLocalReplyFilter : public Http::PassThroughFilter {
+public:
+  Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap& request_headers, bool) override {
+    if (request_headers.get(Http::LowerCaseString("reset")).size() != 0) {
+      reset_ = true;
+    }
+    decoder_callbacks_->sendLocalReply(Http::Code::BadRequest, "body", nullptr, absl::nullopt,
+                                       "details");
+    return Http::FilterHeadersStatus::StopIteration;
+  }
+
+  Http::LocalErrorStatus onLocalReply(LocalReplyData&) override {
+    if (reset_) {
+      return Http::LocalErrorStatus::ContinueAndResetStream;
+    }
+    return Http::LocalErrorStatus::Continue;
+  }
+
+  bool reset_{};
+};
+
+class OnLocalReplyFilterConfig : public Extensions::HttpFilters::Common::EmptyHttpFilterConfig {
+public:
+  OnLocalReplyFilterConfig() : EmptyHttpFilterConfig("on-local-reply-filter") {}
+  Http::FilterFactoryCb createFilter(const std::string&,
+                                     Server::Configuration::FactoryContext&) override {
+    return [](Http::FilterChainFactoryCallbacks& callbacks) -> void {
+      callbacks.addStreamFilter(std::make_shared<::Envoy::OnLocalReplyFilter>());
+    };
+  }
+};
+
+// perform static registration
+static Registry::RegisterFactory<OnLocalReplyFilterConfig,
+                                 Server::Configuration::NamedHttpFilterConfigFactory>
+    register_;
+} // namespace Envoy

test/integration/http2_integration_test.cc

@@ -1589,6 +1589,7 @@ TEST_P(Http2MetadataIntegrationTest, UpstreamMetadataAfterEndStream) {
   upstream_request_->encodeHeaders(response_headers, true);
 
   // Upstream sends metadata.
+
   const Http::MetadataMap response_metadata_map = {{"resp_key1", "resp_value1"}};
   Http::MetadataMapPtr metadata_map_ptr =
       std::make_unique<Http::MetadataMap>(response_metadata_map);
@@ -1603,4 +1604,28 @@ TEST_P(Http2MetadataIntegrationTest, UpstreamMetadataAfterEndStream) {
   EXPECT_EQ("200", response->headers().getStatusValue());
 }
 
+static std::string on_local_reply_filter = R"EOF(
+name: on-local-reply-filter
+typed_config:
+  "@type": type.googleapis.com/google.protobuf.Empty
+)EOF";
+
+TEST_P(Http2IntegrationTest, OnLocalReply) {
+  config_helper_.addFilter(on_local_reply_filter);
+  initialize();
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  {
+    auto response = codec_client_->makeHeaderOnlyRequest(default_request_headers_);
+    response->waitForEndStream();
+    ASSERT_TRUE(response->complete());
+  }
+  {
+    default_request_headers_.addCopy("reset", "yes");
+    auto response = codec_client_->makeHeaderOnlyRequest(default_request_headers_);
+    response->waitForReset();
+    ASSERT_FALSE(response->complete());
+  }
+}
+
 } // namespace Envoy