http: max stream duration upstream support

api/envoy/api/v2/core/protocol.proto

@@ -85,8 +85,6 @@ message HttpProtocolOptions {
 
   // Total duration to keep alive an HTTP request/response stream. If the time limit is reached the stream will be
   // reset independent of any other timeouts. If not specified, this value is not set.
-  // The current implementation implements this timeout on downstream connections only.
-  // [#comment:TODO(shikugawa): add this functionality to upstream.]
   google.protobuf.Duration max_stream_duration = 4;
 
   // Action to take when a client request with a header name containing underscore characters is received.

api/envoy/config/accesslog/v3/accesslog.proto

@@ -240,6 +240,7 @@ message ResponseFlagFilter {
         in: "SI"
         in: "IH"
         in: "DPE"
+        in: "UMSDR"
       }
     }
   }];

api/envoy/config/accesslog/v4alpha/accesslog.proto

@@ -239,6 +239,7 @@ message ResponseFlagFilter {
         in: "SI"
         in: "IH"
         in: "DPE"
+        in: "UMSDR"
       }
     }
   }];

api/envoy/config/core/v3/protocol.proto

@@ -92,8 +92,6 @@ message HttpProtocolOptions {
 
   // Total duration to keep alive an HTTP request/response stream. If the time limit is reached the stream will be
   // reset independent of any other timeouts. If not specified, this value is not set.
-  // The current implementation implements this timeout on downstream connections only.
-  // [#comment:TODO(shikugawa): add this functionality to upstream.]
   google.protobuf.Duration max_stream_duration = 4;
 
   // Action to take when a client request with a header name containing underscore characters is received.

api/envoy/config/core/v4alpha/protocol.proto

@@ -92,8 +92,6 @@ message HttpProtocolOptions {
 
   // Total duration to keep alive an HTTP request/response stream. If the time limit is reached the stream will be
   // reset independent of any other timeouts. If not specified, this value is not set.
-  // The current implementation implements this timeout on downstream connections only.
-  // [#comment:TODO(shikugawa): add this functionality to upstream.]
   google.protobuf.Duration max_stream_duration = 4;
 
   // Action to take when a client request with a header name containing underscore characters is received.

api/envoy/data/accesslog/v3/accesslog.proto

@@ -186,7 +186,7 @@ message AccessLogCommon {
 }
 
 // Flags indicating occurrences during request/response processing.
-// [#next-free-field: 20]
+// [#next-free-field: 21]
 message ResponseFlags {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.data.accesslog.v2.ResponseFlags";
@@ -263,6 +263,9 @@ message ResponseFlags {
 
   // Indicates there was an HTTP protocol error on the downstream request.
   bool downstream_protocol_error = 19;
+
+  // Indicates there was a max stream duration reached on the upstream request.
+  bool upstream_max_stream_duration_reached = 20;
 }
 
 // Properties of a negotiated TLS connection.

docs/root/configuration/observability/access_log/usage.rst

@@ -274,6 +274,7 @@ The following command operators are supported:
       :ref:`strictly-checked header <envoy_api_field_config.filter.http.router.v2.Router.strict_check_headers>` in addition to 400 response code.
     * **SI**: Stream idle timeout in addition to 408 response code.
     * **DPE**: The downstream request had an HTTP protocol error.
+    * **UMSDR**: The upstream request reached to max stream duration.
 
 %ROUTE_NAME%
   Name of the route.

docs/root/configuration/upstream/cluster_manager/cluster_stats.rst

@@ -69,6 +69,7 @@ Every cluster has a statistics tree rooted at *cluster.<name>.* with the followi
   upstream_rq_cancelled, Counter, Total requests cancelled before obtaining a connection pool connection
   upstream_rq_maintenance_mode, Counter, Total requests that resulted in an immediate 503 due to :ref:`maintenance mode<config_http_filters_router_runtime_maintenance_mode>`
   upstream_rq_timeout, Counter, Total requests that timed out waiting for a response
+  upstream_rq_max_duration_reached, Counter, Total requests closed due to max duration reached
   upstream_rq_per_try_timeout, Counter, Total requests that hit the per try timeout
   upstream_rq_rx_reset, Counter, Total requests that were reset remotely
   upstream_rq_tx_reset, Counter, Total requests that were reset locally

docs/root/faq/configuration/timeouts.rst

@@ -59,10 +59,7 @@ context request/stream is interchangeable.
   HTTP request/response streams periodically. You can't use :ref:`request_timeout 
   <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.request_timeout>`
   in this situation because this timer will be disarmed if a response header is received on the request/response streams.
-
-  .. attention::
-
-    The current implementation implements this timeout on downstream connections only.
+  This timeout is available on both upstream and downstream connections.
 
 Route timeouts
 ^^^^^^^^^^^^^^

include/envoy/stream_info/stream_info.h

@@ -72,8 +72,10 @@ enum ResponseFlag {
   InvalidEnvoyRequestHeaders = 0x20000,
   // Downstream request had an HTTP protocol error
   DownstreamProtocolError = 0x40000,
+  // Upstream request reached to user defined max stream duration.
+  UpstreamMaxStreamDurationReached = 0x80000,
   // ATTENTION: MAKE SURE THIS REMAINS EQUAL TO THE LAST FLAG.
-  LastFlag = DownstreamProtocolError
+  LastFlag = UpstreamMaxStreamDurationReached
 };
 
 /**
@@ -139,6 +141,8 @@ struct ResponseCodeDetailValues {
   const std::string UpstreamTimeout = "upstream_response_timeout";
   // The final upstream try timed out
   const std::string UpstreamPerTryTimeout = "upstream_per_try_timeout";
+  // The request was destroyed because of user defined max stream duration.
+  const std::string UpstreamMaxStreamDurationReached = "upstream_max_stream_duration_reached";
   // The upstream connection was reset before a response was started. This
   // will generally be accompanied by details about why the reset occurred.
   const std::string EarlyUpstreamReset = "upstream_reset_before_response_started";

include/envoy/upstream/upstream.h

@@ -571,6 +571,7 @@ class PrioritySet {
   COUNTER(upstream_rq_cancelled)                                                                   \
   COUNTER(upstream_rq_completed)                                                                   \
   COUNTER(upstream_rq_maintenance_mode)                                                            \
+  COUNTER(upstream_rq_max_duration_reached)                                                        \
   COUNTER(upstream_rq_pending_failure_eject)                                                       \
   COUNTER(upstream_rq_pending_overflow)                                                            \
   COUNTER(upstream_rq_pending_total)                                                               \
@@ -728,6 +729,12 @@ class ClusterInfo {
    */
   virtual const envoy::config::core::v3::Http2ProtocolOptions& http2Options() const PURE;
 
+  /**
+   * @return const envoy::config::core::v3::HttpProtocolOptions for all of HTTP versions.
+   */
+  virtual const envoy::config::core::v3::HttpProtocolOptions&
+  commonHttpProtocolOptions() const PURE;
+
   /**
    * @param name std::string containing the well-known name of the extension for which protocol
    *        options are desired

source/common/router/router.cc

@@ -984,6 +984,29 @@ void Filter::onPerTryTimeout(UpstreamRequest& upstream_request) {
                          StreamInfo::ResponseCodeDetails::get().UpstreamPerTryTimeout);
 }
 
+void Filter::onStreamMaxDurationReached(UpstreamRequest& upstream_request) {
+  upstream_request.resetStream();
+
+  if (maybeRetryReset(Http::StreamResetReason::LocalReset, upstream_request)) {
+    return;
+  }
+
+  upstream_request.removeFromList(upstream_requests_);
+  cleanup();
+
+  if (downstream_response_started_) {
+    callbacks_->streamInfo().setResponseCodeDetails(
+        StreamInfo::ResponseCodeDetails::get().UpstreamMaxStreamDurationReached);
+    callbacks_->resetStream();
+  } else {
+    callbacks_->streamInfo().setResponseFlag(
+        StreamInfo::ResponseFlag::UpstreamMaxStreamDurationReached);
+    callbacks_->sendLocalReply(
+        Http::Code::RequestTimeout, "upstream max stream duration reached", modify_headers_,
+        absl::nullopt, StreamInfo::ResponseCodeDetails::get().UpstreamMaxStreamDurationReached);
+  }
+}
+
 void Filter::updateOutlierDetection(Upstream::Outlier::Result result,
                                     UpstreamRequest& upstream_request,
                                     absl::optional<uint64_t> code) {

source/common/router/router.h

@@ -265,6 +265,7 @@ class RouterFilterInterface {
                                UpstreamRequest& upstream_request) PURE;
   virtual void onUpstreamHostSelected(Upstream::HostDescriptionConstSharedPtr host) PURE;
   virtual void onPerTryTimeout(UpstreamRequest& upstream_request) PURE;
+  virtual void onStreamMaxDurationReached(UpstreamRequest& upstream_request) PURE;
 
   virtual Http::StreamDecoderFilterCallbacks* callbacks() PURE;
   virtual Upstream::ClusterInfoConstSharedPtr cluster() PURE;
@@ -432,6 +433,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
                        UpstreamRequest& upstream_request) override;
   void onUpstreamHostSelected(Upstream::HostDescriptionConstSharedPtr host) override;
   void onPerTryTimeout(UpstreamRequest& upstream_request) override;
+  void onStreamMaxDurationReached(UpstreamRequest& upstream_request) override;
   Http::StreamDecoderFilterCallbacks* callbacks() override { return callbacks_; }
   Upstream::ClusterInfoConstSharedPtr cluster() override { return cluster_; }
   FilterConfig& config() override { return config_; }

source/common/router/upstream_request.cc

@@ -78,6 +78,9 @@ UpstreamRequest::~UpstreamRequest() {
     // Allows for testing.
     per_try_timeout_->disableTimer();
   }
+  if (max_stream_duration_timer_ != nullptr) {
+    max_stream_duration_timer_->disableTimer();
+  }
   clearRequestEncoder();
 
   // If desired, fire the per-try histogram when the UpstreamRequest
@@ -382,7 +385,18 @@ void UpstreamRequest::onPoolReady(
     paused_for_connect_ = true;
   }
 
+  if (upstream_host_->cluster().commonHttpProtocolOptions().has_max_stream_duration()) {
+    const auto max_stream_duration = std::chrono::milliseconds(DurationUtil::durationToMilliseconds(
+        upstream_host_->cluster().commonHttpProtocolOptions().max_stream_duration()));
+    if (max_stream_duration.count()) {
+      max_stream_duration_timer_ = parent_.callbacks()->dispatcher().createTimer(
+          [this]() -> void { onStreamMaxDurationReached(); });
+      max_stream_duration_timer_->enableTimer(max_stream_duration);
+    }
+  }
+
   upstream_->encodeHeaders(*parent_.downstreamHeaders(), shouldSendEndStream());
+
   calling_encode_headers_ = false;
 
   if (!paused_for_connect_) {
@@ -426,6 +440,13 @@ void UpstreamRequest::encodeBodyAndTrailers() {
   }
 }
 
+void UpstreamRequest::onStreamMaxDurationReached() {
+  upstream_host_->cluster().stats().upstream_rq_max_duration_reached_.inc();
+
+  // The upstream had closed then try to retry along with retry policy.
+  parent_.onStreamMaxDurationReached(*this);
+}
+
 void UpstreamRequest::clearRequestEncoder() {
   // Before clearing the encoder, unsubscribe from callbacks.
   if (upstream_) {

source/common/router/upstream_request.h

@@ -107,6 +107,7 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
   UpstreamRequest* upstreamRequest() override { return this; }
 
   void clearRequestEncoder();
+  void onStreamMaxDurationReached();
 
   struct DownstreamWatermarkManager : public Http::DownstreamWatermarkCallbacks {
     DownstreamWatermarkManager(UpstreamRequest& parent) : parent_(parent) {}
@@ -188,6 +189,8 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
   // Sentinel to indicate if timeout budget tracking is configured for the cluster,
   // and if so, if the per-try histogram should record a value.
   bool record_timeout_budget_ : 1;
+
+  Event::TimerPtr max_stream_duration_timer_;
 };
 
 class HttpConnPool : public GenericConnPool, public Http::ConnectionPool::Callbacks {

source/common/stream_info/utility.cc

@@ -25,6 +25,7 @@ const std::string ResponseFlagUtils::RATELIMIT_SERVICE_ERROR = "RLSE";
 const std::string ResponseFlagUtils::STREAM_IDLE_TIMEOUT = "SI";
 const std::string ResponseFlagUtils::INVALID_ENVOY_REQUEST_HEADERS = "IH";
 const std::string ResponseFlagUtils::DOWNSTREAM_PROTOCOL_ERROR = "DPE";
+const std::string ResponseFlagUtils::UPSTREAM_MAX_STREAM_DURATION_REACHED = "UMSDR";
 
 void ResponseFlagUtils::appendString(std::string& result, const std::string& append) {
   if (result.empty()) {
@@ -37,7 +38,7 @@ void ResponseFlagUtils::appendString(std::string& result, const std::string& app
 const std::string ResponseFlagUtils::toShortString(const StreamInfo& stream_info) {
   std::string result;
 
-  static_assert(ResponseFlag::LastFlag == 0x40000, "A flag has been added. Fix this code.");
+  static_assert(ResponseFlag::LastFlag == 0x80000, "A flag has been added. Fix this code.");
 
   if (stream_info.hasResponseFlag(ResponseFlag::FailedLocalHealthCheck)) {
     appendString(result, FAILED_LOCAL_HEALTH_CHECK);
@@ -114,6 +115,9 @@ const std::string ResponseFlagUtils::toShortString(const StreamInfo& stream_info
     appendString(result, DOWNSTREAM_PROTOCOL_ERROR);
   }
 
+  if (stream_info.hasResponseFlag(ResponseFlag::UpstreamMaxStreamDurationReached)) {
+    appendString(result, UPSTREAM_MAX_STREAM_DURATION_REACHED);
+  }
   return result.empty() ? NONE : result;
 }
 
@@ -140,6 +144,8 @@ absl::optional<ResponseFlag> ResponseFlagUtils::toResponseFlag(const std::string
       {ResponseFlagUtils::STREAM_IDLE_TIMEOUT, ResponseFlag::StreamIdleTimeout},
       {ResponseFlagUtils::INVALID_ENVOY_REQUEST_HEADERS, ResponseFlag::InvalidEnvoyRequestHeaders},
       {ResponseFlagUtils::DOWNSTREAM_PROTOCOL_ERROR, ResponseFlag::DownstreamProtocolError},
+      {ResponseFlagUtils::UPSTREAM_MAX_STREAM_DURATION_REACHED,
+       ResponseFlag::UpstreamMaxStreamDurationReached},
   };
   const auto& it = map.find(flag);
   if (it != map.end()) {

source/common/stream_info/utility.h

@@ -40,6 +40,7 @@ class ResponseFlagUtils {
   const static std::string STREAM_IDLE_TIMEOUT;
   const static std::string INVALID_ENVOY_REQUEST_HEADERS;
   const static std::string DOWNSTREAM_PROTOCOL_ERROR;
+  const static std::string UPSTREAM_MAX_STREAM_DURATION_REACHED;
 };
 
 /**

source/common/upstream/upstream_impl.cc

@@ -685,6 +685,7 @@ ClusterInfoImpl::ClusterInfoImpl(
       features_(parseFeatures(config)),
       http1_settings_(Http::Utility::parseHttp1Settings(config.http_protocol_options())),
       http2_options_(Http2::Utility::initializeAndValidateOptions(config.http2_protocol_options())),
+      common_http_protocol_options_(config.common_http_protocol_options()),
       extension_protocol_options_(parseExtensionProtocolOptions(config, validation_visitor)),
       resource_managers_(config, runtime, name_, *stats_scope_),
       maintenance_mode_runtime_key_(absl::StrCat("upstream.maintenance_mode.", name_)),

source/common/upstream/upstream_impl.h

@@ -539,6 +539,9 @@ class ClusterInfoImpl : public ClusterInfo, protected Logger::Loggable<Logger::I
   const envoy::config::core::v3::Http2ProtocolOptions& http2Options() const override {
     return http2_options_;
   }
+  const envoy::config::core::v3::HttpProtocolOptions& commonHttpProtocolOptions() const override {
+    return common_http_protocol_options_;
+  }
   ProtocolOptionsConfigConstSharedPtr
   extensionProtocolOptions(const std::string& name) const override;
   LoadBalancerType lbType() const override { return lb_type_; }
@@ -626,6 +629,7 @@ class ClusterInfoImpl : public ClusterInfo, protected Logger::Loggable<Logger::I
   const uint64_t features_;
   const Http::Http1Settings http1_settings_;
   const envoy::config::core::v3::Http2ProtocolOptions http2_options_;
+  const envoy::config::core::v3::HttpProtocolOptions common_http_protocol_options_;
   const std::map<std::string, ProtocolOptionsConfigConstSharedPtr> extension_protocol_options_;
   mutable ResourceManagers resource_managers_;
   const std::string maintenance_mode_runtime_key_;

source/extensions/access_loggers/grpc/grpc_access_log_utils.cc

@@ -37,7 +37,7 @@ void Utility::responseFlagsToAccessLogResponseFlags(
     envoy::data::accesslog::v3::AccessLogCommon& common_access_log,
     const StreamInfo::StreamInfo& stream_info) {
 
-  static_assert(StreamInfo::ResponseFlag::LastFlag == 0x40000,
+  static_assert(StreamInfo::ResponseFlag::LastFlag == 0x80000,
                 "A flag has been added. Fix this code.");
 
   if (stream_info.hasResponseFlag(StreamInfo::ResponseFlag::FailedLocalHealthCheck)) {
@@ -116,6 +116,9 @@ void Utility::responseFlagsToAccessLogResponseFlags(
   if (stream_info.hasResponseFlag(StreamInfo::ResponseFlag::DownstreamProtocolError)) {
     common_access_log.mutable_response_flags()->set_downstream_protocol_error(true);
   }
+  if (stream_info.hasResponseFlag(StreamInfo::ResponseFlag::UpstreamMaxStreamDurationReached)) {
+    common_access_log.mutable_response_flags()->set_upstream_max_stream_duration_reached(true);
+  }
 }
 
 void Utility::extractCommonAccessLogProperties(