http: max stream duration upstream support

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

@@ -64,8 +64,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;
 }
 

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

@@ -71,8 +71,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;
 }
 

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

include/envoy/upstream/upstream.h

@@ -570,6 +570,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)                                                               \
@@ -699,6 +700,11 @@ class ClusterInfo {
    */
   virtual std::chrono::milliseconds connectTimeout() const PURE;
 
+  /**
+   * @return maximum duration time to keep alive stream
+   */
+  virtual absl::optional<std::chrono::milliseconds> maxStreamDuration() const PURE;
+
   /**
    * @return the idle timeout for upstream connection pool connections.
    */

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
@@ -360,7 +363,15 @@ void UpstreamRequest::onPoolReady(
   // If end_stream is set in headers, and there are metadata to send, delays end_stream. The case
   // only happens when decoding headers filters return ContinueAndEndStream.
   const bool delay_headers_end_stream = end_stream && !downstream_metadata_map_vector_.empty();
+
+  const auto max_stream_duration = upstream_host_->cluster().maxStreamDuration();
+  if (max_stream_duration.has_value() && max_stream_duration.value().count()) {
+    max_stream_duration_timer_ = parent_.callbacks()->dispatcher().createTimer(
+        [this]() -> void { onStreamMaxDurationReached(); });
+    max_stream_duration_timer_->enableTimer(upstream_host_->cluster().maxStreamDuration().value());
+  }
   upstream_->encodeHeaders(*parent_.downstreamHeaders(), end_stream && !delay_headers_end_stream);
+
   calling_encode_headers_ = false;
 
   // It is possible to get reset in the middle of an encodeHeaders() call. This happens for
@@ -398,6 +409,14 @@ void UpstreamRequest::onPoolReady(
   }
 }
 
+void UpstreamRequest::onStreamMaxDurationReached() {
+  ASSERT(upstream_host_);
+  upstream_host_->cluster().stats().upstream_rq_max_duration_reached_.inc();
+  max_stream_duration_timer_->disableTimer();
+  max_stream_duration_timer_.reset();
+  resetStream();
+}
+
 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) {}
@@ -174,6 +175,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/upstream/upstream_impl.cc

@@ -667,6 +667,8 @@ ClusterInfoImpl::ClusterInfoImpl(
                                          Http::DEFAULT_MAX_HEADERS_COUNT))),
       connect_timeout_(
           std::chrono::milliseconds(PROTOBUF_GET_MS_REQUIRED(config, connect_timeout))),
+      max_stream_duration_(
+          PROTOBUF_GET_OPTIONAL_MS(config.common_http_protocol_options(), max_stream_duration)),
       per_connection_buffer_limit_bytes_(
           PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, per_connection_buffer_limit_bytes, 1024 * 1024)),
       socket_matcher_(std::move(socket_matcher)), stats_scope_(std::move(stats_scope)),

source/common/upstream/upstream_impl.h

@@ -529,6 +529,9 @@ class ClusterInfoImpl : public ClusterInfo, protected Logger::Loggable<Logger::I
     return common_lb_config_;
   }
   std::chrono::milliseconds connectTimeout() const override { return connect_timeout_; }
+  absl::optional<std::chrono::milliseconds> maxStreamDuration() const override {
+    return max_stream_duration_;
+  }
   const absl::optional<std::chrono::milliseconds> idleTimeout() const override {
     return idle_timeout_;
   }
@@ -616,6 +619,7 @@ class ClusterInfoImpl : public ClusterInfo, protected Logger::Loggable<Logger::I
   const uint64_t max_requests_per_connection_;
   const uint32_t max_response_headers_count_;
   const std::chrono::milliseconds connect_timeout_;
+  const absl::optional<std::chrono::milliseconds> max_stream_duration_;
   absl::optional<std::chrono::milliseconds> idle_timeout_;
   const uint32_t per_connection_buffer_limit_bytes_;
   TransportSocketMatcherPtr socket_matcher_;

test/common/router/router_test.cc

@@ -125,6 +125,12 @@ class RouterTestBase : public testing::Test {
     EXPECT_CALL(*per_try_timeout_, disableTimer());
   }
 
