http: wire ConnectionLifetimeCallbacks into pool connection events (#43702)

## Commit Message
http: wire ConnectionLifetimeCallbacks into pool connection events

## Description
Wires up the `ConnectionLifetimeCallbacks` interface that is there but
is not connected to actual connection events from what I see at least.
The cluster manager now passes the LB's callbacks into each HTTP pool at
creation time, and `MultiplexedActiveClientBase` fires them on
Connected, Close, and GOAWAY events.

Primary motivation: allows the reverse tunnel cluster's LB to detect
GOAWAY and report it to the tunnel reporter ([raw
example](aakugan@31094c9)).
Also applicable to DFP clusters for connection coalescing awareness from
what I [see](#17874).

## Risk Level
Should be low. Additive-only changes to production code. Callbacks are
no-ops unless explicitly set by the LB. No behavior change for existing
clusters.

## Testing
- HTTP/2 conn pool unit tests (7 tests): no-op without callbacks,
Connected/RemoteClose/LocalClose with arg verification, GOAWAY with
active streams, idle GOAWAY, callback replacement.
- HTTP/3 conn pool unit test: Connected + idle GOAWAY through
`MultiplexedActiveClientBase`.
- Cluster manager unit tests: verifies `setLifetimeCallbacks` is called
on pool creation.

## Docs Changes
N/A

## Release Notes
N/A

---------

Signed-off-by: aakugan <aakashganapathy2@gmail.com>

Author: aakugan

envoy/http/conn_pool.h

@@ -2,6 +2,7 @@
 
 #include <functional>
 #include <memory>
+#include <vector>
 
 #include "envoy/common/conn_pool.h"
 #include "envoy/common/pure.h"
@@ -119,6 +120,9 @@ class Instance : public Envoy::ConnectionPool::Instance, public Event::DeferredD
    * @return absl::string_view a protocol description for logging.
    */
   virtual absl::string_view protocolDescription() const PURE;
+
+  virtual void setLifetimeCallbacks(OptRef<ConnectionLifetimeCallbacks> callbacks,
+                                    std::vector<uint8_t> hash_key) PURE;
 };
 
 using InstancePtr = std::unique_ptr<Instance>;

source/common/http/codec_client.h

@@ -155,6 +155,9 @@ class CodecClient : protected Logger::Loggable<Logger::Id::client>,
 
   bool connectCalled() const { return connect_called_; }
 
+  // Get the underlying connection of the codec client for lifetimeCallbacks.
+  const Network::Connection& connection() const { return *connection_; }
+
 protected:
   /**
    * Create a codec client and connect to a remote host/port.

source/common/http/conn_pool_base.cc

@@ -104,6 +104,24 @@ void HttpConnPoolImplBase::onPoolReady(Envoy::ConnectionPool::ActiveClient& clie
                         http_client->codec_client_->protocol());
 }
 
+void HttpConnPoolImplBase::setLifetimeCallbacks(
+    OptRef<ConnectionPool::ConnectionLifetimeCallbacks> callbacks, std::vector<uint8_t> hash_key) {
+  callbacks_ = callbacks;
+  hash_key_ = std::move(hash_key);
+}
+
+void HttpConnPoolImplBase::onConnectionOpen(const Network::Connection& connection) {
+  if (callbacks_.has_value()) {
+    callbacks_->onConnectionOpen(*this, hash_key_, connection);
+  }
+}
+
+void HttpConnPoolImplBase::onConnectionDraining(const Network::Connection& connection) {
+  if (callbacks_.has_value()) {
+    callbacks_->onConnectionDraining(*this, hash_key_, connection);
+  }
+}
+
 // All streams are 2^31. Client streams are half that, minus stream 0. Just to be on the safe
 // side we do 2^29.
 constexpr uint32_t DEFAULT_MAX_STREAMS = 1U << 29;
@@ -115,6 +133,7 @@ void MultiplexedActiveClientBase::onGoAway(Http::GoAwayErrorCode) {
     if (codec_client_->numActiveRequests() == 0) {
       codec_client_->close();
     } else {
+      parent().onConnectionDraining(codec_client_->connection());
       parent_.transitionActiveClientState(*this, ActiveClient::State::Draining);
     }
   }
@@ -222,5 +241,17 @@ MultiplexedActiveClientBase::newStreamEncoder(ResponseDecoderHandlePtr response_
   return codec_client_->newStream(std::move(response_decoder_handle));
 }
 
+void MultiplexedActiveClientBase::onEvent(Network::ConnectionEvent event) {
+  if (event == Network::ConnectionEvent::Connected ||
+      event == Network::ConnectionEvent::ConnectedZeroRtt) {
+    parent().onConnectionOpen(codec_client_->connection());
+  } else if (event == Network::ConnectionEvent::LocalClose ||
+             event == Network::ConnectionEvent::RemoteClose) {
+    parent().onConnectionDraining(codec_client_->connection());
+  }
+
+  ActiveClient::onEvent(event);
+}
+
 } // namespace Http
 } // namespace Envoy

source/common/http/conn_pool_base.h

@@ -98,6 +98,13 @@ class HttpConnPoolImplBase : public Envoy::ConnectionPool::ConnPoolImplBase,
   virtual absl::optional<HttpServerPropertiesCache::Origin>& origin() { return origin_; }
   virtual Http::HttpServerPropertiesCacheSharedPtr cache() { return nullptr; }
 
+  void setLifetimeCallbacks(OptRef<ConnectionPool::ConnectionLifetimeCallbacks> callbacks,
+                            std::vector<uint8_t> hash_key) override;
+
+  void onConnectionOpen(const Network::Connection& connection);
+
+  void onConnectionDraining(const Network::Connection& connection);
+
 protected:
   friend class ActiveClient;
 
@@ -107,6 +114,8 @@ class HttpConnPoolImplBase : public Envoy::ConnectionPool::ConnPoolImplBase,
 
 private:
   absl::optional<HttpServerPropertiesCache::Origin> origin_;
+  OptRef<ConnectionPool::ConnectionLifetimeCallbacks> callbacks_;
+  std::vector<uint8_t> hash_key_;
 };
 
 // An implementation of Envoy::ConnectionPool::ActiveClient for HTTP/1.1 and HTTP/2
@@ -242,6 +251,9 @@ class MultiplexedActiveClientBase : public CodecClientCallbacks,
   void onGoAway(Http::GoAwayErrorCode error_code) override;
   void onSettings(ReceivedSettings& settings) override;
 
+  // Override to provide the lifetimeCallbacks.
+  void onEvent(Network::ConnectionEvent event) override;
+
 private:
   bool closed_with_active_rq_{};
 };

source/common/http/conn_pool_grid.cc

@@ -417,6 +417,7 @@ ConnectionPool::InstancePtr ConnectivityGrid::createHttp3Pool(bool attempt_alter
 
 void ConnectivityGrid::setupPool(ConnectionPool::Instance& pool) {
   pool.addIdleCallback([this]() { onIdleReceived(); });
+  pool.setLifetimeCallbacks(makeOptRefFromPtr(this), hash_key_);
 }
 
 bool ConnectivityGrid::hasActiveConnections() const {
@@ -610,5 +611,25 @@ void ConnectivityGrid::onZeroRttHandshakeFailed() {
   getHttp3StatusTracker().markHttp3FailedRecently();
 }
 
+void ConnectivityGrid::onConnectionOpen(ConnectionPool::Instance&, std::vector<uint8_t>&,
+                                        const Network::Connection& connection) {
+  if (callbacks_.has_value()) {
+    callbacks_->onConnectionOpen(*this, hash_key_, connection);
+  }
+}
+
+void ConnectivityGrid::onConnectionDraining(ConnectionPool::Instance&, std::vector<uint8_t>&,
+                                            const Network::Connection& connection) {
+  if (callbacks_.has_value()) {
+    callbacks_->onConnectionDraining(*this, hash_key_, connection);
+  }
+}
+
+void ConnectivityGrid::setLifetimeCallbacks(
+    OptRef<ConnectionPool::ConnectionLifetimeCallbacks> callbacks, std::vector<uint8_t> hash_key) {
+  callbacks_ = callbacks;
+  hash_key_ = std::move(hash_key);
+}
+
 } // namespace Http
 } // namespace Envoy

source/common/http/conn_pool_grid.h

@@ -30,6 +30,7 @@ namespace Http {
 // the host's address list.
 class ConnectivityGrid : public ConnectionPool::Instance,
                          public Http3::PoolConnectResultCallback,
+                         public ConnectionPool::ConnectionLifetimeCallbacks,
                          protected Logger::Loggable<Logger::Id::pool> {
 public:
   struct ConnectivityOptions {
@@ -228,6 +229,15 @@ class ConnectivityGrid : public ConnectionPool::Instance,
   void onHandshakeComplete() override;
   void onZeroRttHandshakeFailed() override;
 
+  // ConnectionPool::ConnectionLifetimeCallbacks
+  void onConnectionOpen(ConnectionPool::Instance& pool, std::vector<uint8_t>& hash_key,
+                        const Network::Connection& connection) override;
+  void onConnectionDraining(ConnectionPool::Instance& pool, std::vector<uint8_t>& hash_key,
+                            const Network::Connection& connection) override;
+
+  void setLifetimeCallbacks(OptRef<ConnectionPool::ConnectionLifetimeCallbacks> callbacks,
+                            std::vector<uint8_t> hash_key) override;
+
 protected:
   // Set the required idle callback on the pool.
   void setupPool(ConnectionPool::Instance& pool);
@@ -308,6 +318,9 @@ class ConnectivityGrid : public ConnectionPool::Instance,
   bool deferred_deleting_{};
 
   OptRef<Quic::EnvoyQuicNetworkObserverRegistry> network_observer_registry_;
+
+  OptRef<ConnectionPool::ConnectionLifetimeCallbacks> callbacks_;
+  std::vector<uint8_t> hash_key_;
 };
 
 } // namespace Http

source/common/upstream/cluster_manager_impl.cc

@@ -2023,6 +2023,8 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::ClusterEntry::httpConnPoolImp
           parent.httpConnPoolIsIdle(host, priority, hash_key);
         });
 
+        pool->setLifetimeCallbacks(lb_->lifetimeCallbacks(), hash_key);
+
         return pool;
       });