tcp: make it possible for TCP connections to be creatd by non-TCP pool) (envoyproxy#13889)

As part of envoyproxy#3431 making sure the ALPN pool can create raw TCP active clients by allowing them to be created by a generic connection pool.

Signed-off-by: Alyssa Wilk <[email protected]>

Author: alyssawilk

source/common/conn_pool/conn_pool_base.h

@@ -174,6 +174,7 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   const Network::TransportSocketOptionsSharedPtr& transportSocketOptions() {
     return transport_socket_options_;
   }
+  bool hasPendingStreams() const { return !pending_streams_.empty(); }
 
 protected:
   // Creates up to 3 connections, based on the prefetch ratio.

source/common/tcp/conn_pool.cc

@@ -12,7 +12,8 @@
 namespace Envoy {
 namespace Tcp {
 
-ActiveTcpClient::ActiveTcpClient(ConnPoolImpl& parent, const Upstream::HostConstSharedPtr& host,
+ActiveTcpClient::ActiveTcpClient(Envoy::ConnectionPool::ConnPoolImplBase& parent,
+                                 const Upstream::HostConstSharedPtr& host,
                                  uint64_t concurrent_stream_limit)
     : Envoy::ConnectionPool::ActiveClient(parent, host->cluster().maxRequestsPerConnection(),
                                           concurrent_stream_limit),
@@ -24,6 +25,12 @@ ActiveTcpClient::ActiveTcpClient(ConnPoolImpl& parent, const Upstream::HostConst
   connection_->addConnectionCallbacks(*this);
   connection_->detectEarlyCloseWhenReadDisabled(false);
   connection_->addReadFilter(std::make_shared<ConnReadFilter>(*this));
+  connection_->setConnectionStats({host->cluster().stats().upstream_cx_rx_bytes_total_,
+                                   host->cluster().stats().upstream_cx_rx_bytes_buffered_,
+                                   host->cluster().stats().upstream_cx_tx_bytes_total_,
+                                   host->cluster().stats().upstream_cx_tx_bytes_buffered_,
+                                   &host->cluster().stats().bind_errors_, nullptr});
+
   connection_->connect();
 }
 
@@ -37,13 +44,12 @@ ActiveTcpClient::~ActiveTcpClient() {
     parent_.onStreamClosed(*this, true);
     parent_.checkForDrained();
   }
-  parent_.onConnDestroyed();
 }
 
 void ActiveTcpClient::clearCallbacks() {
-  if (state_ == Envoy::ConnectionPool::ActiveClient::State::BUSY ||
-      state_ == Envoy::ConnectionPool::ActiveClient::State::DRAINING) {
-    parent_.onConnReleased(*this);
+  if (state_ == Envoy::ConnectionPool::ActiveClient::State::BUSY && parent_.hasPendingStreams()) {
+    auto* pool = &parent_;
+    pool->dispatcher().post([pool]() -> void { pool->onUpstreamReady(); });
   }
   callbacks_ = nullptr;
   tcp_connection_data_ = nullptr;

source/common/tcp/conn_pool.h

@@ -84,8 +84,8 @@ class ActiveTcpClient : public Envoy::ConnectionPool::ActiveClient {
     Network::ClientConnection& connection_;
   };
 
-  ActiveTcpClient(ConnPoolImpl& parent, const Upstream::HostConstSharedPtr& host,
-                  uint64_t concurrent_stream_limit);
+  ActiveTcpClient(Envoy::ConnectionPool::ConnPoolImplBase& parent,
+                  const Upstream::HostConstSharedPtr& host, uint64_t concurrent_stream_limit);
   ~ActiveTcpClient() override;
 
   // Override the default's of Envoy::ConnectionPool::ActiveClient for class-specific functions.
@@ -106,9 +106,9 @@ class ActiveTcpClient : public Envoy::ConnectionPool::ActiveClient {
       close();
     }
   }
-  void clearCallbacks();
+  virtual void clearCallbacks();
 
-  ConnPoolImpl& parent_;
+  Envoy::ConnectionPool::ConnPoolImplBase& parent_;
   ConnectionPool::UpstreamCallbacks* callbacks_{};
   Network::ClientConnectionPtr connection_;
   ConnectionPool::ConnectionStatePtr connection_state_;
@@ -123,11 +123,7 @@ class ConnPoolImpl : public Envoy::ConnectionPool::ConnPoolImplBase,
                const Network::ConnectionSocket::OptionsSharedPtr& options,
                Network::TransportSocketOptionsSharedPtr transport_socket_options)
       : Envoy::ConnectionPool::ConnPoolImplBase(host, priority, dispatcher, options,
-                                                transport_socket_options),
-        upstream_ready_cb_(dispatcher.createSchedulableCallback([this]() {
-          upstream_ready_enabled_ = false;
-          onUpstreamReady();
-        })) {}
+                                                transport_socket_options) {}
   ~ConnPoolImpl() override { destructAllConnections(); }
 
   void addDrainedCallback(DrainedCb cb) override { addDrainedCallbackImpl(cb); }
@@ -196,18 +192,8 @@ class ConnPoolImpl : public Envoy::ConnectionPool::ConnPoolImplBase,
   }
 
   // These two functions exist for testing parity between old and new Tcp Connection Pools.
-  virtual void onConnReleased(Envoy::ConnectionPool::ActiveClient& client) {
-    if (client.state_ == Envoy::ConnectionPool::ActiveClient::State::BUSY) {
-      if (!pending_streams_.empty() && !upstream_ready_enabled_) {
-        upstream_ready_cb_->scheduleCallbackCurrentIteration();
-      }
-    }
-  }
+  virtual void onConnReleased(Envoy::ConnectionPool::ActiveClient&) {}
   virtual void onConnDestroyed() {}
-
-protected:
-  Event::SchedulableCallbackPtr upstream_ready_cb_;
-  bool upstream_ready_enabled_{};
 };
 
 } // namespace Tcp

