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

@@ -272,6 +272,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

@@ -117,14 +117,14 @@ 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) {
+    if (stream_flush_timeout_.count() > 0) {
       stream_idle_timer_ = dispatcher_.createTimer([this] { onPendingFlushTimer(); });
-      stream_idle_timer_->enableTimer(stream_idle_timeout_);
+      stream_idle_timer_->enableTimer(stream_flush_timeout_);
     }
   }
 
@@ -137,7 +137,7 @@ class MultiplexedStreamImplBase : public Stream, public StreamCallbackHelper {
 private:
   Event::Dispatcher& dispatcher_;
   // See HttpConnectionManager.stream_idle_timeout.
-  std::chrono::milliseconds stream_idle_timeout_{};
+  std::chrono::milliseconds stream_flush_timeout_{};
   Event::TimerPtr stream_idle_timer_;
 };
 

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(
@@ -2217,14 +2224,17 @@ void ConnectionManagerImpl::ActiveStream::setVirtualHostRoute(
   refreshTracing();
   refreshDurationTimeout();
   refreshIdleTimeout();
+  refreshFlushTimeout();
 }
 
 void ConnectionManagerImpl::ActiveStream::refreshIdleTimeout() {
   if (hasCachedRoute()) {
     const Router::RouteEntry* route_entry = cached_route_.value()->routeEntry();
-    if (route_entry != nullptr && route_entry->idleTimeout()) {
+    if (route_entry == nullptr) {
+      return;
+    }
+    if (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) {
@@ -2242,6 +2252,23 @@ void ConnectionManagerImpl::ActiveStream::refreshIdleTimeout() {
   }
 }
 
+void ConnectionManagerImpl::ActiveStream::refreshFlushTimeout() {
+  if (!hasCachedRoute()) {
+    return;
+  }
+  const Router::RouteEntry* route_entry = cached_route_.value()->routeEntry();
+  if (route_entry == nullptr) {
+    return;
+  }
+  if (route_entry->flushTimeout().has_value()) {
+    response_encoder_->getStream().setFlushTimeout(route_entry->flushTimeout().value());
+    return;
+  }
+  if (!has_explicit_global_flush_timeout_ && route_entry->idleTimeout().has_value()) {
+    response_encoder_->getStream().setFlushTimeout(route_entry->idleTimeout().value());
+  }
+}
+
 void ConnectionManagerImpl::ActiveStream::refreshAccessLogFlushTimer() {
   if (connection_manager_.config_->accessLogFlushInterval().has_value()) {
     access_log_flush_timer_->enableTimer(

source/common/http/conn_manager_impl.h

@@ -334,6 +334,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
 
     void refreshDurationTimeout();
     void refreshIdleTimeout();
+    void refreshFlushTimeout();
     void refreshAccessLogFlushTimer();
     void refreshTracing();
 
@@ -473,6 +474,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/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;

source/extensions/filters/network/http_connection_manager/config.cc

@@ -395,6 +395,8 @@ HttpConnectionManagerConfig::HttpConnectionManagerConfig(
           PROTOBUF_GET_OPTIONAL_MS(config.common_http_protocol_options(), max_stream_duration)),
       stream_idle_timeout_(
           PROTOBUF_GET_MS_OR_DEFAULT(config, stream_idle_timeout, StreamIdleTimeoutMs)),
+      stream_flush_timeout_(
+          PROTOBUF_GET_MS_OR_DEFAULT(config, stream_flush_timeout, stream_idle_timeout_.count())),
       request_timeout_(PROTOBUF_GET_MS_OR_DEFAULT(config, request_timeout, RequestTimeoutMs)),
       request_headers_timeout_(
           PROTOBUF_GET_MS_OR_DEFAULT(config, request_headers_timeout, RequestHeaderTimeoutMs)),

source/extensions/filters/network/http_connection_manager/config.h

@@ -184,6 +184,9 @@ class HttpConnectionManagerConfig : Logger::Loggable<Logger::Id::config>,
     return http1_safe_max_connection_duration_;
   }
   std::chrono::milliseconds streamIdleTimeout() const override { return stream_idle_timeout_; }
+  absl::optional<std::chrono::milliseconds> streamFlushTimeout() const override {
+    return stream_flush_timeout_;
+  }
   std::chrono::milliseconds requestTimeout() const override { return request_timeout_; }
   std::chrono::milliseconds requestHeadersTimeout() const override {
     return request_headers_timeout_;
@@ -330,6 +333,7 @@ class HttpConnectionManagerConfig : Logger::Loggable<Logger::Id::config>,
   const bool http1_safe_max_connection_duration_;
   absl::optional<std::chrono::milliseconds> max_stream_duration_;
   std::chrono::milliseconds stream_idle_timeout_;
+  absl::optional<std::chrono::milliseconds> stream_flush_timeout_;
   std::chrono::milliseconds request_timeout_;
   std::chrono::milliseconds request_headers_timeout_;
   Router::RouteConfigProviderSharedPtr route_config_provider_;

source/server/admin/admin.h

@@ -152,6 +152,9 @@ class AdminImpl : public Admin,
   uint32_t maxRequestHeadersKb() const override { return max_request_headers_kb_; }
   uint32_t maxRequestHeadersCount() const override { return max_request_headers_count_; }
   std::chrono::milliseconds streamIdleTimeout() const override { return {}; }
+  absl::optional<std::chrono::milliseconds> streamFlushTimeout() const override {
+    return std::nullopt;
+  }
   std::chrono::milliseconds requestTimeout() const override { return {}; }
   std::chrono::milliseconds requestHeadersTimeout() const override { return {}; }
   std::chrono::milliseconds delayedCloseTimeout() const override { return {}; }

test/common/http/conn_manager_impl_fuzz_test.cc

@@ -160,6 +160,9 @@ class FuzzConfig : public ConnectionManagerConfig {
     return max_stream_duration_;
   }
   std::chrono::milliseconds streamIdleTimeout() const override { return stream_idle_timeout_; }
+  absl::optional<std::chrono::milliseconds> streamFlushTimeout() const override {
+    return stream_flush_timeout_;
+  }
   std::chrono::milliseconds requestTimeout() const override { return request_timeout_; }
   std::chrono::milliseconds requestHeadersTimeout() const override {
     return request_headers_timeout_;
@@ -282,6 +285,7 @@ class FuzzConfig : public ConnectionManagerConfig {
   bool http1_safe_max_connection_duration_{false};
   absl::optional<std::chrono::milliseconds> max_stream_duration_;
   std::chrono::milliseconds stream_idle_timeout_{};
+  std::chrono::milliseconds stream_flush_timeout_{};
   std::chrono::milliseconds request_timeout_{};
   std::chrono::milliseconds request_headers_timeout_{};
   std::chrono::milliseconds delayed_close_timeout_{};