Flow control for Http::Filters

include/envoy/http/codec.h

@@ -151,6 +151,13 @@ class Stream {
    * @param disable informs if reads should be disabled (true) or re-enabled (false).
    */
   virtual void readDisable(bool disable) PURE;
+
+  /*
+   * Return the number of bytes this stream is allowed to buffer, or 0 if there is no limit
+   * configured.
+   * @return the stream's configured buffer limits.
+   */
+  virtual uint32_t bufferLimit() PURE;
 };
 
 /**

include/envoy/http/filter.h

@@ -229,6 +229,18 @@ class StreamFilterBase {
 public:
   virtual ~StreamFilterBase() {}
 
+  /**
+   * This routine is called on filter creation, setting the buffer limit for the
+   * filter.  Filters should abide by these limits unless custom configuration
+   * overrides the limit.  A buffer limit of 0 bytes indicates no limits are applied.
+   *
+   * If filters buffer enough bytes to hit the high watermark, it should either
+   * call on[Encoder|Decoder]FilterAboveWriteBufferHighWatermark to halt the
+   * flow of data or send an error response such as 413 (Payload Too Large).
+   * @param byte_limit supplies number of bytes this filter may buffer by default.
+   */
+  virtual void setBufferLimit(uint32_t byte_limit) PURE;
+
   /**
    * This routine is called prior to a filter being destroyed. This may happen after normal stream
    * finish (both downstream and upstream) or due to reset. Every filter is responsible for making

source/common/buffer/watermark_buffer.h

@@ -54,6 +54,7 @@ class WatermarkBuffer : public LibEventInstance {
   void postProcess() override { checkLowWatermark(); }
 
   void setWatermarks(uint32_t low_watermark, uint32_t high_watermark);
+  uint32_t high_watermark() const { return high_watermark_; }
 
 private:
   void checkHighWatermark();

source/common/dynamo/dynamo_filter.cc

@@ -32,6 +32,16 @@ Http::FilterDataStatus DynamoFilter::decodeData(Buffer::Instance& data, bool end
     onDecodeComplete(data);
   }
 
+  // If the request is too large, send a 413 to the user.   This could be applied before the check
+  // above, but as Envoy buffer limits are soft limits, if we already have the data buffered might
+  // as well pass it through.
+  if (buffer_limit_ > 0 && data.length() > buffer_limit_) {
+    scope_.counter(fmt::format("{}rq_too_large", stat_prefix_)).inc();
+    Http::Utility::sendLocalReply(*decoder_callbacks_, stream_reset_, Http::Code::PayloadTooLarge,
+                                  Http::CodeUtility::toString(Http::Code::PayloadTooLarge));
+    return Http::FilterDataStatus::StopIterationAndBuffer;
+  }
+
   if (end_stream) {
     return Http::FilterDataStatus::Continue;
   } else {

source/common/dynamo/dynamo_filter.h

@@ -27,7 +27,8 @@ class DynamoFilter : public Http::StreamFilter {
   }
 
   // Http::StreamFilterBase
-  void onDestroy() override {}
+  void setBufferLimit(uint32_t limit) override { buffer_limit_ = limit; }
+  void onDestroy() override { stream_reset_ = true; }
 
   // Http::StreamDecoderFilter
   Http::FilterHeadersStatus decodeHeaders(Http::HeaderMap& headers, bool end_stream) override;
@@ -60,14 +61,16 @@ class DynamoFilter : public Http::StreamFilter {
   std::string stat_prefix_;
   Stats::Scope& scope_;
 
-  bool enabled_{};
   std::string operation_{};
   RequestParser::TableDescriptor table_descriptor_{"", true};
   std::string error_type_{};
   MonotonicTime start_decode_;
   Http::HeaderMap* response_headers_;
   Http::StreamDecoderFilterCallbacks* decoder_callbacks_{};
   Http::StreamEncoderFilterCallbacks* encoder_callbacks_{};
+  uint32_t buffer_limit_{0};
+  bool enabled_{};
+  bool stream_reset_{false};
 };
 
 } // namespace Dynamo

source/common/grpc/grpc_web_filter.cc

@@ -2,6 +2,7 @@
 
 #include <arpa/inet.h>
 
+#include "common/common/assert.h"
 #include "common/common/base64.h"
 #include "common/common/utility.h"
 #include "common/grpc/common.h"
@@ -90,6 +91,8 @@ Http::FilterDataStatus GrpcWebFilter::decodeData(Buffer::Instance& data, bool) {
   decoding_buffer_.drain(decoding_buffer_.length());
   decoding_buffer_.move(data);
   data.add(decoded);
+  // Any block of 4 bytes or more should have been decoded and passed through.
+  ASSERT(decoding_buffer_.length() < 4);
   return Http::FilterDataStatus::Continue;
 }
 

source/common/grpc/grpc_web_filter.h

@@ -21,6 +21,8 @@ class GrpcWebFilter : public Http::StreamFilter, NonCopyable {
   virtual ~GrpcWebFilter(){};
 
   // Http::StreamFilterBase
+  // Ignore buffer limtis: see ASSERT in decodeData: decoding_buffer_ buffers less than 4 bytes.
+  void setBufferLimit(uint32_t) override {}
   void onDestroy() override{};
 
   // Implements StreamDecoderFilter.

source/common/grpc/http1_bridge_filter.cc

@@ -9,9 +9,11 @@
 #include "common/common/enum_to_int.h"
 #include "common/common/utility.h"
 #include "common/grpc/common.h"
+#include "common/http/codes.h"
 #include "common/http/filter_utility.h"
 #include "common/http/headers.h"
 #include "common/http/http1/codec_impl.h"
+#include "common/http/utility.h"
 
 namespace Envoy {
 namespace Grpc {
@@ -48,10 +50,17 @@ Http::FilterHeadersStatus Http1BridgeFilter::encodeHeaders(Http::HeaderMap& head
   }
 }
 
-Http::FilterDataStatus Http1BridgeFilter::encodeData(Buffer::Instance&, bool end_stream) {
+Http::FilterDataStatus Http1BridgeFilter::encodeData(Buffer::Instance& data, bool end_stream) {
   if (!do_bridging_ || end_stream) {
     return Http::FilterDataStatus::Continue;
   } else {
+    // If the response is too large to fit in the buffer, send an error to the user.
+    if (buffer_limit_ > 0 && data.length() > buffer_limit_) {
+      Http::Utility::sendLocalReply(*decoder_callbacks_, stream_reset_,
+                                    Http::Code::InternalServerError,
+                                    Http::CodeUtility::toString(Http::Code::InternalServerError));
+      return Http::FilterDataStatus::StopIterationAndBuffer;
+    }
     return Http::FilterDataStatus::StopIterationAndBuffer;
   }
 }

source/common/grpc/http1_bridge_filter.h

@@ -16,7 +16,8 @@ class Http1BridgeFilter : public Http::StreamFilter {
   Http1BridgeFilter(Upstream::ClusterManager& cm) : cm_(cm) {}
 
   // Http::StreamFilterBase
-  void onDestroy() override {}
+  void setBufferLimit(uint32_t limit) override { buffer_limit_ = limit; }
+  void onDestroy() override { stream_reset_ = true; }
 
   // Http::StreamDecoderFilter
   Http::FilterHeadersStatus decodeHeaders(Http::HeaderMap& headers, bool end_stream) override;
@@ -48,9 +49,11 @@ class Http1BridgeFilter : public Http::StreamFilter {
   Http::HeaderMap* response_headers_{};
   bool do_bridging_{};
   bool do_stat_tracking_{};
+  bool stream_reset_{false};
   Upstream::ClusterInfoConstSharedPtr cluster_;
   std::string grpc_service_;
   std::string grpc_method_;
+  uint32_t buffer_limit_{0};
 };
 
 } // namespace Grpc

source/common/grpc/json_transcoder_filter.h

@@ -96,6 +96,7 @@ class JsonTranscoderFilter : public Http::StreamFilter, public Logger::Loggable<
   void setEncoderFilterCallbacks(Http::StreamEncoderFilterCallbacks& callbacks) override;
 
   // Http::StreamFilterBase
+  void setBufferLimit(uint32_t) override {} // Unimplemented: see TODO in .cc file.
   void onDestroy() override { stream_reset_ = true; }
 
 private:

source/common/http/filter/BUILD

@@ -22,8 +22,10 @@ envoy_cc_library(
         "//source/common/buffer:buffer_lib",
         "//source/common/common:assert_lib",
         "//source/common/common:enum_to_int",
+        "//source/common/http:codes_lib",
         "//source/common/http:header_map_lib",
         "//source/common/http:headers_lib",
+        "//source/common/http:utility_lib",
     ],
 )
 

source/common/http/filter/buffer_filter.cc

@@ -9,8 +9,10 @@
 
 #include "common/common/assert.h"
 #include "common/common/enum_to_int.h"
+#include "common/http/codes.h"
 #include "common/http/header_map_impl.h"
 #include "common/http/headers.h"
+#include "common/http/utility.h"
 
 namespace Envoy {
 namespace Http {
@@ -38,10 +40,8 @@ FilterDataStatus BufferFilter::decodeData(Buffer::Instance&, bool end_stream) {
     return FilterDataStatus::Continue;
   } else if (callbacks_->decodingBuffer() &&
              callbacks_->decodingBuffer()->length() > config_->max_request_bytes_) {
-    // TODO(htuch): Switch this to Utility::sendLocalReply().
-    Http::HeaderMapPtr response_headers{new HeaderMapImpl{
-        {Headers::get().Status, std::to_string(enumToInt(Http::Code::PayloadTooLarge))}}};
-    callbacks_->encodeHeaders(std::move(response_headers), true);
+    Http::Utility::sendLocalReply(*callbacks_, stream_reset_, Http::Code::PayloadTooLarge,
+                                  CodeUtility::toString(Http::Code::PayloadTooLarge));
     config_->stats_.rq_too_large_.inc();
   }
 
@@ -58,7 +58,10 @@ BufferFilterStats BufferFilter::generateStats(const std::string& prefix, Stats::
   return {ALL_BUFFER_FILTER_STATS(POOL_COUNTER_PREFIX(scope, final_prefix))};
 }
 
-void BufferFilter::onDestroy() { resetInternalState(); }
+void BufferFilter::onDestroy() {
+  stream_reset_ = true;
+  resetInternalState();
+}
 
 void BufferFilter::onRequestTimeout() {
   // TODO(htuch): Switch this to Utility::sendLocalReply().

source/common/http/filter/buffer_filter.h

@@ -51,6 +51,7 @@ class BufferFilter : public StreamDecoderFilter {
   static BufferFilterStats generateStats(const std::string& prefix, Stats::Scope& scope);
 
   // Http::StreamFilterBase
+  void setBufferLimit(uint32_t) override {} // Buffer filter uses its own configured limits.
   void onDestroy() override;
 
   // Http::StreamDecoderFilter
@@ -66,6 +67,7 @@ class BufferFilter : public StreamDecoderFilter {
   BufferFilterConfigConstSharedPtr config_;
   StreamDecoderFilterCallbacks* callbacks_{};
   Event::TimerPtr request_timeout_;
+  bool stream_reset_{false};
 };
 
 } // Http

source/common/http/filter/fault_filter.cc

@@ -73,7 +73,10 @@ FaultFilterConfig::FaultFilterConfig(const Json::Object& json_config, Runtime::L
 
 FaultFilter::FaultFilter(FaultFilterConfigSharedPtr config) : config_(config) {}
 
-FaultFilter::~FaultFilter() { ASSERT(!delay_timer_); }
+FaultFilter::~FaultFilter() {
+  ASSERT(!delay_timer_);
+  ASSERT(!high_watermark_called_);
+}
 
 // Delays and aborts are independent events. One can inject a delay
 // followed by an abort or inject just a delay or abort. In this callback,
@@ -209,8 +212,16 @@ void FaultFilter::recordAbortsInjectedStats() {
 }
 
 FilterDataStatus FaultFilter::decodeData(Buffer::Instance&, bool) {
-  return delay_timer_ == nullptr ? FilterDataStatus::Continue
-                                 : FilterDataStatus::StopIterationAndBuffer;
+  if (delay_timer_ == nullptr) {
+    return FilterDataStatus::Continue;
+  }
+  // The fault filter mimizes buffering even more aggressively than configured, to avoid
+  // accumulating data during a delay.
+  if (limiting_buffers_ && !high_watermark_called_) {
+    high_watermark_called_ = true;
+    callbacks_->onDecoderFilterAboveWriteBufferHighWatermark();
+  }
+  return FilterDataStatus::StopIterationAndBuffer;
 }
 
 FilterTrailersStatus FaultFilter::decodeTrailers(HeaderMap&) {
@@ -276,6 +287,13 @@ void FaultFilter::resetTimerState() {
     delay_timer_->disableTimer();
     delay_timer_.reset();
   }
+  // It's not necessarily true that the buffer has drained below the low watermark, but this filter
+  // is no longer causing data to be buffered and this is the single safest place for watermark
+  // resumption.
+  if (high_watermark_called_) {
+    high_watermark_called_ = false;
+    callbacks_->onDecoderFilterBelowWriteBufferLowWatermark();
+  }
 }
 
 void FaultFilter::setDecoderFilterCallbacks(StreamDecoderFilterCallbacks& callbacks) {

source/common/http/filter/fault_filter.h

@@ -81,6 +81,9 @@ class FaultFilter : public StreamDecoderFilter {
   ~FaultFilter();
 
   // Http::StreamFilterBase
+  // Fault filter is more aggressive and limits buffered data even before
+  // hitting the buffer limit.
+  void setBufferLimit(uint32_t limit) override { limiting_buffers_ = limit > 0; }
   void onDestroy() override;
 
   // Http::StreamDecoderFilter
@@ -107,6 +110,8 @@ class FaultFilter : public StreamDecoderFilter {
   StreamDecoderFilterCallbacks* callbacks_{};
   Event::TimerPtr delay_timer_;
   std::string downstream_cluster_{};
+  bool limiting_buffers_{false};
+  bool high_watermark_called_{false};
 
   std::string downstream_cluster_delay_percent_key_{};
   std::string downstream_cluster_abort_percent_key_{};

source/common/http/filter/ip_tagging_filter.h

@@ -58,6 +58,7 @@ class IpTaggingFilter : public StreamDecoderFilter {
   ~IpTaggingFilter();
 
   // Http::StreamFilterBase
+  void setBufferLimit(uint32_t) override {} // Non-buffering filter.
   void onDestroy() override;
 
   // Http::StreamDecoderFilter

source/common/http/filter/ratelimit.cc

@@ -84,8 +84,16 @@ FilterHeadersStatus Filter::decodeHeaders(HeaderMap& headers, bool) {
 
 FilterDataStatus Filter::decodeData(Buffer::Instance&, bool) {
   ASSERT(state_ != State::Responded);
-  return state_ == State::Calling ? FilterDataStatus::StopIterationAndBuffer
-                                  : FilterDataStatus::Continue;
+  if (state_ != State::Calling) {
+    return FilterDataStatus::Continue;
+  }
+  // The fault filter mimizes buffering even more aggressively than configured, to avoid
+  // accumulating data for rate limited requests.
+  if (limiting_buffers_ && !high_watermark_called_) {
+    high_watermark_called_ = true;
+    callbacks_->onDecoderFilterAboveWriteBufferHighWatermark();
+  }
+  return FilterDataStatus::StopIterationAndBuffer;
 }
 
 FilterTrailersStatus Filter::decodeTrailers(HeaderMap&) {
@@ -99,6 +107,10 @@ void Filter::setDecoderFilterCallbacks(StreamDecoderFilterCallbacks& callbacks)
 }
 
 void Filter::onDestroy() {
+  if (high_watermark_called_) {
+    high_watermark_called_ = false;
+    callbacks_->onDecoderFilterBelowWriteBufferLowWatermark();
+  }
   if (state_ == State::Calling) {
     state_ = State::Complete;
     client_->cancel();
@@ -108,6 +120,10 @@ void Filter::onDestroy() {
 void Filter::complete(Envoy::RateLimit::LimitStatus status) {
   state_ = State::Complete;
 
+  if (high_watermark_called_) {
+    high_watermark_called_ = false;
+    callbacks_->onDecoderFilterBelowWriteBufferLowWatermark();
+  }
   switch (status) {
   case Envoy::RateLimit::LimitStatus::OK:
     cluster_->statsScope().counter("ratelimit.ok").inc();

source/common/http/filter/ratelimit.h

@@ -80,6 +80,7 @@ class Filter : public StreamDecoderFilter, public Envoy::RateLimit::RequestCallb
       : config_(config), client_(std::move(client)) {}
 
   // Http::StreamFilterBase
+  void setBufferLimit(uint32_t limit) override { limiting_buffers_ = limit > 0; }
   void onDestroy() override;
 
   // Http::StreamDecoderFilter
@@ -103,9 +104,11 @@ class Filter : public StreamDecoderFilter, public Envoy::RateLimit::RequestCallb
   FilterConfigSharedPtr config_;
   Envoy::RateLimit::ClientPtr client_;
   StreamDecoderFilterCallbacks* callbacks_{};
-  bool initiating_call_{};
   State state_{State::NotStarted};
   Upstream::ClusterInfoConstSharedPtr cluster_;
+  bool initiating_call_{};
+  bool limiting_buffers_{false};
+  bool high_watermark_called_{false};
 };
 
 } // namespace RateLimit

source/common/http/http1/codec_impl.h

@@ -42,7 +42,9 @@ class StreamEncoderImpl : public StreamEncoder,
   void addCallbacks(StreamCallbacks& callbacks) override { addCallbacks_(callbacks); }
   void removeCallbacks(StreamCallbacks& callbacks) override { removeCallbacks_(callbacks); }
   void resetStream(StreamResetReason reason) override;
+  // TODO(alyssawilk) HTTP/1.1 flow control.
   void readDisable(bool) override {}
+  uint32_t bufferLimit() override { return 0; }
 
 protected:
   StreamEncoderImpl(ConnectionImpl& connection) : connection_(connection) {}

source/common/http/http2/codec_impl.h

@@ -147,6 +147,7 @@ class ConnectionImpl : public virtual Connection, Logger::Loggable<Logger::Id::h
     void removeCallbacks(StreamCallbacks& callbacks) override { removeCallbacks_(callbacks); }
     void resetStream(StreamResetReason reason) override;
     virtual void readDisable(bool disable) override;
+    virtual uint32_t bufferLimit() override { return pending_recv_data_.high_watermark(); }
 
     void setWriteBufferWatermarks(uint32_t low_watermark, uint32_t high_watermark) {
       pending_recv_data_.setWatermarks(low_watermark, high_watermark);

source/common/router/BUILD

@@ -107,6 +107,7 @@ envoy_cc_library(
         "//include/envoy/stats:stats_macros",
         "//include/envoy/upstream:cluster_manager_interface",
         "//include/envoy/upstream:upstream_interface",
+        "//source/common/buffer:watermark_buffer_lib",
         "//source/common/common:assert_lib",
         "//source/common/common:empty_string",
         "//source/common/common:enum_to_int",