conn_pool: support max stream duration for upstream connections

include/envoy/upstream/upstream.h

@@ -555,6 +555,7 @@ class PrioritySet {
   COUNTER(upstream_cx_http2_total)                                                                 \
   COUNTER(upstream_cx_idle_timeout)                                                                \
   COUNTER(upstream_cx_max_requests)                                                                \
+  COUNTER(upstream_cx_max_duration)                                                                \
   COUNTER(upstream_cx_none_healthy)                                                                \
   COUNTER(upstream_cx_overflow)                                                                    \
   COUNTER(upstream_cx_pool_overflow)                                                               \
@@ -706,6 +707,11 @@ class ClusterInfo {
    */
   virtual const absl::optional<std::chrono::milliseconds> idleTimeout() const PURE;
 
+  /**
+   * @return the max duration for upstream connection pool connections.
+   */
+  virtual const absl::optional<std::chrono::milliseconds> maxConnectionDuration() const PURE;
+
   /**
    * @return soft limit on size of the cluster's connections read and write buffers.
    */

source/common/conn_pool/conn_pool_base.cc

@@ -391,7 +391,11 @@ ActiveClient::ActiveClient(ConnPoolImplBase& parent, uint64_t lifetime_request_l
   conn_length_ = std::make_unique<Stats::HistogramCompletableTimespanImpl>(
       parent_.host()->cluster().stats().upstream_cx_length_ms_, parent_.dispatcher().timeSource());
   connect_timer_->enableTimer(parent_.host()->cluster().connectTimeout());
-
+  const auto max_connection_duration = parent_.host()->cluster().maxConnectionDuration();
+  if (max_connection_duration) {
+    lifetime_timer_ = parent_.dispatcher().createTimer([this]() -> void { onLifetimeTimeout(); });
+    lifetime_timer_->enableTimer(max_connection_duration.value());
+  }
   parent_.host()->stats().cx_total_.inc();
   parent_.host()->stats().cx_active_.inc();
   parent_.host()->cluster().stats().upstream_cx_total_.inc();
@@ -424,5 +428,14 @@ void ActiveClient::onConnectTimeout() {
   close();
 }
 
+void ActiveClient::onLifetimeTimeout() {
+  if (state_ != ActiveClient::State::CLOSED) {
+    ENVOY_CONN_LOG(debug, "lifetime timeout, DRAINING", *this);
+    parent_.host()->cluster().stats().upstream_cx_max_duration_.inc();
+    parent_.transitionActiveClientState(*this,
+                                        Envoy::ConnectionPool::ActiveClient::State::DRAINING);
+  }
+}
+
 } // namespace ConnectionPool
 } // namespace Envoy

source/common/conn_pool/conn_pool_base.h

@@ -42,6 +42,10 @@ class ActiveClient : public LinkedObject<ActiveClient>,
   // Called if the connection does not complete within the cluster's connectTimeout()
   void onConnectTimeout();
 
+  // Called if the maximum connection duration is reached. If set, this puts an upper
+  // bound on the lifetime of any connection.
+  void onLifetimeTimeout();
+
   // Returns the concurrent request limit, accounting for if the total request limit
   // is less than the concurrent request limit.
   uint64_t effectiveConcurrentRequestLimit() const {
@@ -74,6 +78,7 @@ class ActiveClient : public LinkedObject<ActiveClient>,
   Stats::TimespanPtr conn_connect_ms_;
   Stats::TimespanPtr conn_length_;
   Event::TimerPtr connect_timer_;
+  Event::TimerPtr lifetime_timer_;
   bool resources_released_{false};
   bool timed_out_{false};
 };

source/common/upstream/upstream_impl.cc

@@ -789,6 +789,16 @@ ClusterInfoImpl::ClusterInfoImpl(
     idle_timeout_ = std::chrono::hours(1);
   }
 
+  if (config.common_http_protocol_options().has_max_connection_duration()) {
+    max_connection_duration_ = std::chrono::milliseconds(DurationUtil::durationToMilliseconds(
+        config.common_http_protocol_options().max_connection_duration()));
+    if (max_connection_duration_.value().count() == 0) {
+      max_connection_duration_ = absl::nullopt;
+    }
+  } else {
+    max_connection_duration_ = absl::nullopt;
+  }
+
   if (config.has_eds_cluster_config()) {
     if (config.type() != envoy::config::cluster::v3::Cluster::EDS) {
       throw EnvoyException("eds_cluster_config set in a non-EDS cluster");

source/common/upstream/upstream_impl.h

@@ -534,6 +534,9 @@ class ClusterInfoImpl : public ClusterInfo, protected Logger::Loggable<Logger::I
   const absl::optional<std::chrono::milliseconds> idleTimeout() const override {
     return idle_timeout_;
   }
+  const absl::optional<std::chrono::milliseconds> maxConnectionDuration() const override {
+    return max_connection_duration_;
+  }
   uint32_t perConnectionBufferLimitBytes() const override {
     return per_connection_buffer_limit_bytes_;
   }
@@ -629,6 +632,7 @@ class ClusterInfoImpl : public ClusterInfo, protected Logger::Loggable<Logger::I
   const uint32_t max_response_headers_count_;
   const std::chrono::milliseconds connect_timeout_;
   absl::optional<std::chrono::milliseconds> idle_timeout_;
+  absl::optional<std::chrono::milliseconds> max_connection_duration_;
   const uint32_t per_connection_buffer_limit_bytes_;
   TransportSocketMatcherPtr socket_matcher_;
   Stats::ScopePtr stats_scope_;

test/common/http/http1/conn_pool_test.cc

@@ -69,6 +69,7 @@ class ConnPoolImplForTest : public ConnPoolImpl {
     Network::MockClientConnection* connection_;
     CodecClient* codec_client_;
     Event::MockTimer* connect_timer_;
+    Event::MockTimer* lifetime_timer_;
     Event::DispatcherPtr client_dispatcher_;
   };
 

test/mocks/upstream/cluster_info.cc

@@ -36,6 +36,13 @@ MockIdleTimeEnabledClusterInfo::MockIdleTimeEnabledClusterInfo() {
 
 MockIdleTimeEnabledClusterInfo::~MockIdleTimeEnabledClusterInfo() = default;
 
+MockMaxConnectionDurationEnabledClusterInfo::MockMaxConnectionDurationEnabledClusterInfo() {
+  ON_CALL(*this, maxConnectionDuration()).WillByDefault(Return(std::chrono::milliseconds(1000)));
+}
+
+MockMaxConnectionDurationEnabledClusterInfo::~MockMaxConnectionDurationEnabledClusterInfo() =
+    default;
+
 MockClusterInfo::MockClusterInfo()
     : http2_options_(::Envoy::Http2::Utility::initializeAndValidateOptions(
           envoy::config::core::v3::Http2ProtocolOptions())),
@@ -51,6 +58,8 @@ MockClusterInfo::MockClusterInfo()
           circuit_breakers_stats_, absl::nullopt, absl::nullopt)) {
   ON_CALL(*this, connectTimeout()).WillByDefault(Return(std::chrono::milliseconds(1)));
   ON_CALL(*this, idleTimeout()).WillByDefault(Return(absl::optional<std::chrono::milliseconds>()));
+  ON_CALL(*this, maxConnectionDuration())
+      .WillByDefault(Return(absl::optional<std::chrono::milliseconds>()));
   ON_CALL(*this, name()).WillByDefault(ReturnRef(name_));
   ON_CALL(*this, edsServiceName()).WillByDefault(ReturnPointee(&eds_service_name_));
   ON_CALL(*this, http1Settings()).WillByDefault(ReturnRef(http1_settings_));

test/mocks/upstream/cluster_info.h

@@ -89,6 +89,7 @@ class MockClusterInfo : public ClusterInfo {
   MOCK_METHOD(bool, addedViaApi, (), (const));
   MOCK_METHOD(std::chrono::milliseconds, connectTimeout, (), (const));
   MOCK_METHOD(const absl::optional<std::chrono::milliseconds>, idleTimeout, (), (const));
+  MOCK_METHOD(const absl::optional<std::chrono::milliseconds>, maxConnectionDuration, (), (const));
   MOCK_METHOD(uint32_t, perConnectionBufferLimitBytes, (), (const));
   MOCK_METHOD(uint64_t, features, (), (const));
   MOCK_METHOD(const Http::Http1Settings&, http1Settings, (), (const));
@@ -181,5 +182,11 @@ class MockIdleTimeEnabledClusterInfo : public MockClusterInfo {
   ~MockIdleTimeEnabledClusterInfo() override;
 };
 
+class MockMaxConnectionDurationEnabledClusterInfo : public MockClusterInfo {
+public:
+  MockMaxConnectionDurationEnabledClusterInfo();
+  ~MockMaxConnectionDurationEnabledClusterInfo() override;
+};
+
 } // namespace Upstream
 } // namespace Envoy