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

@@ -222,6 +222,14 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
   virtual void addDownstreamWatermarkCallbacks(DownstreamWatermarkCallbacks& callbacks) PURE;
 };
 
+// FIXME alyssar doxygen.
+enum class FilterType { STREAMING, BUFFERING };
+
+struct BufferLimitSettings {
+  uint32_t buffer_limit_;
+  FilterType filter_type_;
+};
+
 /**
  * Common base class for both decoder and encoder filters.
  */
@@ -246,6 +254,22 @@ class StreamFilterBase {
  */
 class StreamDecoderFilter : public StreamFilterBase {
 public:
+  /**
+   * 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).
+   *
+   * If the filter will return StopIterationAndBuffer it may override the buffer
+   * limit in settings, or the filter type.
+   *
+   * @param settings supplies the default buffer limit and filter type for this filter.
+   */
+  virtual void setDecoderBufferLimit(BufferLimitSettings& settings) PURE;
+
   /**
    * Called with decoded headers, optionally indicating end of stream.
    * @param headers supplies the decoded headers map.
@@ -363,6 +387,15 @@ class StreamEncoderFilter : public StreamFilterBase {
    * use. Callbacks will not be invoked by the filter after onDestroy() is called.
    */
   virtual void setEncoderFilterCallbacks(StreamEncoderFilterCallbacks& callbacks) PURE;
+
+  /**
+   * 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.
+   *
+   * FIXME(alyssar) comment.
+   */
+  virtual void setEncoderBufferLimit(BufferLimitSettings& settings) PURE;
 };
 
 typedef std::shared_ptr<StreamEncoderFilter> StreamEncoderFilterSharedPtr;

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 highWatermark() const { return high_watermark_; }
 
 private:
   void checkHighWatermark();
@@ -73,5 +74,7 @@ class WatermarkBuffer : public LibEventInstance {
   bool above_high_watermark_called_{false};
 };
 
+typedef std::unique_ptr<WatermarkBuffer> WatermarkBufferPtr;
+
 } // namespace Buffer
 } // namespace Envoy

source/common/dynamo/dynamo_filter.h

