Common: Introduce StopAllIteration filter status for decoding and encoding filters

include/envoy/http/filter.h

@@ -33,7 +33,15 @@ enum class FilterHeadersStatus {
   StopIteration,
   // Continue iteration to remaining filters, but ignore any subsequent data or trailers. This
   // results in creating a header only request/response.
-  ContinueAndEndStream
+  ContinueAndEndStream,
+  // Stop iterations for headers as well as data and trailers that follows the headers. Returning
+  // StopAllTypesIteration means the filter that returns the status and all the filters following
+  // that filter stop processing. The filters before are not impacted. continueDecoding() MUST be
+  // called if continued filter iteration is desired. To avoid buffering large amount of data, a
+  // rate-limiting filter can be added before the filter that can return StopAllTypesIteration.
+  // Only used in decoding path.
+  // TODO(soya3129): add tests for encoding path, and remove the decoding path condition.
+  StopAllTypesIteration
 };
 
 /**

source/common/http/conn_manager_impl.cc

@@ -1498,6 +1498,7 @@ void ConnectionManagerImpl::ActiveStreamFilterBase::commonContinue() {
                    static_cast<const void*>(this));
   ASSERT(stopped_);
   stopped_ = false;
+  stopped_all_ = false;
 
   // Only resume with do100ContinueHeaders() if we've actually seen a 100-Continue.
   if (parent_.has_continue_headers_ && !continue_headers_continued_) {
@@ -1553,6 +1554,10 @@ bool ConnectionManagerImpl::ActiveStreamFilterBase::commonHandleAfterHeadersCall
   if (status == FilterHeadersStatus::StopIteration) {
     stopped_ = true;
     return false;
+  } else if (status == FilterHeadersStatus::StopAllTypesIteration) {
+    stopped_ = true;
+    stopped_all_ = true;
+    return false;
   } else if (status == FilterHeadersStatus::ContinueAndEndStream) {
     // Set headers_only to true so we know to end early if necessary,
     // but continue filter iteration so we actually write the headers/run the cleanup code.
@@ -1589,7 +1594,9 @@ bool ConnectionManagerImpl::ActiveStreamFilterBase::commonHandleAfterDataCallbac
   if (status == FilterDataStatus::Continue) {
     if (stopped_) {
       commonHandleBufferData(provided_data);
-      commonContinue();
+      if (!stopped_all_) {
+        commonContinue();
+      }
       return false;
     } else {
       ASSERT(headers_continued_);
@@ -1613,7 +1620,9 @@ bool ConnectionManagerImpl::ActiveStreamFilterBase::commonHandleAfterTrailersCal
 
   if (status == FilterTrailersStatus::Continue) {
     if (stopped_) {
-      commonContinue();
+      if (!stopped_all_) {
+        commonContinue();
+      }
       return false;
     } else {
       ASSERT(headers_continued_);

source/common/http/conn_manager_impl.h

@@ -96,7 +96,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
   struct ActiveStreamFilterBase : public virtual StreamFilterCallbacks {
     ActiveStreamFilterBase(ActiveStream& parent, bool dual_filter)
         : parent_(parent), headers_continued_(false), continue_headers_continued_(false),
-          stopped_(false), end_stream_(false), dual_filter_(dual_filter) {}
+          stopped_(false), stopped_all_(false), end_stream_(false), dual_filter_(dual_filter) {}
 
     bool commonHandleAfter100ContinueHeadersCallback(FilterHeadersStatus status);
     bool commonHandleAfterHeadersCallback(FilterHeadersStatus status, bool& headers_only);
@@ -132,6 +132,9 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     bool headers_continued_ : 1;
     bool continue_headers_continued_ : 1;
     bool stopped_ : 1;
+    // If true, filter iteration has been stopped for this filter and all the filters followed on
+    // all frame types.
+    bool stopped_all_ : 1;
     // If true, end_stream is called for this filter.
     bool end_stream_ : 1;
     const bool dual_filter_ : 1;

test/integration/BUILD

@@ -219,6 +219,8 @@ envoy_cc_test_library(
         "//source/extensions/filters/network/http_connection_manager:config",
         "//test/common/upstream:utility_lib",
         "//test/integration/filters:add_trailers_filter_config_lib",
+        "//test/integration/filters:call_decodedata_once_filter_config_lib",
+        "//test/integration/filters:decode_headers_return_stop_all_filter_config_lib",
         "//test/integration/filters:headers_only_filter_config_lib",
         "//test/integration/filters:passthrough_filter_config_lib",
         "//test/integration/filters:pause_filter_lib",

test/integration/filters/BUILD

@@ -107,3 +107,35 @@ envoy_cc_test_library(
         "//source/extensions/filters/http/common:pass_through_filter_lib",
     ],
 )
+
+envoy_cc_test_library(
+    name = "decode_headers_return_stop_all_filter_config_lib",
+    srcs = [
+        "decode_headers_return_stop_all_filter.cc",
+    ],
+    deps = [
+        ":common_lib",
+        "//include/envoy/event:timer_interface",
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/registry",
+        "//include/envoy/server:filter_config_interface",
+        "//source/extensions/filters/http/common:empty_http_filter_config_lib",
+        "//source/extensions/filters/http/common:pass_through_filter_lib",
+    ],
+)
+
+envoy_cc_test_library(
+    name = "call_decodedata_once_filter_config_lib",
+    srcs = [
+        "call_decodedata_once_filter.cc",
+    ],
+    deps = [
+        ":common_lib",
+        "//include/envoy/event:timer_interface",
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/registry",
+        "//include/envoy/server:filter_config_interface",
+        "//source/extensions/filters/http/common:empty_http_filter_config_lib",
+        "//source/extensions/filters/http/common:pass_through_filter_lib",
+    ],
+)

test/integration/filters/call_decodedata_once_filter.cc

@@ -0,0 +1,40 @@
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+
+#include "extensions/filters/http/common/empty_http_filter_config.h"
+#include "extensions/filters/http/common/pass_through_filter.h"
+
+#include "test/integration/filters/common.h"
+
+namespace Envoy {
+
+// A filter that only allows decodeData() to be called once. Multiple calls will result in assert
+// failure.
+class CallDecodeDataOnceFilter : public Http::PassThroughFilter {
+public:
+  constexpr static char name[] = "call-decodedata-once-filter";
+
+  Http::FilterHeadersStatus decodeHeaders(Http::HeaderMap&, bool) override {
+    return Http::FilterHeadersStatus::Continue;
+  }
+
+  Http::FilterDataStatus decodeData(Buffer::Instance&, bool) override {
+    ASSERT(call_count_ == 0);
+    call_count_++;
+    return Http::FilterDataStatus::Continue;
+  }
+
+  Http::FilterTrailersStatus decodeTrailers(Http::HeaderMap&) override {
+    return Http::FilterTrailersStatus::Continue;
+  }
+
+private:
+  int call_count_ = 0;
+};
+
+constexpr char CallDecodeDataOnceFilter::name[];
+static Registry::RegisterFactory<SimpleFilterConfig<CallDecodeDataOnceFilter>,
+                                 Server::Configuration::NamedHttpFilterConfigFactory>
+    register_;
+
+} // namespace Envoy

test/integration/filters/decode_headers_return_stop_all_filter.cc

@@ -0,0 +1,67 @@
+#include <chrono>
+#include <string>
+
+#include "envoy/event/timer.h"
+#include "envoy/http/filter.h"
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+
+#include "extensions/filters/http/common/empty_http_filter_config.h"
+#include "extensions/filters/http/common/pass_through_filter.h"
+
+#include "test/integration/filters/common.h"
+
+namespace Envoy {
+
+// A filter returns StopAllTypesIteration for headers. The iteration continues when end_stream is
+// received.
+class DecodeHeadersReturnStopAllFilter : public Http::PassThroughFilter {
+public:
+  constexpr static char name[] = "decode-headers-return-stop-all-filter";
+
+  // Returns Http::FilterHeadersStatus::StopAllTypesIteration for headers.
+  Http::FilterHeadersStatus decodeHeaders(Http::HeaderMap&, bool) override {
+    return Http::FilterHeadersStatus::StopAllTypesIteration;
+  }
+
+  // Continues the iteration 2s after end_stream is received.
+  Http::FilterDataStatus decodeData(Buffer::Instance&, bool end_stream) override {
+    call_count_++;
+    if (end_stream) {
+      // Verifies decodeData() is called more than once.
+      ASSERT(call_count_ > 1);
+      continued_ = true;
+      createTimerForContinue();
+    }
+    return Http::FilterDataStatus::Continue;
+  }
+
+  // Continues the iteration 2s after trailers are received.
+  Http::FilterTrailersStatus decodeTrailers(Http::HeaderMap&) override {
+    if (!continued_) {
+      // Verifies decodeData() is called more than once.
+      ASSERT(call_count_ > 1);
+      createTimerForContinue();
+    }
+    return Http::FilterTrailersStatus::Continue;
+  }
+
+private:
+  // Creates a timer to continue iteration after 2s.
+  void createTimerForContinue() {
+    delay_timer_ = decoder_callbacks_->dispatcher().createTimer(
+        [this]() -> void { decoder_callbacks_->continueDecoding(); });
+    delay_timer_->enableTimer(std::chrono::seconds(2));
+  }
+
+  Event::TimerPtr delay_timer_;
+  int call_count_ = 0;
+  bool continued_ = false;
+};
+
+constexpr char DecodeHeadersReturnStopAllFilter::name[];
+static Registry::RegisterFactory<SimpleFilterConfig<DecodeHeadersReturnStopAllFilter>,
+                                 Server::Configuration::NamedHttpFilterConfigFactory>
+    register_;
+
+} // namespace Envoy

test/integration/http2_integration_test.cc

@@ -477,6 +477,8 @@ TEST_P(Http2IntegrationTest, DecodingHeaderOnlyResponse) { testHeadersOnlyFilter
 
 TEST_P(Http2IntegrationTest, DecodingHeaderOnlyInterleaved) { testHeadersOnlyFilterInterleaved(); }
 
+TEST_P(Http2IntegrationTest, DecodeHeadersReturnsStopAll) { testDecodeHeadersReturnsStopAll(); }
+
 TEST_P(Http2IntegrationTest, DownstreamResetBeforeResponseComplete) {
   testDownstreamResetBeforeResponseComplete();
 }

test/integration/http_integration.cc

@@ -1636,4 +1636,58 @@ std::string HttpIntegrationTest::listenerStatPrefix(const std::string& stat_name
   return "listener.[__1]_0." + stat_name;
 }
 
+// Tests StopAllTypesIteration. decode-headers-return-stop-all-filter returns StopAllTypesIteration
+// for headers. As a result, call-decodedata-once-filter that follows headers-return-stop-all-filter
+// will stop processing headers, data and trailer until the iteration resumes. Verifies decodeData()
+// is called multiple times in decode-headers-return-stop-all-filter, and only called once in
+// call-decodedata-once-filter after iteration resumes.
+void HttpIntegrationTest::testDecodeHeadersReturnsStopAll() {
+  config_helper_.addFilter(R"EOF(
+name: call-decodedata-once-filter
+)EOF");
+  config_helper_.addFilter(R"EOF(
+name: decode-headers-return-stop-all-filter
+)EOF");
+  config_helper_.addFilter(R"EOF(
+name: passthrough-filter
+)EOF");
+
+  initialize();
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  // Sends a request with headers and data.
+  auto encoder_decoder = codec_client_->startRequest(default_request_headers_);
+  request_encoder_ = &encoder_decoder.first;
+  auto response = std::move(encoder_decoder.second);
+  const int count = 10;
+  const int size = 100;
+  for (int i = 0; i < count; i++) {
+    codec_client_->sendData(*request_encoder_, size, false);
+  }
+  codec_client_->sendData(*request_encoder_, size, true);
+  waitForNextUpstreamRequest();
+
+  upstream_request_->encodeHeaders(default_response_headers_, true);
+  response->waitForEndStream();
+  ASSERT_TRUE(response->complete());
+  EXPECT_EQ((count + 1) * size, upstream_request_->bodyLength());
+  EXPECT_EQ(true, upstream_request_->complete());
+
+  // Sends a request with headers, data, and trailers.
+  auto encoder_decoder_2 = codec_client_->startRequest(default_request_headers_);
+  request_encoder_ = &encoder_decoder_2.first;
+  response = std::move(encoder_decoder_2.second);
+  for (int i = 0; i < count; i++) {
+    codec_client_->sendData(*request_encoder_, size, false);
+  }
+  Http::TestHeaderMapImpl request_trailers{{"trailer", "trailer"}};
+  codec_client_->sendTrailers(*request_encoder_, request_trailers);
+  waitForNextUpstreamRequest();
+
+  upstream_request_->encodeHeaders(default_response_headers_, true);
+  response->waitForEndStream();
+  EXPECT_EQ(count * size, upstream_request_->bodyLength());
+  EXPECT_EQ(true, upstream_request_->complete());
+}
+
 } // namespace Envoy

test/integration/http_integration.h

@@ -163,6 +163,7 @@ class HttpIntegrationTest : public BaseIntegrationTest {
   void testHeadersOnlyFilterEncodingIntermediateFilters();
   void testHeadersOnlyFilterDecodingIntermediateFilters();
   void testHeadersOnlyFilterInterleaved();
+  void testDecodeHeadersReturnsStopAll();
 
   // Test that a request returns the same content with both allow_absolute_urls enabled and
   // allow_absolute_urls disabled

test/integration/integration_test.cc

@@ -213,6 +213,8 @@ TEST_P(IntegrationTest, DecodingHeaderOnlyResponseIntermediateFilters) {
 
 TEST_P(IntegrationTest, DecodingHeaderOnlyInterleaved) { testHeadersOnlyFilterInterleaved(); }
 
+TEST_P(IntegrationTest, testDecodeHeadersReturnsStopAll) { testDecodeHeadersReturnsStopAll(); }
+
 TEST_P(IntegrationTest, RetryHittingBufferLimit) { testRetryHittingBufferLimit(); }
 
 TEST_P(IntegrationTest, HittingDecoderFilterLimit) { testHittingDecoderFilterLimit(); }