+  void expectMaxStreamDurationTimer() {
+    max_stream_duration_timer_ = new Event::MockTimer(&callbacks_.dispatcher_);
+    EXPECT_CALL(*max_stream_duration_timer_, enableTimer(_, _));
+    EXPECT_CALL(*max_stream_duration_timer_, disableTimer());
+  }
+
   AssertionResult verifyHostUpstreamStats(uint64_t success, uint64_t error) {
     if (success != cm_.conn_pool_.host_->stats_.rq_success_.value()) {
       return AssertionFailure() << fmt::format("rq_success {} does not match expected {}",
@@ -348,6 +354,7 @@ class RouterTestBase : public testing::Test {
   RouterTestFilter router_;
   Event::MockTimer* response_timeout_{};
   Event::MockTimer* per_try_timeout_{};
+  Event::MockTimer* max_stream_duration_timer_{};
   Network::Address::InstanceConstSharedPtr host_address_{
       Network::Utility::resolveUrl("tcp://10.0.0.5:9211")};
   NiceMock<Http::MockRequestEncoder> original_encoder_;
@@ -3551,6 +3558,79 @@ TEST_F(RouterTest, RetryTimeoutDuringRetryDelay) {
   EXPECT_TRUE(verifyHostUpstreamStats(0, 1));
 }
 
+TEST_F(RouterTest, MaxStreamDurationValidlyConfigured) {
+  NiceMock<Http::MockRequestEncoder> encoder1;
+  Http::ResponseDecoder* response_decoder = nullptr;
+  ON_CALL(cm_.conn_pool_.host_->cluster_, maxStreamDuration())
+      .WillByDefault(Return(std::chrono::milliseconds(500)));
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke(
+          [&](Http::ResponseDecoder& decoder,
+              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
+            response_decoder = &decoder;
+            callbacks.onPoolReady(encoder1, cm_.conn_pool_.host_, upstream_stream_info_);
+            return nullptr;
+          }));
+  expectMaxStreamDurationTimer();
+
+  Http::TestRequestHeaderMapImpl headers;
+  HttpTestUtility::addDefaultHeaders(headers);
+  router_.decodeHeaders(headers, false);
+  max_stream_duration_timer_->invokeCallback();
+
+  router_.onDestroy();
+  EXPECT_TRUE(verifyHostUpstreamStats(0, 0));
+}
+
+TEST_F(RouterTest, MaxStreamDurationDisabledIfSetToZero) {
+  NiceMock<Http::MockRequestEncoder> encoder1;
+  Http::ResponseDecoder* response_decoder = nullptr;
+  ON_CALL(cm_.conn_pool_.host_->cluster_, maxStreamDuration())
+      .WillByDefault(Return(std::chrono::milliseconds(0)));
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke(
+          [&](Http::ResponseDecoder& decoder,
+              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
+            response_decoder = &decoder;
+            callbacks.onPoolReady(encoder1, cm_.conn_pool_.host_, upstream_stream_info_);
+            return nullptr;
+          }));
+
+  // not to be called timer creation.
+  EXPECT_CALL(callbacks_.dispatcher_, createTimer_).Times(0);
+
+  Http::TestRequestHeaderMapImpl headers;
+  HttpTestUtility::addDefaultHeaders(headers);
+  router_.decodeHeaders(headers, false);
+
+  router_.onDestroy();
+  EXPECT_TRUE(verifyHostUpstreamStats(0, 0));
+}
+
+TEST_F(RouterTest, MaxStreamDurationCallbackNotCalled) {
+  NiceMock<Http::MockRequestEncoder> encoder1;
+  Http::ResponseDecoder* response_decoder = nullptr;
+  ON_CALL(cm_.conn_pool_.host_->cluster_, maxStreamDuration())
+      .WillByDefault(Return(std::chrono::milliseconds(5000)));
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke(
+          [&](Http::ResponseDecoder& decoder,
+              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
+            response_decoder = &decoder;
+            callbacks.onPoolReady(encoder1, cm_.conn_pool_.host_, upstream_stream_info_);
+            return nullptr;
+          }));
+
+  expectMaxStreamDurationTimer();
+
+  Http::TestRequestHeaderMapImpl headers;
+  HttpTestUtility::addDefaultHeaders(headers);
+  router_.decodeHeaders(headers, false);
+
+  router_.onDestroy();
+  EXPECT_TRUE(verifyHostUpstreamStats(0, 0));
+}
+
 TEST_F(RouterTest, RetryTimeoutDuringRetryDelayWithUpstreamRequestNoHost) {
   NiceMock<Http::MockRequestEncoder> encoder1;
   Http::ResponseDecoder* response_decoder = nullptr;

test/common/tcp_proxy/upstream_test.cc

@@ -26,6 +26,7 @@ class HttpUpstreamTest : public testing::Test {
   Http::MockRequestEncoder encoder_;
   NiceMock<Tcp::ConnectionPool::MockUpstreamCallbacks> callbacks_;
   std::unique_ptr<HttpUpstream> upstream_;
+  testing::NiceMock<Event::MockDispatcher> dispatcher_;
   std::string hostname_{"default.host.com"};
 };
 

test/integration/http2_integration_test.cc

@@ -70,6 +70,8 @@ TEST_P(Http2IntegrationTest, RetryAttemptCount) { testRetryAttemptCountHeader();
 
 TEST_P(Http2IntegrationTest, LargeRequestTrailersRejected) { testLargeRequestTrailers(66, 60); }
 
+TEST_P(Http2IntegrationTest, BasicMaxStreamDuration) { testMaxStreamDuration(); }
+
 static std::string response_metadata_filter = R"EOF(
 name: response-metadata-filter
 typed_config:

test/integration/http_integration.cc

@@ -1204,6 +1204,29 @@ void HttpIntegrationTest::testAdminDrain(Http::CodecClient::Type admin_request_t
       nullptr, true));
 }
 
+void HttpIntegrationTest::testMaxStreamDuration() {
+  config_helper_.addConfigModifier([](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+    auto* static_resources = bootstrap.mutable_static_resources();
+    auto* cluster = static_resources->mutable_clusters(0);
+    auto* http_protocol_options = cluster->mutable_common_http_protocol_options();
+    http_protocol_options->mutable_max_stream_duration()->MergeFrom(
+        ProtobufUtil::TimeUtil::MillisecondsToDuration(200));
+  });
+
+  initialize();
+  fake_upstreams_[0]->set_allow_unexpected_disconnects(true);
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  auto encoder_decoder = codec_client_->startRequest(default_request_headers_);
+  request_encoder_ = &encoder_decoder.first;
+  auto response = std::move(encoder_decoder.second);
+
+  ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_));
+  ASSERT_TRUE(fake_upstream_connection_->waitForNewStream(*dispatcher_, upstream_request_));
+
+  test_server_->waitForCounterGe("cluster.cluster_0.upstream_rq_max_duration_reached", 1);
+}
+
 std::string HttpIntegrationTest::listenerStatPrefix(const std::string& stat_name) {
   if (version_ == Network::Address::IpVersion::v4) {
     return "listener.127.0.0.1_0." + stat_name;

test/integration/http_integration.h

@@ -220,7 +220,8 @@ class HttpIntegrationTest : public BaseIntegrationTest {
                     bool response_trailers_present);
   // Test /drain_listener from admin portal.
   void testAdminDrain(Http::CodecClient::Type admin_request_type);
-
+  // test max stream duration
+  void testMaxStreamDuration();
   Http::CodecClient::Type downstreamProtocol() const { return downstream_protocol_; }
   // Prefix listener stat with IP:port, including IP version dependent loopback address.
   std::string listenerStatPrefix(const std::string& stat_name);

test/mocks/upstream/cluster_info.h

@@ -85,6 +85,7 @@ class MockClusterInfo : public ClusterInfo {
   // Upstream::ClusterInfo
   MOCK_METHOD(bool, addedViaApi, (), (const));
   MOCK_METHOD(std::chrono::milliseconds, connectTimeout, (), (const));
+  MOCK_METHOD(absl::optional<std::chrono::milliseconds>, maxStreamDuration, (), (const));
   MOCK_METHOD(const absl::optional<std::chrono::milliseconds>, idleTimeout, (), (const));
   MOCK_METHOD(uint32_t, perConnectionBufferLimitBytes, (), (const));
   MOCK_METHOD(uint64_t, features, (), (const));
@@ -158,6 +159,7 @@ class MockClusterInfo : public ClusterInfo {
   envoy::config::cluster::v3::Cluster::CommonLbConfig lb_config_;
   envoy::config::core::v3::Metadata metadata_;
   std::unique_ptr<Envoy::Config::TypedMetadata> typed_metadata_;
+  absl::optional<std::chrono::milliseconds> max_stream_duration_;
 };
 
 class MockIdleTimeEnabledClusterInfo : public MockClusterInfo {