@@ -36,6 +36,9 @@ class DynamoFilter : public Http::StreamFilter {
   void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override {
     decoder_callbacks_ = &callbacks;
   }
+  void setDecoderBufferLimit(Http::BufferLimitSettings& settings) override {
+    settings.filter_type_ = Http::FilterType::BUFFERING;
+  }
 
   // Http::StreamEncoderFilter
   Http::FilterHeadersStatus encodeHeaders(Http::HeaderMap&, bool) override;
@@ -44,6 +47,9 @@ class DynamoFilter : public Http::StreamFilter {
   void setEncoderFilterCallbacks(Http::StreamEncoderFilterCallbacks& callbacks) override {
     encoder_callbacks_ = &callbacks;
   }
+  void setEncoderBufferLimit(Http::BufferLimitSettings& settings) override {
+    settings.filter_type_ = Http::FilterType::BUFFERING;
+  }
 
 private:
   void onDecodeComplete(const Buffer::Instance& data);

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,7 +21,7 @@ class GrpcWebFilter : public Http::StreamFilter, NonCopyable {
   virtual ~GrpcWebFilter(){};
 
   // Http::StreamFilterBase
-  void onDestroy() override{};
+  void onDestroy() override {}
 
   // Implements StreamDecoderFilter.
   Http::FilterHeadersStatus decodeHeaders(Http::HeaderMap&, bool) override;
@@ -32,6 +32,10 @@ class GrpcWebFilter : public Http::StreamFilter, NonCopyable {
   void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override {
     decoder_callbacks_ = &callbacks;
   }
+  void setDecoderBufferLimit(Http::BufferLimitSettings& settings) override {
+    // Ignore buffer limits: see ASSERT in decodeData: decoding_buffer_ buffers less than 4 bytes.
+    settings.filter_type_ = Http::FilterType::STREAMING;
+  }
 
   // Implements StreamEncoderFilter.
   Http::FilterHeadersStatus encodeHeaders(Http::HeaderMap&, bool) override;
@@ -40,6 +44,9 @@ class GrpcWebFilter : public Http::StreamFilter, NonCopyable {
   void setEncoderFilterCallbacks(Http::StreamEncoderFilterCallbacks& callbacks) override {
     encoder_callbacks_ = &callbacks;
   }
+  void setEncoderBufferLimit(Http::BufferLimitSettings& settings) override {
+    settings.filter_type_ = Http::FilterType::STREAMING;
+  }
 
 private:
   friend class GrpcWebFilterTest;

source/common/grpc/http1_bridge_filter.h

@@ -29,6 +29,9 @@ class Http1BridgeFilter : public Http::StreamFilter {
   void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override {
     decoder_callbacks_ = &callbacks;
   }
+  void setDecoderBufferLimit(Http::BufferLimitSettings& settings) override {
+    settings.filter_type_ = Http::FilterType::BUFFERING;
+  }
 
   // Http::StreamEncoderFilter
   Http::FilterHeadersStatus encodeHeaders(Http::HeaderMap& headers, bool end_stream) override;
@@ -37,6 +40,9 @@ class Http1BridgeFilter : public Http::StreamFilter {
   void setEncoderFilterCallbacks(Http::StreamEncoderFilterCallbacks& callbacks) override {
     encoder_callbacks_ = &callbacks;
   }
+  void setEncoderBufferLimit(Http::BufferLimitSettings& settings) override {
+    settings.filter_type_ = Http::FilterType::BUFFERING;
+  }
 
 private:
   void chargeStat(const Http::HeaderMap& headers);

source/common/grpc/json_transcoder_filter.h

@@ -88,12 +88,18 @@ class JsonTranscoderFilter : public Http::StreamFilter, public Logger::Loggable<
   Http::FilterDataStatus decodeData(Buffer::Instance& data, bool end_stream) override;
   Http::FilterTrailersStatus decodeTrailers(Http::HeaderMap& trailers) override;
   void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override;
+  void setDecoderBufferLimit(Http::BufferLimitSettings& settings) override {
+    settings.filter_type_ = Http::FilterType::STREAMING;
+  }
 
   // Http::StreamEncoderFilter
   Http::FilterHeadersStatus encodeHeaders(Http::HeaderMap& headers, bool end_stream) override;
   Http::FilterDataStatus encodeData(Buffer::Instance& data, bool end_stream) override;
   Http::FilterTrailersStatus encodeTrailers(Http::HeaderMap& trailers) override;
   void setEncoderFilterCallbacks(Http::StreamEncoderFilterCallbacks& callbacks) override;
+  void setEncoderBufferLimit(Http::BufferLimitSettings& settings) override {
+    settings.filter_type_ = Http::FilterType::BUFFERING;
+  }
 
   // Http::StreamFilterBase
   void onDestroy() override { stream_reset_ = true; }

source/common/http/conn_manager_impl.cc

@@ -149,6 +149,7 @@ void ConnectionManagerImpl::doEndStream(ActiveStream& stream) {
 }
 
 void ConnectionManagerImpl::doDeferredStreamDestroy(ActiveStream& stream) {
+  stream.destroyed_ = true;
   for (auto& filter : stream.decoder_filters_) {
     filter->handle_->onDestroy();
   }
@@ -172,6 +173,7 @@ StreamDecoder& ConnectionManagerImpl::newStream(StreamEncoder& response_encoder)
   ActiveStreamPtr new_stream(new ActiveStream(*this));
   new_stream->response_encoder_ = &response_encoder;
   new_stream->response_encoder_->getStream().addCallbacks(*new_stream);
+  new_stream->buffer_limit_ = new_stream->response_encoder_->getStream().bufferLimit();
   config_.filterFactory().createFilterChain(*new_stream);
   new_stream->moveIntoList(std::move(new_stream), streams_);
   return **streams_.begin();
@@ -367,13 +369,31 @@ void ConnectionManagerImpl::ActiveStream::addStreamDecoderFilterWorker(
     StreamDecoderFilterSharedPtr filter, bool dual_filter) {
   ActiveStreamDecoderFilterPtr wrapper(new ActiveStreamDecoderFilter(*this, filter, dual_filter));
   filter->setDecoderFilterCallbacks(*wrapper);
+
+  BufferLimitSettings settings{buffer_limit_, Http::FilterType::STREAMING};
+  filter->setDecoderBufferLimit(settings);
+  // FIXME corner cases with 0 here and below.
+  if (settings.buffer_limit_ > buffer_limit_) {
+    buffer_limit_ = settings.buffer_limit_;
+  }
+  if (settings.filter_type_ == Http::FilterType::BUFFERING) {
+    decoder_filters_all_streaming_ = false;
+  }
   wrapper->moveIntoListBack(std::move(wrapper), decoder_filters_);
 }
 
 void ConnectionManagerImpl::ActiveStream::addStreamEncoderFilterWorker(
     StreamEncoderFilterSharedPtr filter, bool dual_filter) {
   ActiveStreamEncoderFilterPtr wrapper(new ActiveStreamEncoderFilter(*this, filter, dual_filter));
   filter->setEncoderFilterCallbacks(*wrapper);
+  BufferLimitSettings settings{buffer_limit_, Http::FilterType::STREAMING};
+  filter->setEncoderBufferLimit(settings);
+  if (settings.buffer_limit_ > buffer_limit_) {
+    buffer_limit_ = settings.buffer_limit_;
+  }
+  if (settings.filter_type_ == Http::FilterType::BUFFERING) {
+    encoder_filters_all_streaming_ = false;
+  }
   wrapper->moveIntoListBack(std::move(wrapper), encoder_filters_);
 }
 
@@ -966,7 +986,7 @@ void ConnectionManagerImpl::ActiveStreamFilterBase::commonHandleBufferData(
   // rebuffer, because we assume the filter has modified the buffer as it wishes in place.
   if (bufferedData().get() != &provided_data) {
     if (!bufferedData()) {
-      bufferedData().reset(new Buffer::OwnedImpl());
+      bufferedData().reset(createBuffer().release());
     }
     bufferedData()->move(provided_data);
   }
@@ -1067,6 +1087,30 @@ void ConnectionManagerImpl::ActiveStreamDecoderFilter::
   parent_.connection_manager_.stats_.named_.downstream_flow_control_paused_reading_total_.inc();
 }
 
+void ConnectionManagerImpl::ActiveStreamDecoderFilter::requestDataTooLarge() {
+  if (parent_.encoder_filters_all_streaming_) {
+    onDecoderFilterAboveWriteBufferHighWatermark();
+  } else {
+    HeaderMapPtr response_headers{new HeaderMapImpl{
+        {Headers::get().Status, std::to_string(enumToInt(Http::Code::PayloadTooLarge))}}};
+    std::string body_text = CodeUtility::toString(Http::Code::PayloadTooLarge);
+    response_headers->insertContentLength().value(body_text.size());
+    response_headers->insertContentType().value(Headers::get().ContentTypeValues.Text);
+
+    encodeHeaders(std::move(response_headers), false);
+    if (!parent_.destroyed_) {
+      Buffer::OwnedImpl buffer(body_text);
+      encodeData(buffer, true);
+    }
+  }
+}
+
+void ConnectionManagerImpl::ActiveStreamDecoderFilter::requestDataDrained() {
+  if (parent_.encoder_filters_all_streaming_) {
+    onDecoderFilterBelowWriteBufferLowWatermark();
+  }
+}
+
 void ConnectionManagerImpl::ActiveStreamDecoderFilter::
     onDecoderFilterBelowWriteBufferLowWatermark() {
   ENVOY_STREAM_LOG(debug, "Read-enabling downstream stream due to filter callbacks.", parent_);
@@ -1092,6 +1136,31 @@ void ConnectionManagerImpl::ActiveStreamEncoderFilter::
 
 void ConnectionManagerImpl::ActiveStreamEncoderFilter::continueEncoding() { commonContinue(); }
 
+void ConnectionManagerImpl::ActiveStreamEncoderFilter::responseDataTooLarge() {
+  if (parent_.encoder_filters_all_streaming_) {
+    onEncoderFilterAboveWriteBufferHighWatermark();
+  } else {
+    // FIXME 500
+    HeaderMapPtr response_headers{new HeaderMapImpl{
+        {Headers::get().Status, std::to_string(enumToInt(Http::Code::InternalServerError))}}};
+    std::string body_text = CodeUtility::toString(Http::Code::InternalServerError);
+    response_headers->insertContentLength().value(body_text.size());
+    response_headers->insertContentType().value(Headers::get().ContentTypeValues.Text);
+
+    parent_.response_headers_ = std::move(response_headers);
+    parent_.encodeHeaders(nullptr, *parent_.response_headers_, false);
+    if (!parent_.destroyed_) {
+      Buffer::OwnedImpl buffer(body_text);
+      parent_.encodeData(nullptr, buffer, true);
+    }
+  }
+}
+void ConnectionManagerImpl::ActiveStreamEncoderFilter::responseDataDrained() {
+  if (parent_.encoder_filters_all_streaming_) {
+    onEncoderFilterBelowWriteBufferLowWatermark();
+  }
+}
+
 void ConnectionManagerImpl::ActiveStreamFilterBase::resetStream() {
   parent_.connection_manager_.stats_.named_.downstream_rq_tx_reset_.inc();
   parent_.connection_manager_.doEndStream(this->parent_);