upstream: support max connection duration for upstream HTTP connections

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

@@ -141,7 +141,6 @@ message HttpProtocolOptions {
   // the connection will be closed. Drain sequence will occur prior to closing the connection if
   // if's applicable. See :ref:`drain_timeout
   // <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.drain_timeout>`.
-  // Note: not implemented for upstream connections.
   google.protobuf.Duration max_connection_duration = 3;
 
   // The maximum number of headers. If unconfigured, the default

envoy/upstream/upstream.h

@@ -743,6 +743,11 @@ class ClusterInfo {
    */
   virtual const absl::optional<std::chrono::milliseconds> idleTimeout() const PURE;
 
+  /**
+   * @return optional maximum connection duration timeout for manager connections.
+   */
+  virtual const absl::optional<std::chrono::milliseconds> maxConnectionDuration() const PURE;
+
   /**
    * @return how many streams should be anticipated per each current stream.
    */

source/common/conn_pool/conn_pool_base.cc

@@ -564,6 +564,11 @@ ActiveClient::ActiveClient(ConnPoolImplBase& parent, uint32_t lifetime_stream_li
   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();
@@ -596,6 +601,15 @@ 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);
+  }
+}
+
 void ActiveClient::drain() {
   if (currentUnusedCapacity() <= 0) {
     return;

source/common/conn_pool/conn_pool_base.h

@@ -45,6 +45,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 stream limit, accounting for if the total stream limit
   // is less than the concurrent stream limit.
   uint32_t effectiveConcurrentStreamLimit() const {
@@ -106,6 +110,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.h

@@ -613,6 +613,9 @@ class ClusterInfoImpl : public ClusterInfo,
   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_;
+  }
   float perUpstreamPreconnectRatio() const override { return per_upstream_preconnect_ratio_; }
   float peekaheadRatio() const override { return peekahead_ratio_; }
   uint32_t perConnectionBufferLimitBytes() const override {
@@ -765,6 +768,7 @@ class ClusterInfoImpl : public ClusterInfo,
   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 float per_upstream_preconnect_ratio_;
   const float peekahead_ratio_;
   const uint32_t per_connection_buffer_limit_bytes_;

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())),

test/mocks/upstream/cluster_info.h

@@ -90,6 +90,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(const absl::optional<std::chrono::milliseconds>, maxStreamDuration, (), (const));
   MOCK_METHOD(const absl::optional<std::chrono::milliseconds>, grpcTimeoutHeaderMax, (), (const));
   MOCK_METHOD(const absl::optional<std::chrono::milliseconds>, grpcTimeoutHeaderOffset, (),
@@ -215,5 +216,11 @@ class MockIdleTimeEnabledClusterInfo : public MockClusterInfo {
   ~MockIdleTimeEnabledClusterInfo() override;
 };
 
+class MockMaxConnectionDurationEnabledClusterInfo : public MockClusterInfo {
+public:
+  MockMaxConnectionDurationEnabledClusterInfo();
+  ~MockMaxConnectionDurationEnabledClusterInfo() override;
+};
+
 } // namespace Upstream
 } // namespace Envoy