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

@@ -579,6 +579,7 @@ class PrioritySet {
   COUNTER(upstream_rq_retry_success)                                                               \
   COUNTER(upstream_rq_rx_reset)                                                                    \
   COUNTER(upstream_rq_timeout)                                                                     \
+  COUNTER(upstream_rq_max_duration_reached)                                                        \
   COUNTER(upstream_rq_total)                                                                       \
   COUNTER(upstream_rq_tx_reset)                                                                    \
   GAUGE(lb_subsets_active, Accumulate)                                                             \
@@ -698,6 +699,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/http/codec_client.cc

@@ -20,7 +20,7 @@ namespace Http {
 CodecClient::CodecClient(Type type, Network::ClientConnectionPtr&& connection,
                          Upstream::HostDescriptionConstSharedPtr host,
                          Event::Dispatcher& dispatcher)
-    : type_(type), connection_(std::move(connection)), host_(host),
+    : dispatcher_(dispatcher), type_(type), connection_(std::move(connection)), host_(host),
       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

@@ -120,6 +120,8 @@ class CodecClient : Logger::Loggable<Logger::Id::client>,
 
   const StreamInfo::StreamInfo& streamInfo() { return connection_->streamInfo(); }
 
+  Event::Dispatcher& dispatcher_;
+
 protected:
   /**
    * Create a codec client and connect to a remote host/port.
@@ -188,7 +190,14 @@ class CodecClient : Logger::Loggable<Logger::Id::client>,
                          public StreamCallbacks,
                          public ResponseDecoderWrapper {
     ActiveRequest(CodecClient& parent, ResponseDecoder& inner)
-        : ResponseDecoderWrapper(inner), parent_(parent) {}
+        : ResponseDecoderWrapper(inner), parent_(parent),
+          max_stream_duration_(parent_.host_->cluster().maxStreamDuration()) {
+      if (max_stream_duration_.has_value()) {
+        max_stream_duration_timer_ =
+            parent.dispatcher_.createTimer([this]() -> void { onStreamMaxDurationReached(); });
+        max_stream_duration_timer_->enableTimer(max_stream_duration_.value());
+      }
+    }
 
     // StreamCallbacks
     void onResetStream(StreamResetReason reason, absl::string_view) override {
@@ -201,8 +210,17 @@ class CodecClient : Logger::Loggable<Logger::Id::client>,
     void onPreDecodeComplete() override { parent_.responseDecodeComplete(*this); }
     void onDecodeComplete() override {}
 
+    void onStreamMaxDurationReached() {
+      parent_.host_->cluster().stats().upstream_rq_max_duration_reached_.inc();
+      max_stream_duration_timer_->disableTimer();
+      max_stream_duration_timer_.reset();
+      onResetStream(StreamResetReason::ConnectionTermination, "");
+    }
+
     RequestEncoder* encoder_{};
     CodecClient& parent_;
+    Event::TimerPtr max_stream_duration_timer_;
+    const absl::optional<std::chrono::milliseconds> max_stream_duration_;
   };
 
   using ActiveRequestPtr = std::unique_ptr<ActiveRequest>;

source/common/upstream/upstream_impl.cc

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

@@ -527,6 +527,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_;
   }
@@ -614,6 +617,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/http/codec_client_test.cc

@@ -39,9 +39,9 @@ namespace Envoy {
 namespace Http {
 namespace {
 
-class CodecClientTest : public testing::Test {
+template <class ClusterInfoType> class CodecClientTestBase : public testing::Test {
 public:
-  CodecClientTest() {
+  CodecClientTestBase() {
     connection_ = new NiceMock<Network::MockClientConnection>();
 
     EXPECT_CALL(*connection_, detectEarlyCloseWhenReadDisabled(false));
@@ -52,29 +52,33 @@ class CodecClientTest : public testing::Test {
             Invoke([this](Network::ReadFilterSharedPtr filter) -> void { filter_ = filter; }));
 
     codec_ = new Http::MockClientConnection();
-
-    Network::ClientConnectionPtr connection{connection_};
-    EXPECT_CALL(dispatcher_, createTimer_(_));
-    client_ = std::make_unique<CodecClientForTest>(CodecClient::Type::HTTP1, std::move(connection),
-                                                   codec_, nullptr, host_, dispatcher_);
-    ON_CALL(*connection_, streamInfo()).WillByDefault(ReturnRef(stream_info_));
   }
 
-  ~CodecClientTest() override { EXPECT_EQ(0U, client_->numActiveRequests()); }
+  ~CodecClientTestBase() override { EXPECT_EQ(0U, client_->numActiveRequests()); }
 
   Event::MockDispatcher dispatcher_;
   Network::MockClientConnection* connection_;
   Http::MockClientConnection* codec_;
   std::unique_ptr<CodecClientForTest> client_;
   Network::ConnectionCallbacks* connection_cb_;
   Network::ReadFilterSharedPtr filter_;
-  std::shared_ptr<Upstream::MockIdleTimeEnabledClusterInfo> cluster_{
-      new NiceMock<Upstream::MockIdleTimeEnabledClusterInfo>()};
+  std::shared_ptr<ClusterInfoType> cluster_{new NiceMock<ClusterInfoType>()};
   Upstream::HostDescriptionConstSharedPtr host_{
       Upstream::makeTestHostDescription(cluster_, "tcp://127.0.0.1:80")};
   NiceMock<StreamInfo::MockStreamInfo> stream_info_;
 };
 
+class CodecClientTest : public CodecClientTestBase<Upstream::MockIdleTimeEnabledClusterInfo> {
+public:
+  CodecClientTest() : CodecClientTestBase<Upstream::MockIdleTimeEnabledClusterInfo>() {
+    Network::ClientConnectionPtr connection{connection_};
+    EXPECT_CALL(dispatcher_, createTimer_(_));
+    client_ = std::make_unique<CodecClientForTest>(CodecClient::Type::HTTP1, std::move(connection),
+                                                   codec_, nullptr, host_, dispatcher_);
+    ON_CALL(*connection_, streamInfo()).WillByDefault(ReturnRef(stream_info_));
+  }
+};
+
 TEST_F(CodecClientTest, NotCallDetectEarlyCloseWhenReadDiabledUsingHttp3) {
   auto connection = std::make_unique<NiceMock<Network::MockClientConnection>>();
 
@@ -282,6 +286,36 @@ TEST_F(CodecClientTest, SSLConnectionInfo) {
   EXPECT_EQ(session_id, stream_info_.downstreamSslConnection()->sessionId());
 }
 
+class CodecMaxStreamDurationTest
+    : public CodecClientTestBase<Upstream::MockMaxStreamDurationEnabledClusterInfo> {
+public:
+  CodecMaxStreamDurationTest()
+      : CodecClientTestBase<Upstream::MockMaxStreamDurationEnabledClusterInfo>() {
+    Network::ClientConnectionPtr connection{connection_};
+    client_ = std::make_unique<CodecClientForTest>(CodecClient::Type::HTTP1, std::move(connection),
+                                                   codec_, nullptr, host_, dispatcher_);
+    ON_CALL(*connection_, streamInfo()).WillByDefault(ReturnRef(stream_info_));
+  }
+};
+
+TEST_F(CodecMaxStreamDurationTest, Basic) {
+  ResponseDecoder* inner_decoder;
+  NiceMock<MockRequestEncoder> inner_encoder;
+  EXPECT_CALL(*codec_, newStream(_))
+      .WillOnce(Invoke([&](ResponseDecoder& decoder) -> RequestEncoder& {
+        inner_decoder = &decoder;
+        return inner_encoder;
+      }));
+
+  Event::MockTimer* timer = new Event::MockTimer(&dispatcher_);
+  Http::MockResponseDecoder outer_decoder;
+  EXPECT_CALL(*timer, enableTimer(cluster_->maxStreamDuration().value(), _));
+  client_->newStream(outer_decoder);
+  EXPECT_CALL(*timer, disableTimer());
+  timer->invokeCallback();
+  EXPECT_EQ(1U, cluster_->stats_.upstream_rq_max_duration_reached_.value());
+}
+
 // Test the codec getting input from a real TCP connection.
 class CodecNetworkTest : public testing::TestWithParam<Network::Address::IpVersion> {
 public:

test/mocks/upstream/cluster_info.cc

@@ -36,6 +36,13 @@ MockIdleTimeEnabledClusterInfo::MockIdleTimeEnabledClusterInfo() {
 
 MockIdleTimeEnabledClusterInfo::~MockIdleTimeEnabledClusterInfo() = default;
 
+MockMaxStreamDurationEnabledClusterInfo::MockMaxStreamDurationEnabledClusterInfo() {
+  ON_CALL(*this, maxStreamDuration())
+      .WillByDefault(Return(absl::optional<std::chrono::milliseconds>(50)));
+}
+
+MockMaxStreamDurationEnabledClusterInfo::~MockMaxStreamDurationEnabledClusterInfo() = default;
+
 MockClusterInfo::MockClusterInfo()
     : http2_options_(::Envoy::Http2::Utility::initializeAndValidateOptions(
           envoy::config::core::v3::Http2ProtocolOptions())),

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 {
@@ -166,5 +168,11 @@ class MockIdleTimeEnabledClusterInfo : public MockClusterInfo {
   ~MockIdleTimeEnabledClusterInfo() override;
 };
 
+class MockMaxStreamDurationEnabledClusterInfo : public MockClusterInfo {
+public:
+  MockMaxStreamDurationEnabledClusterInfo();
+  ~MockMaxStreamDurationEnabledClusterInfo() override;
+};
+
 } // namespace Upstream
 } // namespace Envoy