test/common/tcp/conn_pool_test.cc

@@ -67,6 +67,22 @@ struct ConnPoolCallbacks : public Tcp::ConnectionPool::Callbacks {
   Upstream::HostDescriptionConstSharedPtr host_;
 };
 
+class TestActiveTcpClient : public ActiveTcpClient {
+public:
+  using ActiveTcpClient::ActiveTcpClient;
+
+  ~TestActiveTcpClient() override { parent().onConnDestroyed(); }
+  void clearCallbacks() override {
+    if (state_ == Envoy::ConnectionPool::ActiveClient::State::BUSY ||
+        state_ == Envoy::ConnectionPool::ActiveClient::State::DRAINING) {
+      parent().onConnReleased(*this);
+    }
+    ActiveTcpClient::clearCallbacks();
+  }
+
+  Envoy::Tcp::ConnPoolImpl& parent() { return *static_cast<ConnPoolImpl*>(&parent_); }
+};
+
 /**
  * A wrapper around a ConnectionPoolImpl which tracks when the bridge between
  * the pool and the consumer of the connection is released and destroyed.
@@ -143,7 +159,13 @@ class ConnPoolBase : public Tcp::ConnectionPool::Instance {
       parent_.onConnReleasedForTest();
     }
 
+    Envoy::ConnectionPool::ActiveClientPtr instantiateActiveClient() override {
+      return std::make_unique<TestActiveTcpClient>(
+          *this, Envoy::ConnectionPool::ConnPoolImplBase::host(), 1);
+    }
+
     void onConnDestroyed() override { parent_.onConnDestroyedForTest(); }
+    Event::PostCb post_cb_;
     ConnPoolBase& parent_;
   };
 
@@ -202,12 +224,11 @@ void ConnPoolBase::expectEnableUpstreamReady(bool run) {
   if (!test_new_connection_pool_) {
     dynamic_cast<OriginalConnPoolImplForTest*>(conn_pool_.get())->expectEnableUpstreamReady(run);
   } else {
-    if (!run) {
-      EXPECT_CALL(*mock_upstream_ready_cb_, scheduleCallbackCurrentIteration())
-          .Times(1)
-          .RetiresOnSaturation();
+    Event::PostCb& post_cb = dynamic_cast<ConnPoolImplForTest*>(conn_pool_.get())->post_cb_;
+    if (run) {
+      post_cb();
     } else {
-      mock_upstream_ready_cb_->invokeCallback();
+      EXPECT_CALL(mock_dispatcher_, post(_)).WillOnce(testing::SaveArg<0>(&post_cb));
     }
   }
 }
@@ -219,7 +240,9 @@ class TcpConnPoolImplTest : public testing::TestWithParam<bool> {
 public:
   TcpConnPoolImplTest()
       : test_new_connection_pool_(GetParam()),
-        upstream_ready_cb_(new NiceMock<Event::MockSchedulableCallback>(&dispatcher_)),
+        upstream_ready_cb_(test_new_connection_pool_
+                               ? nullptr
+                               : new NiceMock<Event::MockSchedulableCallback>(&dispatcher_)),
         host_(Upstream::makeTestHost(cluster_, "tcp://127.0.0.1:9000")),
         conn_pool_(dispatcher_, host_, upstream_ready_cb_, test_new_connection_pool_) {}
 
@@ -244,7 +267,9 @@ class TcpConnPoolImplDestructorTest : public testing::TestWithParam<bool> {
 public:
   TcpConnPoolImplDestructorTest()
       : test_new_connection_pool_(GetParam()),
-        upstream_ready_cb_(new NiceMock<Event::MockSchedulableCallback>(&dispatcher_)) {
+        upstream_ready_cb_(test_new_connection_pool_
+                               ? nullptr
+                               : new NiceMock<Event::MockSchedulableCallback>(&dispatcher_)) {
     host_ = Upstream::makeTestHost(cluster_, "tcp://127.0.0.1:9000");
     if (test_new_connection_pool_) {
       conn_pool_ = std::make_unique<ConnPoolImpl>(