Overflow watermark

include/envoy/buffer/buffer.h

@@ -371,10 +371,13 @@ class WatermarkFactory {
    *   low watermark.
    * @param above_high_watermark supplies a function to call if the buffer goes over a configured
    *   high watermark.
+   * @param above_overflow_watermark supplies a function to call if the buffer goes over a
+   *   configured "overflow" watermark.
    * @return a newly created InstancePtr.
    */
   virtual InstancePtr create(std::function<void()> below_low_watermark,
-                             std::function<void()> above_high_watermark) PURE;
+                             std::function<void()> above_high_watermark,
+                             std::function<void()> above_overflow_watermark) PURE;
 };
 
 using WatermarkFactoryPtr = std::unique_ptr<WatermarkFactory>;

include/envoy/http/filter.h

@@ -377,7 +377,7 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
   virtual void encodeMetadata(MetadataMapPtr&& metadata_map) PURE;
 
   /**
-   * Called when the buffer for a decoder filter or any buffers the filter sends data to go over
+   * Called when the buffer for a decoder filter, or any buffers the filter sends data to, go over
    * their high watermark.
    *
    * In the case of a filter such as the router filter, which spills into multiple buffers (codec,

source/common/buffer/BUILD

@@ -13,6 +13,7 @@ envoy_cc_library(
     srcs = ["watermark_buffer.cc"],
     hdrs = ["watermark_buffer.h"],
     deps = [
+        "//include/envoy/runtime:runtime_interface",
         "//source/common/buffer:buffer_lib",
         "//source/common/common:assert_lib",
     ],

source/common/buffer/watermark_buffer.cc

@@ -7,32 +7,32 @@ namespace Buffer {
 
 void WatermarkBuffer::add(const void* data, uint64_t size) {
   OwnedImpl::add(data, size);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::add(absl::string_view data) {
   OwnedImpl::add(data);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::add(const Instance& data) {
   OwnedImpl::add(data);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::prepend(absl::string_view data) {
   OwnedImpl::prepend(data);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::prepend(Instance& data) {
   OwnedImpl::prepend(data);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::commit(RawSlice* iovecs, uint64_t num_iovecs) {
   OwnedImpl::commit(iovecs, num_iovecs);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::drain(uint64_t size) {
@@ -42,23 +42,23 @@ void WatermarkBuffer::drain(uint64_t size) {
 
 void WatermarkBuffer::move(Instance& rhs) {
   OwnedImpl::move(rhs);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 void WatermarkBuffer::move(Instance& rhs, uint64_t length) {
   OwnedImpl::move(rhs, length);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
 }
 
 Api::IoCallUint64Result WatermarkBuffer::read(Network::IoHandle& io_handle, uint64_t max_length) {
   Api::IoCallUint64Result result = OwnedImpl::read(io_handle, max_length);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
   return result;
 }
 
 uint64_t WatermarkBuffer::reserve(uint64_t length, RawSlice* iovecs, uint64_t num_iovecs) {
   uint64_t bytes_reserved = OwnedImpl::reserve(length, iovecs, num_iovecs);
-  checkHighWatermark();
+  checkHighAndOverflowWatermarks();
   return bytes_reserved;
 }
 
@@ -72,12 +72,16 @@ void WatermarkBuffer::setWatermarks(uint32_t low_watermark, uint32_t high_waterm
   ASSERT(low_watermark < high_watermark || (high_watermark == 0 && low_watermark == 0));
   low_watermark_ = low_watermark;
   high_watermark_ = high_watermark;
-  checkHighWatermark();
+  overflow_watermark_ = high_watermark * overflow_multiplier_;
+  // TODO(adip): What should be done if there's an overflow (overflow_watermark_ <
+  // overflow_watermark_)? should this be a release_assert?
+  ASSERT((high_watermark <= overflow_watermark_) || (overflow_multiplier_ == 0));
+  checkHighAndOverflowWatermarks();
   checkLowWatermark();
 }
 
 void WatermarkBuffer::checkLowWatermark() {
-  if (!above_high_watermark_called_ ||
+  if (above_overflow_watermark_called_ || !above_high_watermark_called_ ||
       (high_watermark_ != 0 && OwnedImpl::length() > low_watermark_)) {
     return;
   }
@@ -86,14 +90,19 @@ void WatermarkBuffer::checkLowWatermark() {
   below_low_watermark_();
 }
 
-void WatermarkBuffer::checkHighWatermark() {
-  if (above_high_watermark_called_ || high_watermark_ == 0 ||
-      OwnedImpl::length() <= high_watermark_) {
+void WatermarkBuffer::checkHighAndOverflowWatermarks() {
+  if (!above_overflow_watermark_called_ && overflow_watermark_ != 0 &&
+      OwnedImpl::length() > overflow_watermark_) {
+    above_overflow_watermark_called_ = true;
+    above_overflow_watermark_();
     return;
   }
 
-  above_high_watermark_called_ = true;
-  above_high_watermark_();
+  if (!above_high_watermark_called_ && high_watermark_ != 0 &&
+      OwnedImpl::length() > high_watermark_) {
+    above_high_watermark_called_ = true;
+    above_high_watermark_();
+  }
 }
 
 } // namespace Buffer

source/common/buffer/watermark_buffer.h

@@ -3,6 +3,8 @@
 #include <functional>
 #include <string>
 
+#include "envoy/runtime/runtime.h"
+
 #include "common/buffer/buffer_impl.h"
 
 namespace Envoy {
@@ -16,8 +18,12 @@ namespace Buffer {
 class WatermarkBuffer : public OwnedImpl {
 public:
   WatermarkBuffer(std::function<void()> below_low_watermark,
-                  std::function<void()> above_high_watermark)
-      : below_low_watermark_(below_low_watermark), above_high_watermark_(above_high_watermark) {}
+                  std::function<void()> above_high_watermark,
+                  std::function<void()> above_overflow_watermark)
+      : below_low_watermark_(below_low_watermark), above_high_watermark_(above_high_watermark),
+        above_overflow_watermark_(above_overflow_watermark),
+        overflow_multiplier_(Runtime::LoaderSingleton::get().threadsafeSnapshot()->getInteger(
+            "buffer.overflow.high_watermark_multiplier", 2)) {}
 
   // Override all functions from Instance which can result in changing the size
   // of the underlying buffer.
@@ -40,20 +46,25 @@ class WatermarkBuffer : public OwnedImpl {
   uint32_t highWatermark() const { return high_watermark_; }
 
 private:
-  void checkHighWatermark();
+  void checkHighAndOverflowWatermarks();
   void checkLowWatermark();
 
   std::function<void()> below_low_watermark_;
   std::function<void()> above_high_watermark_;
+  std::function<void()> above_overflow_watermark_;
 
+  const uint32_t overflow_multiplier_{0};
   // Used for enforcing buffer limits (off by default). If these are set to non-zero by a call to
   // setWatermarks() the watermark callbacks will be called as described above.
+  uint32_t overflow_watermark_{0};
   uint32_t high_watermark_{0};
   uint32_t low_watermark_{0};
-  // Tracks the latest state of watermark callbacks.
-  // True between the time above_high_watermark_ has been called until above_high_watermark_ has
-  // been called.
+  // Set to true after above_high_watermark_ has been called, and reset to false after
+  // below_low_watermark_ has been called.
   bool above_high_watermark_called_{false};
+  // Set to true after above_overflow_watermark_ has been called. Never reset, because we expect
+  // the associated connection and/or stream will be closed immediately in response to an overflow.
+  bool above_overflow_watermark_called_{false};
 };
 
 using WatermarkBufferPtr = std::unique_ptr<WatermarkBuffer>;
@@ -62,8 +73,10 @@ class WatermarkBufferFactory : public WatermarkFactory {
 public:
   // Buffer::WatermarkFactory
   InstancePtr create(std::function<void()> below_low_watermark,
-                     std::function<void()> above_high_watermark) override {
-    return InstancePtr{new WatermarkBuffer(below_low_watermark, above_high_watermark)};
+                     std::function<void()> above_high_watermark,
+                     std::function<void()> above_overflow_watermark) override {
+    return InstancePtr{
+        new WatermarkBuffer(below_low_watermark, above_high_watermark, above_overflow_watermark)};
   }
 };
 

source/common/http/conn_manager_impl.cc

@@ -2219,9 +2219,9 @@ void ConnectionManagerImpl::ActiveStreamFilterBase::clearRouteCache() {
 }
 
 Buffer::WatermarkBufferPtr ConnectionManagerImpl::ActiveStreamDecoderFilter::createBuffer() {
-  auto buffer =
-      std::make_unique<Buffer::WatermarkBuffer>([this]() -> void { this->requestDataDrained(); },
-                                                [this]() -> void { this->requestDataTooLarge(); });
+  auto buffer = std::make_unique<Buffer::WatermarkBuffer>(
+      [this]() -> void { this->requestDataDrained(); },
+      [this]() -> void { this->requestDataTooLarge(); }, [this]() -> void { this->resetStream(); });
   buffer->setWatermarks(parent_.buffer_limit_);
   return buffer;
 }
@@ -2388,8 +2388,10 @@ ConnectionManagerImpl::ActiveStreamDecoderFilter::routeConfig() {
 }
 
 Buffer::WatermarkBufferPtr ConnectionManagerImpl::ActiveStreamEncoderFilter::createBuffer() {
-  auto buffer = new Buffer::WatermarkBuffer([this]() -> void { this->responseDataDrained(); },
-                                            [this]() -> void { this->responseDataTooLarge(); });
+  auto buffer = new Buffer::WatermarkBuffer(
+      [this]() -> void { this->responseDataDrained(); },
+      [this]() -> void { this->responseDataExceedsHighWatermark(); },
+      [this]() -> void { this->responseDataExceedsOverflowWatermark(); });
   buffer->setWatermarks(parent_.buffer_limit_);
   return Buffer::WatermarkBufferPtr{buffer};
 }
@@ -2441,43 +2443,49 @@ void ConnectionManagerImpl::ActiveStreamEncoderFilter::
 
 void ConnectionManagerImpl::ActiveStreamEncoderFilter::continueEncoding() { commonContinue(); }
 
-void ConnectionManagerImpl::ActiveStreamEncoderFilter::responseDataTooLarge() {
+void ConnectionManagerImpl::ActiveStreamEncoderFilter::responseDataExceedsHighWatermark() {
   if (parent_.state_.encoder_filters_streaming_) {
+    // If we are streaming the response, there's still an opportunity to drain the response buffer.
     onEncoderFilterAboveWriteBufferHighWatermark();
   } else {
-    parent_.connection_manager_.stats_.named_.rs_too_large_.inc();
-
-    // If headers have not been sent to the user, send a 500.
-    if (!headers_continued_) {
-      // Make sure we won't end up with nested watermark calls from the body buffer.
-      parent_.state_.encoder_filters_streaming_ = true;
-      allowIteration();
-
-      parent_.stream_info_.setResponseCodeDetails(
-          StreamInfo::ResponseCodeDetails::get().RequestHeadersTooLarge);
-      // This does not call the standard sendLocalReply because if there is already response data
-      // we do not want to pass a second set of response headers through the filter chain.
-      // Instead, call the encodeHeadersInternal / encodeDataInternal helpers
-      // directly, which maximizes shared code with the normal response path.
-      Http::Utility::sendLocalReply(
-          Grpc::Common::hasGrpcContentType(*parent_.request_headers_),
-          [&](ResponseHeaderMapPtr&& response_headers, bool end_stream) -> void {
-            parent_.response_headers_ = std::move(response_headers);
-            parent_.encodeHeadersInternal(*parent_.response_headers_, end_stream);
-          },
-          [&](Buffer::Instance& data, bool end_stream) -> void {
-            parent_.encodeDataInternal(data, end_stream);
-          },
-          parent_.state_.destroyed_, Http::Code::InternalServerError,
-          CodeUtility::toString(Http::Code::InternalServerError), absl::nullopt,
-          parent_.state_.is_head_request_);
-      parent_.maybeEndEncode(parent_.state_.local_complete_);
-    } else {
-      ENVOY_STREAM_LOG(
-          debug, "Resetting stream. Response data too large and headers have already been sent",
-          *this);
-      resetStream();
-    }
+    // Otherwise, treat it as an overflow: send a 500 if possible and reset the stream.
+    responseDataExceedsOverflowWatermark();
+  }
+}
+
+void ConnectionManagerImpl::ActiveStreamEncoderFilter::responseDataExceedsOverflowWatermark() {
+  parent_.connection_manager_.stats_.named_.rs_too_large_.inc();
+
+  // If headers have not been sent to the user, send a 500.
+  if (!headers_continued_) {
+    // Make sure we won't end up with nested watermark calls from the body buffer.
+    parent_.state_.encoder_filters_streaming_ = true;
+    allowIteration();
+
+    parent_.stream_info_.setResponseCodeDetails(
+        StreamInfo::ResponseCodeDetails::get().RequestHeadersTooLarge);
+    // This does not call the standard sendLocalReply because if there is already response data
+    // we do not want to pass a second set of response headers through the filter chain.
+    // Instead, call the encodeHeadersInternal / encodeDataInternal helpers
+    // directly, which maximizes shared code with the normal response path.
+    Http::Utility::sendLocalReply(
+        Grpc::Common::hasGrpcContentType(*parent_.request_headers_),
+        [&](ResponseHeaderMapPtr&& response_headers, bool end_stream) -> void {
+          parent_.response_headers_ = std::move(response_headers);
+          parent_.encodeHeadersInternal(*parent_.response_headers_, end_stream);
+        },
+        [&](Buffer::Instance& data, bool end_stream) -> void {
+          parent_.encodeDataInternal(data, end_stream);
+        },
+        parent_.state_.destroyed_, Http::Code::InternalServerError,
+        CodeUtility::toString(Http::Code::InternalServerError), absl::nullopt,
+        parent_.state_.is_head_request_);
+    parent_.maybeEndEncode(parent_.state_.local_complete_);
+  } else {
+    ENVOY_STREAM_LOG(debug,
+                     "Resetting stream. Response data too large and headers have already been sent",
+                     *this);
+    resetStream();
   }
 }
 

source/common/http/conn_manager_impl.h

@@ -394,7 +394,8 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
       callback(*parent_.buffered_response_data_.get());
     }
 
-    void responseDataTooLarge();
+    void responseDataExceedsHighWatermark();
+    void responseDataExceedsOverflowWatermark();
     void responseDataDrained();
 
     StreamEncoderFilterSharedPtr handle_;

source/common/http/http1/codec_impl.cc

@@ -295,6 +295,11 @@ void ConnectionImpl::copyToBuffer(const char* data, uint64_t length) {
   output_buffer_.add(data, length);
 }
 
+void ConnectionImpl::onAboveOverflowWatermark() {
+  // TODO(mergeconflict): log and increment counter
+  connection_.close(Network::ConnectionCloseType::NoFlush);
+}
+
 void StreamEncoderImpl::resetStream(StreamResetReason reason) {
   connection_.onResetStreamBase(reason);
 }
@@ -412,7 +417,8 @@ ConnectionImpl::ConnectionImpl(Network::Connection& connection, Stats::Scope& st
           "envoy.reloadable_features.connection_header_sanitization")),
       enable_trailers_(enable_trailers),
       output_buffer_([&]() -> void { this->onBelowLowWatermark(); },
-                     [&]() -> void { this->onAboveHighWatermark(); }),
+                     [&]() -> void { this->onAboveHighWatermark(); },
+                     [&]() -> void { this->onAboveOverflowWatermark(); }),
       max_headers_kb_(max_headers_kb), max_headers_count_(max_headers_count) {
   output_buffer_.setWatermarks(connection.bufferLimit());
   http_parser_init(&parser_, type);

source/common/http/http1/codec_impl.h

@@ -303,6 +303,11 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
    */
   virtual void sendProtocolError(absl::string_view details) PURE;
 
