add override knob for stream flush timeout

api/envoy/config/route/v3/route_components.proto

@@ -793,7 +793,7 @@ message CorsPolicy {
   google.protobuf.BoolValue forward_not_matching_preflights = 13;
 }
 
-// [#next-free-field: 42]
+// [#next-free-field: 43]
 message RouteAction {
   option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.route.RouteAction";
 
@@ -1318,8 +1318,28 @@ message RouteAction {
   // If the :ref:`overload action <config_overload_manager_overload_actions>` "envoy.overload_actions.reduce_timeouts"
   // is configured, this timeout is scaled according to the value for
   // :ref:`HTTP_DOWNSTREAM_STREAM_IDLE <envoy_v3_api_enum_value_config.overload.v3.ScaleTimersOverloadActionConfig.TimerType.HTTP_DOWNSTREAM_STREAM_IDLE>`.
+  //
+  // This timeout may also be used in place of ``flush_timeout`` in very specific cases. See the
+  // documentation for ``flush_timeout`` for more details.
   google.protobuf.Duration idle_timeout = 24;
 
+  // Specifies the codec stream flush timeout for the route.
+  //
+  // If not specified, the first preference is the global :ref:`stream_flush_timeout
+  // <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.stream_flush_timeout>`,
+  // but only if explicitly configured.
+  //
+  // If neither the explicit HCM-wide flush timeout nor this route-specific flush timeout is configured,
+  // the route's stream idle timeout is reused for this timeout. This is for
+  // backwards compatibility since both behaviors were historically controlled by the one timeout.
+  //
+  // If the route also does not have an idle timeout configured, the global :ref:`stream_idle_timeout
+  // <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.stream_idle_timeout>`. used, again
+  // for backwards compatibility. That timeout defaults to 5 minutes.
+  //
+  // A value of 0 via any of the above paths will completely disable the timeout for a given route.
+  google.protobuf.Duration flush_timeout = 42;
+
   // Specifies how to send request over TLS early data.
   // If absent, allows `safe HTTP requests <https://www.rfc-editor.org/rfc/rfc7231#section-4.2.1>`_ to be sent on early data.
   // [#extension-category: envoy.route.early_data_policy]

api/envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.proto

@@ -37,7 +37,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // HTTP connection manager :ref:`configuration overview <config_http_conn_man>`.
 // [#extension: envoy.filters.network.http_connection_manager]
 
-// [#next-free-field: 59]
+// [#next-free-field: 60]
 message HttpConnectionManager {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.filter.network.http_connection_manager.v2.HttpConnectionManager";
@@ -527,16 +527,6 @@ 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 :ref:`max_stream_duration
-  // <envoy_v3_api_field_config.core.v3.HttpProtocolOptions.max_stream_duration>` does not apply to
-  // this corner case.
-  //
   // If the :ref:`overload action <config_overload_manager_overload_actions>` "envoy.overload_actions.reduce_timeouts"
   // is configured, this timeout is scaled according to the value for
   // :ref:`HTTP_DOWNSTREAM_STREAM_IDLE <envoy_v3_api_enum_value_config.overload.v3.ScaleTimersOverloadActionConfig.TimerType.HTTP_DOWNSTREAM_STREAM_IDLE>`.
@@ -549,9 +539,29 @@ message HttpConnectionManager {
   //
   // A value of 0 will completely disable the connection manager stream idle
   // timeout, although per-route idle timeout overrides will continue to apply.
+  //
+  // This timeout is also used as the default value for :ref:`stream_flush_timeout
+  // <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.stream_flush_timeout>`.
   google.protobuf.Duration stream_idle_timeout = 24
       [(udpa.annotations.security).configure_for_untrusted_downstream = true];
 
+  // The stream flush timeout for connections managed by the connection manager.
+  //
+  // If not specified, the value of stream_idle_timeout is used. This is for backwards compatibility
+  // since this was the original behavior. In essence this timeout is an override for the
+  // stream_idle_timeout that applies specifically to the end of stream flush case.
+  //
+  // This timeout 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 :ref:`max_stream_duration
+  // <envoy_v3_api_field_config.core.v3.HttpProtocolOptions.max_stream_duration>` does not apply to
+  // this corner case.
+  google.protobuf.Duration stream_flush_timeout = 59;
+
   // The amount of time that Envoy will wait for the entire request to be received.
   // The timer is activated when the request is initiated, and is disarmed when the last byte of the
   // request is sent upstream (i.e. all decoding filters have processed the request), OR when the

changelogs/current.yaml

@@ -296,6 +296,11 @@ new_features:
 - area: http
   change: |
     Added ``upstream_rq_per_cx`` histogram to track requests per connection for monitoring connection reuse efficiency.
-
+- area: http
+  change: |
+    Added
+    :ref:`stream_flush_timeout
+    <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.stream_flush_timeout>`
+    to allow for configuring a stream flush timeout independently from the stream idle timeout.
 
 deprecated:

envoy/router/router.h

@@ -1033,6 +1033,12 @@ class RouteEntry : public ResponseEntry {
    */
   virtual absl::optional<std::chrono::milliseconds> idleTimeout() const PURE;
 
+  /**
+   * @return optional<std::chrono::milliseconds> the route's flush timeout. Zero indicates a
+   *         disabled idle timeout, while nullopt indicates deference to the global timeout.
+   */
+  virtual absl::optional<std::chrono::milliseconds> flushTimeout() const PURE;
+
   /**
    * @return true if new style max_stream_duration config should be used over the old style.
    */

source/common/http/codec_helper.h

@@ -89,11 +89,11 @@ class StreamCallbackHelper {
 class MultiplexedStreamImplBase : public Stream, public StreamCallbackHelper {
 public:
   MultiplexedStreamImplBase(Event::Dispatcher& dispatcher) : dispatcher_(dispatcher) {}
-  ~MultiplexedStreamImplBase() override { ASSERT(stream_idle_timer_ == nullptr); }
+  ~MultiplexedStreamImplBase() override { ASSERT(stream_flush_timer_ == nullptr); }
   // 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.
-  virtual void destroy() { disarmStreamIdleTimer(); }
+  virtual void destroy() { disarmStreamFlushTimer(); }
 
   void onLocalEndStream() {
     ASSERT(local_end_stream_);
@@ -102,11 +102,11 @@ class MultiplexedStreamImplBase : public Stream, public StreamCallbackHelper {
     }
   }
 
-  void disarmStreamIdleTimer() {
-    if (stream_idle_timer_ != nullptr) {
+  void disarmStreamFlushTimer() {
+    if (stream_flush_timer_ != nullptr) {
       // To ease testing and the destructor assertion.
-      stream_idle_timer_->disableTimer();
-      stream_idle_timer_.reset();
+      stream_flush_timer_->disableTimer();
+      stream_flush_timer_.reset();
     }
   }
 
@@ -117,28 +117,28 @@ class MultiplexedStreamImplBase : public Stream, public StreamCallbackHelper {
 
 protected:
   void setFlushTimeout(std::chrono::milliseconds timeout) override {
-    stream_idle_timeout_ = timeout;
+    stream_flush_timeout_ = timeout;
   }
 
   void createPendingFlushTimer() {
-    ASSERT(stream_idle_timer_ == nullptr);
-    if (stream_idle_timeout_.count() > 0) {
-      stream_idle_timer_ = dispatcher_.createTimer([this] { onPendingFlushTimer(); });
-      stream_idle_timer_->enableTimer(stream_idle_timeout_);
+    ASSERT(stream_flush_timer_ == nullptr);
+    if (stream_flush_timeout_.count() > 0) {
+      stream_flush_timer_ = dispatcher_.createTimer([this] { onPendingFlushTimer(); });
+      stream_flush_timer_->enableTimer(stream_flush_timeout_);
     }
   }
 
-  virtual void onPendingFlushTimer() { stream_idle_timer_.reset(); }
+  virtual void onPendingFlushTimer() { stream_flush_timer_.reset(); }
 
   virtual bool hasPendingData() PURE;
 
   CodecEventCallbacks* codec_callbacks_{nullptr};
 
 private:
   Event::Dispatcher& dispatcher_;
-  // See HttpConnectionManager.stream_idle_timeout.
-  std::chrono::milliseconds stream_idle_timeout_{};
-  Event::TimerPtr stream_idle_timer_;
+  // See HttpConnectionManager.stream_flush_timeout.
+  std::chrono::milliseconds stream_flush_timeout_{};
+  Event::TimerPtr stream_flush_timer_;
 };
 
 } // namespace Http

source/common/http/conn_manager_config.h

@@ -311,6 +311,12 @@ class ConnectionManagerConfig {
    */
   virtual std::chrono::milliseconds streamIdleTimeout() const PURE;
 
+  /**
+   * @return per-stream flush timeout for incoming connection manager connections. Zero indicates a
+   *         disabled idle timeout.
+   */
+  virtual absl::optional<std::chrono::milliseconds> streamFlushTimeout() const PURE;
+
   /**
    * @return request timeout for incoming connection manager connections. Zero indicates
    *         a disabled request timeout.

source/common/http/conn_manager_impl.cc

@@ -431,7 +431,12 @@ RequestDecoder& ConnectionManagerImpl::newStream(ResponseEncoder& response_encod
   new_stream->response_encoder_ = &response_encoder;
   new_stream->response_encoder_->getStream().addCallbacks(*new_stream);
   new_stream->response_encoder_->getStream().registerCodecEventCallbacks(new_stream.get());
-  new_stream->response_encoder_->getStream().setFlushTimeout(new_stream->idle_timeout_ms_);
+  if (config_->streamFlushTimeout().has_value()) {
+    new_stream->response_encoder_->getStream().setFlushTimeout(
+        config_->streamFlushTimeout().value());
+  } else {
+    new_stream->response_encoder_->getStream().setFlushTimeout(config_->streamIdleTimeout());
+  }
   new_stream->streamInfo().setDownstreamBytesMeter(response_encoder.getStream().bytesMeter());
   // 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::addCallbacksHelper.
@@ -846,6 +851,8 @@ ConnectionManagerImpl::ActiveStream::ActiveStream(ConnectionManagerImpl& connect
                       connection_manager_.overload_manager_),
       request_response_timespan_(new Stats::HistogramCompletableTimespanImpl(
           connection_manager_.stats_.named_.downstream_rq_time_, connection_manager_.timeSource())),
+      has_explicit_global_flush_timeout_(
+          connection_manager.config_->streamFlushTimeout().has_value()),
       header_validator_(
           connection_manager.config_->makeHeaderValidator(connection_manager.codec_->protocol())),
       trace_refresh_after_route_refresh_(Runtime::runtimeFeatureEnabled(
@@ -2216,30 +2223,43 @@ void ConnectionManagerImpl::ActiveStream::setVirtualHostRoute(
 
   refreshTracing();
   refreshDurationTimeout();
-  refreshIdleTimeout();
+  refreshIdleAndFlushTimeouts();
 }
 
-void ConnectionManagerImpl::ActiveStream::refreshIdleTimeout() {
-  if (hasCachedRoute()) {
-    const Router::RouteEntry* route_entry = cached_route_.value()->routeEntry();
-    if (route_entry != nullptr && route_entry->idleTimeout()) {
-      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) {
-          stream_idle_timer_ = connection_manager_.dispatcher_->createScaledTimer(
-              Event::ScaledTimerType::HttpDownstreamIdleStreamTimeout,
-              [this]() -> void { onIdleTimeout(); });
-        }
-      } else if (stream_idle_timer_ != nullptr) {
-        // If we had a global stream idle timeout but the route-level idle timeout is set to zero
-        // (to override), we disable the idle timer.
-        stream_idle_timer_->disableTimer();
-        stream_idle_timer_ = nullptr;
+void ConnectionManagerImpl::ActiveStream::refreshIdleAndFlushTimeouts() {
+  if (!hasCachedRoute()) {
+    return;
+  }
+  const Router::RouteEntry* route_entry = cached_route_.value()->routeEntry();
+  if (route_entry == nullptr) {
+    return;
+  }
+
+  if (route_entry->idleTimeout().has_value()) {
+    idle_timeout_ms_ = route_entry->idleTimeout().value();
+    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) {
+        stream_idle_timer_ = connection_manager_.dispatcher_->createScaledTimer(
+            Event::ScaledTimerType::HttpDownstreamIdleStreamTimeout,
+            [this]() -> void { onIdleTimeout(); });
       }
+    } else if (stream_idle_timer_ != nullptr) {
+      // If we had a global stream idle timeout but the route-level idle timeout is set to zero
+      // (to override), we disable the idle timer.
+      stream_idle_timer_->disableTimer();
+      stream_idle_timer_ = nullptr;
     }
   }
+
+  if (route_entry->flushTimeout().has_value()) {
+    response_encoder_->getStream().setFlushTimeout(route_entry->flushTimeout().value());
+  } else if (!has_explicit_global_flush_timeout_ && route_entry->idleTimeout().has_value()) {
+    // If there is no route-level flush timeout, and the global flush timeout was also inherited
+    // from the idle timeout, also inherit the route-level idle timeout. This is for backwards
+    // compatibility.
+    response_encoder_->getStream().setFlushTimeout(idle_timeout_ms_);
+  }
 }
 
 void ConnectionManagerImpl::ActiveStream::refreshAccessLogFlushTimer() {

source/common/http/conn_manager_impl.h

@@ -333,7 +333,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     void refreshCachedRoute(const Router::RouteCallback& cb);
 
     void refreshDurationTimeout();
-    void refreshIdleTimeout();
+    void refreshIdleAndFlushTimeouts();
     void refreshAccessLogFlushTimer();
     void refreshTracing();
 
@@ -473,6 +473,11 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     Event::TimerPtr access_log_flush_timer_;
 
     std::chrono::milliseconds idle_timeout_ms_{};
+    // If an explicit global flush timeout is set, never override it with the route entry idle
+    // timeout. If there is no explicit global flush timeout, then override with the route entry
+    // idle timeout if it exists. This is to prevent breaking existing user expectations that the
+    // flush timeout is the same as the idle timeout.
+    const bool has_explicit_global_flush_timeout_{false};
     State state_;
 
     // Snapshot of the route configuration at the time of request is started. This is used to ensure

source/common/http/http2/codec_impl.cc

@@ -1234,7 +1234,7 @@ int ConnectionImpl::onFrameSend(int32_t stream_id, size_t length, uint8_t type,
       // teardown. As part of the work to remove exceptions we should aim to clean up all of this
       // error handling logic and only handle this type of case at the end of dispatch.
       for (auto& stream : active_streams_) {
-        stream->disarmStreamIdleTimer();
+        stream->disarmStreamFlushTimer();
       }
       return ERR_CALLBACK_FAILURE;
     }

source/common/http/null_route_impl.h

@@ -174,6 +174,7 @@ struct RouteEntryImpl : public Router::RouteEntry {
   }
   bool usingNewTimeouts() const override { return false; }
   absl::optional<std::chrono::milliseconds> idleTimeout() const override { return absl::nullopt; }
+  absl::optional<std::chrono::milliseconds> flushTimeout() const override { return absl::nullopt; }
   absl::optional<std::chrono::milliseconds> maxStreamDuration() const override {
     return absl::nullopt;
   }

source/common/router/config_impl.cc

@@ -1188,6 +1188,7 @@ RouteEntryImplBase::OptionalTimeouts RouteEntryImplBase::buildOptionalTimeouts(
   // Calculate how many values are actually set, to initialize `OptionalTimeouts` packed_struct,
   // avoiding memory re-allocation on each set() call.
   int num_timeouts_set = route.has_idle_timeout() ? 1 : 0;
+  num_timeouts_set += route.has_flush_timeout() ? 1 : 0;
   num_timeouts_set += route.has_max_grpc_timeout() ? 1 : 0;
   num_timeouts_set += route.has_grpc_timeout_offset() ? 1 : 0;
   if (route.has_max_stream_duration()) {
@@ -1200,6 +1201,10 @@ RouteEntryImplBase::OptionalTimeouts RouteEntryImplBase::buildOptionalTimeouts(
     timeouts.set<OptionalTimeoutNames::IdleTimeout>(
         std::chrono::milliseconds(PROTOBUF_GET_MS_REQUIRED(route, idle_timeout)));
   }
+  if (route.has_flush_timeout()) {
+    timeouts.set<OptionalTimeoutNames::FlushTimeout>(
+        std::chrono::milliseconds(PROTOBUF_GET_MS_REQUIRED(route, flush_timeout)));
+  }
   if (route.has_max_grpc_timeout()) {
     timeouts.set<OptionalTimeoutNames::MaxGrpcTimeout>(
         std::chrono::milliseconds(PROTOBUF_GET_MS_REQUIRED(route, max_grpc_timeout)));

source/common/router/config_impl.h

@@ -747,13 +747,17 @@ class RouteEntryImplBase : public RouteEntryAndRoute,
     GrpcTimeoutHeaderMax,
     GrpcTimeoutHeaderOffset,
     MaxGrpcTimeout,
-    GrpcTimeoutOffset
+    GrpcTimeoutOffset,
+    FlushTimeout,
   };
-  using OptionalTimeouts = PackedStruct<std::chrono::milliseconds, 6, OptionalTimeoutNames>;
+  using OptionalTimeouts = PackedStruct<std::chrono::milliseconds, 7, OptionalTimeoutNames>;
 
   absl::optional<std::chrono::milliseconds> idleTimeout() const override {
     return getOptionalTimeout<OptionalTimeoutNames::IdleTimeout>();
   }
+  absl::optional<std::chrono::milliseconds> flushTimeout() const override {
+    return getOptionalTimeout<OptionalTimeoutNames::FlushTimeout>();
+  }
   absl::optional<std::chrono::milliseconds> maxStreamDuration() const override {
     return getOptionalTimeout<OptionalTimeoutNames::MaxStreamDuration>();
   }

source/common/router/delegating_route_impl.cc

@@ -93,6 +93,10 @@ absl::optional<std::chrono::milliseconds> DelegatingRouteEntry::idleTimeout() co
   return base_route_entry_->idleTimeout();
 }
 
+absl::optional<std::chrono::milliseconds> DelegatingRouteEntry::flushTimeout() const {
+  return base_route_entry_->flushTimeout();
+}
+
 bool DelegatingRouteEntry::usingNewTimeouts() const {
   return base_route_entry_->usingNewTimeouts();
 }

source/common/router/delegating_route_impl.h

@@ -108,6 +108,7 @@ class DelegatingRouteEntry : public DelegatingRouteBase<RouteEntryAndRoute> {
   const std::vector<Router::ShadowPolicyPtr>& shadowPolicies() const override;
   std::chrono::milliseconds timeout() const override;
   absl::optional<std::chrono::milliseconds> idleTimeout() const override;
+  absl::optional<std::chrono::milliseconds> flushTimeout() const override;
   bool usingNewTimeouts() const override;
   absl::optional<std::chrono::milliseconds> maxStreamDuration() const override;
   absl::optional<std::chrono::milliseconds> grpcTimeoutHeaderMax() const override;