http/2: add stats and stream flush timeout (#139)

This commit adds a new stream flush timeout to guard against a
remote server that does not open window once an entire stream has
been buffered for flushing. Additional stats have also been added
to better understand the codecs view of active streams as well as
amount of data buffered.

Signed-off-by: Matt Klein <mklein@lyft.com>
Signed-off-by: Dmitri Dolguikh <ddolguik@redhat.com>

api/envoy/config/filter/network/http_connection_manager/v2/http_connection_manager.proto

@@ -300,6 +300,14 @@ message HttpConnectionManager {
   // is terminated with a 408 Request Timeout error code if no upstream response
   // header has been received, otherwise a stream reset occurs.
   //
+  // This timeout also specifies the amount of time that Envoy will wait for the peer to open enough
+  // window to write any remaining stream data once the entirety of stream data (local end stream is
+  // true) has been buffered pending available window. In other words, this timeout defends against
+  // a peer that does not release enough window to completely write the stream, even though all
+  // data has been proxied within available flow control windows. If the timeout is hit in this
+  // case, the :ref:`tx_flush_timeout <config_http_conn_man_stats_per_codec>` counter will be
+  // incremented.
+  //
   // Note that it is possible to idle timeout even if the wire traffic for a stream is non-idle, due
   // to the granularity of events presented to the connection manager. For example, while receiving
   // very large request headers, it may be the case that there is traffic regularly arriving on the

api/envoy/config/filter/network/http_connection_manager/v3alpha/http_connection_manager.proto

@@ -287,6 +287,14 @@ message HttpConnectionManager {
   // is terminated with a 408 Request Timeout error code if no upstream response
   // header has been received, otherwise a stream reset occurs.
   //
+  // This timeout also specifies the amount of time that Envoy will wait for the peer to open enough
+  // window to write any remaining stream data once the entirety of stream data (local end stream is
+  // true) has been buffered pending available window. In other words, this timeout defends against
+  // a peer that does not release enough window to completely write the stream, even though all
+  // data has been proxied within available flow control windows. If the timeout is hit in this
+  // case, the :ref:`tx_flush_timeout <config_http_conn_man_stats_per_codec>` counter will be
+  // incremented.
+  //
   // Note that it is possible to idle timeout even if the wire traffic for a stream is non-idle, due
   // to the granularity of events presented to the connection manager. For example, while receiving
   // very large request headers, it may be the case that there is traffic regularly arriving on the

docs/root/configuration/http/http_conn_man/stats.rst

@@ -137,7 +137,16 @@ All http2 statistics are rooted at *http2.*
    rx_reset, Counter, Total number of reset stream frames received by Envoy
    too_many_header_frames, Counter, Total number of times an HTTP2 connection is reset due to receiving too many headers frames. Envoy currently supports proxying at most one header frame for 100-Continue one non-100 response code header frame and one frame with trailers
    trailers, Counter, Total number of trailers seen on requests coming from downstream
+   tx_flush_timeout, Counter, Total number of :ref:`stream idle timeouts <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.stream_idle_timeout>` waiting for open stream window to flush the remainder of a stream
    tx_reset, Counter, Total number of reset stream frames transmitted by Envoy
+   streams_active, Gauge, Active streams as observed by the codec
+   pending_send_bytes, Gauge, Currently buffered body data in bytes waiting to be written when stream/connection window is opened.
+
+.. attention::
+
+  The HTTP/2 `streams_active` gauge may be greater than the HTTP connection manager
+  `downstream_rq_active` gauge due to differences in stream accounting between the codec and the
+  HTTP connection manager.
 
 Tracing statistics
 ------------------

docs/root/faq/configuration/timeouts.rst

@@ -52,7 +52,9 @@ context request/stream is interchangeable.
   <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.stream_idle_timeout>`
   is the amount of time that the connection manager will allow a stream to exist with no upstream
   or downstream activity. The default stream idle timeout is *5 minutes*. This timeout is strongly
-  recommended for streaming APIs (requests or responses that never end).
+  recommended for all requests (not just streaming requests/responses) as it additionally defends
+  against an HTTP/2 peer that does not open stream window once an entire response has been buffered
+  to be sent to a downstream client).
 
 Route timeouts
 ^^^^^^^^^^^^^^

docs/root/intro/version_history.rst

@@ -3,6 +3,8 @@ Version history
 
 1.12.5 (Pending)
 ================
+* http: the :ref:`stream_idle_timeout <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.stream_idle_timeout>`
+  now also defends against an HTTP/2 peer that does not open stream window once an entire response has been buffered to be sent to a downstream client.
 * listener: add runtime support for `per-listener limits <config_listeners_runtime>` on active/accepted connections.
 * overload management: add runtime support for :ref:`global limits <config_overload_manager>` on active/accepted connections.
 

include/envoy/http/codec.h

@@ -203,6 +203,13 @@ class Stream {
    * @return uint32_t the stream's configured buffer limits.
    */
   virtual uint32_t bufferLimit() PURE;
+
+  /**
+   * Set the flush timeout for the stream. At the codec level this is used to bound the amount of
+   * time the codec will wait to flush body data pending open stream window. It does *not* count
+   * small window updates as satisfying the idle timeout as this is a potential DoS vector.
+   */
+  virtual void setFlushTimeout(std::chrono::milliseconds timeout) PURE;
 };
 
 /**

source/common/http/codec_client.cc

@@ -17,7 +17,7 @@ namespace Http {
 CodecClient::CodecClient(Type type, Network::ClientConnectionPtr&& connection,
                          Upstream::HostDescriptionConstSharedPtr host,
                          Event::Dispatcher& dispatcher)
-    : type_(type), connection_(std::move(connection)), host_(host),
+    : type_(type), host_(host), connection_(std::move(connection)),
       idle_timeout_(host_->cluster().idleTimeout()) {
   if (type_ != Type::HTTP3) {
     // Make sure upstream connections process data and then the FIN, rather than processing

source/common/http/codec_client.h

@@ -155,9 +155,11 @@ class CodecClient : Logger::Loggable<Logger::Id::client>,
   }
 
   const Type type_;
-  ClientConnectionPtr codec_;
-  Network::ClientConnectionPtr connection_;
+  // The order of host_, connection_, and codec_ matter as during destruction each can refer to
+  // the previous, at least in tests.
   Upstream::HostDescriptionConstSharedPtr host_;
+  Network::ClientConnectionPtr connection_;
+  ClientConnectionPtr codec_;
   Event::TimerPtr idle_timer_;
   const absl::optional<std::chrono::milliseconds> idle_timeout_;
 

source/common/http/conn_manager_impl.cc

@@ -261,6 +261,7 @@ StreamDecoder& ConnectionManagerImpl::newStream(StreamEncoder& response_encoder,
   new_stream->state_.is_internally_created_ = is_internally_created;
   new_stream->response_encoder_ = &response_encoder;
   new_stream->response_encoder_->getStream().addCallbacks(*new_stream);
+  new_stream->response_encoder_->getStream().setFlushTimeout(new_stream->idle_timeout_ms_);
   new_stream->buffer_limit_ = new_stream->response_encoder_->getStream().bufferLimit();
   // If the network connection is backed up, the stream should be made aware of it on creation.
   // Both HTTP/1.x and HTTP/2 codecs handle this in StreamCallbackHelper::addCallbacks_.
@@ -866,7 +867,10 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
   if (hasCachedRoute()) {
     const Router::RouteEntry* route_entry = cached_route_.value()->routeEntry();
     if (route_entry != nullptr && route_entry->idleTimeout()) {
+      // TODO(mattklein123): Technically if the cached route changes, we should also see if the
+      // route idle timeout has changed and update the value.
       idle_timeout_ms_ = route_entry->idleTimeout().value();
+      response_encoder_->getStream().setFlushTimeout(idle_timeout_ms_);
       if (idle_timeout_ms_.count()) {
         // If we have a route-level idle timeout but no global stream idle timeout, create a timer.
         if (stream_idle_timer_ == nullptr) {

source/common/http/http1/codec_impl.h

@@ -65,6 +65,11 @@ class StreamEncoderImpl : public StreamEncoder,
   void resetStream(StreamResetReason reason) override;
   void readDisable(bool disable) override;
   uint32_t bufferLimit() override;
+  void setFlushTimeout(std::chrono::milliseconds) override {
+    // HTTP/1 has one stream per connection, thus any data encoded is immediately written to the
+    // connection, invoking any watermarks as necessary. There is no internal buffering that would
+    // require a flush timeout not already covered by other timeouts.
+  }
 
   void isResponseToHeadRequest(bool value) { is_response_to_head_request_ = value; }
 

source/common/http/http2/codec_impl.cc

@@ -56,11 +56,25 @@ ConnectionImpl::StreamImpl::StreamImpl(ConnectionImpl& parent, uint32_t buffer_l
       waiting_for_non_informational_headers_(false),
       pending_receive_buffer_high_watermark_called_(false),
       pending_send_buffer_high_watermark_called_(false), reset_due_to_messaging_error_(false) {
+  parent_.stats_.streams_active_.inc();
   if (buffer_limit > 0) {
     setWriteBufferWatermarks(buffer_limit / 2, buffer_limit);
   }
 }
 
+ConnectionImpl::StreamImpl::~StreamImpl() { ASSERT(stream_idle_timer_ == nullptr); }
+
+void ConnectionImpl::StreamImpl::destroy() {
+  if (stream_idle_timer_ != nullptr) {
+    // To ease testing and the destructor assertion.
+    stream_idle_timer_->disableTimer();
+    stream_idle_timer_.reset();
+  }
+
+  parent_.stats_.streams_active_.dec();
+  parent_.stats_.pending_send_bytes_.sub(pending_send_data_.length());
+}
+
 static void insertHeader(std::vector<nghttp2_nv>& headers, const HeaderEntry& header) {
   uint8_t flags = 0;
   if (header.key().type() == HeaderString::Type::Reference) {
@@ -130,6 +144,7 @@ void ConnectionImpl::StreamImpl::encodeTrailers(const HeaderMap& trailers) {
     // waiting on window updates. We need to save the trailers so that we can emit them later.
     ASSERT(!pending_trailers_);
     pending_trailers_ = std::make_unique<HeaderMapImpl>(trailers);
+    createPendingFlushTimer();
   } else {
     submitTrailers(trailers);
     parent_.sendPendingFrames();
@@ -262,6 +277,7 @@ int ConnectionImpl::StreamImpl::onDataSourceSend(const uint8_t* framehd, size_t
     return NGHTTP2_ERR_FLOODED;
   }
 
+  parent_.stats_.pending_send_bytes_.sub(length);
   output.move(pending_send_data_, length);
   parent_.connection_.write(output, false);
   return 0;
@@ -283,9 +299,30 @@ void ConnectionImpl::ServerStreamImpl::submitHeaders(const std::vector<nghttp2_n
   ASSERT(rc == 0);
 }
 
+void ConnectionImpl::ServerStreamImpl::createPendingFlushTimer() {
+  ASSERT(stream_idle_timer_ == nullptr);
+  if (stream_idle_timeout_.count() > 0) {
+    stream_idle_timer_ =
+        parent_.connection_.dispatcher().createTimer([this] { onPendingFlushTimer(); });
+    stream_idle_timer_->enableTimer(stream_idle_timeout_);
+  }
+}
+
+void ConnectionImpl::StreamImpl::onPendingFlushTimer() {
+  ENVOY_CONN_LOG(debug, "pending stream flush timeout", parent_.connection_);
+  stream_idle_timer_.reset();
+  parent_.stats_.tx_flush_timeout_.inc();
+  ASSERT(local_end_stream_ && !local_end_stream_sent_);
+  // This will emit a reset frame for this stream and close the stream locally. No reset callbacks
+  // will be run because higher layers think the stream is already finished.
+  resetStreamWorker(StreamResetReason::LocalReset);
+  parent_.sendPendingFrames();
+}
+
 void ConnectionImpl::StreamImpl::encodeData(Buffer::Instance& data, bool end_stream) {
   ASSERT(!local_end_stream_);
   local_end_stream_ = end_stream;
+  parent_.stats_.pending_send_bytes_.add(data.length());
   pending_send_data_.move(data);
   if (data_deferred_) {
     int rc = nghttp2_session_resume_data(parent_.session_, stream_id_);
@@ -295,6 +332,9 @@ void ConnectionImpl::StreamImpl::encodeData(Buffer::Instance& data, bool end_str
   }
 
   parent_.sendPendingFrames();
+  if (local_end_stream_ && pending_send_data_.length() > 0) {
+    createPendingFlushTimer();
+  }
 }
 
 void ConnectionImpl::StreamImpl::resetStream(StreamResetReason reason) {
@@ -373,8 +413,10 @@ bool checkRuntimeOverride(bool config_value, const char* override_key) {
 ConnectionImpl::ConnectionImpl(Network::Connection& connection, Stats::Scope& stats,
                                const Http2Settings& http2_settings, const uint32_t max_headers_kb,
                                const uint32_t max_headers_count)
-    : stats_{ALL_HTTP2_CODEC_STATS(POOL_COUNTER_PREFIX(stats, "http2."))}, connection_(connection),
-      max_headers_kb_(max_headers_kb), max_headers_count_(max_headers_count),
+    : stats_{ALL_HTTP2_CODEC_STATS(POOL_COUNTER_PREFIX(stats, "http2."),
+                                   POOL_GAUGE_PREFIX(stats, "http2."))},
+      connection_(connection), max_headers_kb_(max_headers_kb),
+      max_headers_count_(max_headers_count),
       per_stream_buffer_limit_(http2_settings.initial_stream_window_size_),
       stream_error_on_invalid_http_messaging_(checkRuntimeOverride(
           http2_settings.stream_error_on_invalid_http_messaging_, InvalidHttpMessagingOverrideKey)),
@@ -396,7 +438,12 @@ ConnectionImpl::ConnectionImpl(Network::Connection& connection, Stats::Scope& st
           http2_settings.max_inbound_window_update_frames_per_data_frame_sent_)),
       dispatching_(false), raised_goaway_(false), pending_deferred_reset_(false) {}
 
-ConnectionImpl::~ConnectionImpl() { nghttp2_session_del(session_); }
+ConnectionImpl::~ConnectionImpl() {
+  for (const auto& stream : active_streams_) {
+    stream->destroy();
+  }
+  nghttp2_session_del(session_);
+}
 
 void ConnectionImpl::dispatch(Buffer::Instance& data) {
   ENVOY_CONN_LOG(trace, "dispatching {} bytes", connection_, data.length());
@@ -750,6 +797,7 @@ int ConnectionImpl::onStreamClose(int32_t stream_id, uint32_t error_code) {
       stream->runResetCallbacks(reason);
     }
 
+    stream->destroy();
     connection_.dispatcher().deferredDelete(stream->removeFromList(active_streams_));
     // Any unconsumed data must be consumed before the stream is deleted.
     // nghttp2 does not appear to track this internally, and any stream deleted

source/common/http/http2/codec_impl.h

@@ -40,7 +40,7 @@ const std::string CLIENT_MAGIC_PREFIX = "PRI * HTTP/2";
 /**
  * All stats for the HTTP/2 codec. @see stats_macros.h
  */
-#define ALL_HTTP2_CODEC_STATS(COUNTER)                                                             \
+#define ALL_HTTP2_CODEC_STATS(COUNTER, GAUGE)                                                      \
   COUNTER(dropped_headers_with_underscores)                                                        \
   COUNTER(header_overflow)                                                                         \
   COUNTER(headers_cb_no_stream)                                                                    \
@@ -54,13 +54,16 @@ const std::string CLIENT_MAGIC_PREFIX = "PRI * HTTP/2";
   COUNTER(rx_reset)                                                                                \
   COUNTER(too_many_header_frames)                                                                  \
   COUNTER(trailers)                                                                                \
-  COUNTER(tx_reset)
+  COUNTER(tx_flush_timeout)                                                                        \
+  COUNTER(tx_reset)                                                                                \
+  GAUGE(streams_active, Accumulate)                                                                \
+  GAUGE(pending_send_bytes, Accumulate)
 
 /**
  * Wrapper struct for the HTTP/2 codec stats. @see stats_macros.h
  */
 struct CodecStats {
-  ALL_HTTP2_CODEC_STATS(GENERATE_COUNTER_STRUCT)
+  ALL_HTTP2_CODEC_STATS(GENERATE_COUNTER_STRUCT, GENERATE_GAUGE_STRUCT)
 };
 
 class Utility {
@@ -150,6 +153,11 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
                       public StreamCallbackHelper {
 
     StreamImpl(ConnectionImpl& parent, uint32_t buffer_limit);
+    ~StreamImpl() override;
+    // TODO(mattklein123): Optimally this would be done in the destructor but there are currently
+    // deferred delete lifetime issues that need sorting out if the destructor of the stream is
+    // going to be able to refer to the parent connection.
+    void destroy();
 
     StreamImpl* base() { return this; }
     ssize_t onDataSourceRead(uint64_t length, uint32_t* data_flags);
@@ -161,6 +169,8 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
                                nghttp2_data_provider* provider) PURE;
     void submitTrailers(const HeaderMap& trailers);
     void submitMetadata();
+    virtual void createPendingFlushTimer() PURE;
+    void onPendingFlushTimer();
 
     // Http::StreamEncoder
     void encode100ContinueHeaders(const HeaderMap& headers) override;
@@ -176,6 +186,9 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
     void resetStream(StreamResetReason reason) override;
     void readDisable(bool disable) override;
     uint32_t bufferLimit() override { return pending_recv_data_.highWatermark(); }
+    void setFlushTimeout(std::chrono::milliseconds timeout) override {
+      stream_idle_timeout_ = timeout;
+    }
 
     void setWriteBufferWatermarks(uint32_t low_watermark, uint32_t high_watermark) {
       pending_recv_data_.setWatermarks(low_watermark, high_watermark);
@@ -231,6 +244,9 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
     bool pending_receive_buffer_high_watermark_called_ : 1;
     bool pending_send_buffer_high_watermark_called_ : 1;
     bool reset_due_to_messaging_error_ : 1;
+    // See HttpConnectionManager.stream_idle_timeout.
+    std::chrono::milliseconds stream_idle_timeout_{};
+    Event::TimerPtr stream_idle_timer_;
   };
 
   using StreamImplPtr = std::unique_ptr<StreamImpl>;
@@ -253,6 +269,10 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
         Http::Utility::transformUpgradeResponseFromH2toH1(*headers_, upgrade_type_);
       }
     }
+    void createPendingFlushTimer() override {
+      // Client streams do not create a flush timer because we currently assume that any failure
+      // to flush would be covered by a request/stream/etc. timeout.
+    }
     std::string upgrade_type_;
   };
 
@@ -278,6 +298,7 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
         Http::Utility::transformUpgradeRequestFromH2toH1(*headers_);
       }
     }
+    void createPendingFlushTimer() override;
   };
 
   ConnectionImpl* base() { return this; }

source/extensions/quic_listeners/quiche/envoy_quic_server_stream.h

@@ -34,6 +34,9 @@ class EnvoyQuicServerStream : public quic::QuicSpdyServerStreamBase, public Envo
   // Http::Stream
   void resetStream(Http::StreamResetReason reason) override;
   void readDisable(bool disable) override;
+  void setFlushTimeout(std::chrono::milliseconds) override {
+    // TODO(mattklein123): Actually implement this for HTTP/3 similar to HTTP/2.
+  }
   // quic::QuicSpdyStream
   void OnBodyAvailable() override;
   void OnStreamReset(const quic::QuicRstStreamFrame& frame) override;

test/common/http/conn_manager_impl_test.cc

@@ -167,6 +167,7 @@ class HttpConnectionManagerImplTest : public testing::Test, public ConnectionMan
     EXPECT_CALL(stream_, addCallbacks(_))
         .WillOnce(Invoke(
             [&](Http::StreamCallbacks& callbacks) -> void { stream_callbacks_ = &callbacks; }));
+    EXPECT_CALL(stream_, setFlushTimeout(_));
     EXPECT_CALL(stream_, bufferLimit()).WillOnce(Return(initial_buffer_limit_));
   }