+  /**
+   * Called when output_buffer_ or the underlying connection exceed the "overflow" watermark.
+   */
+  void onAboveOverflowWatermark();
+
   /**
    * Called when output_buffer_ or the underlying connection go from below a low watermark to over
    * a high watermark.

source/common/http/http2/codec_impl.cc

@@ -214,6 +214,12 @@ void ConnectionImpl::StreamImpl::readDisable(bool disable) {
   }
 }
 
+void ConnectionImpl::StreamImpl::pendingRecvBufferOverflowWatermark() {
+  ENVOY_CONN_LOG(warn, "recv buffer overflow ", parent_.connection_);
+  // TODO(mergeconflict): increment a counter
+  resetStream(StreamResetReason::Overflow);
+}
+
 void ConnectionImpl::StreamImpl::pendingRecvBufferHighWatermark() {
   ENVOY_CONN_LOG(debug, "recv buffer over limit ", parent_.connection_);
   ASSERT(!pending_receive_buffer_high_watermark_called_);
@@ -266,6 +272,12 @@ void ConnectionImpl::ServerStreamImpl::decodeTrailers() {
       std::move(absl::get<RequestTrailerMapPtr>(headers_or_trailers_)));
 }
 
+void ConnectionImpl::StreamImpl::pendingSendBufferOverflowWatermark() {
+  ENVOY_CONN_LOG(warn, "send buffer overflow ", parent_.connection_);
+  // TODO(mergeconflict): increment a counter
+  resetStream(StreamResetReason::Overflow);
+}
+
 void ConnectionImpl::StreamImpl::pendingSendBufferHighWatermark() {
   ENVOY_CONN_LOG(debug, "send buffer over limit ", parent_.connection_);
   ASSERT(!pending_send_buffer_high_watermark_called_);