upstream: allow configuration of connection pool limits

api/envoy/api/v2/cluster/circuit_breaker.proto

@@ -51,6 +51,13 @@ message CircuitBreakers {
     // the number of resources remaining until the circuit breakers open. If
     // not specified, the default is false.
     bool track_remaining = 6;
+
+    // The maximum number of connection pools per cluster that Envoy will concurrently support at
+    // once. If not specified, the default is unlimited. Set this for clusters which create a
+    // large number of connection pools. See
+    // :ref:`Circuit Breaking <arch_overview_circuit_break_cluster_maximum_connection_pools>` for
+    // more details.
+    google.protobuf.UInt32Value max_connection_pools = 7;
   }
 
   // If multiple :ref:`Thresholds<envoy_api_msg_cluster.CircuitBreakers.Thresholds>`

docs/root/configuration/cluster_manager/cluster_stats.rst

@@ -149,6 +149,7 @@ Circuit breakers statistics will be rooted at *cluster.<name>.circuit_breakers.<
   :widths: 1, 1, 2
 
   cx_open, Gauge, Whether the connection circuit breaker is closed (0) or open (1)
+  cx_pool_open, Gauge, Whether the connection pool circuit breaker is closed (0) or open (1)
   rq_pending_open, Gauge, Whether the pending requests circuit breaker is closed (0) or open (1)
   rq_open, Gauge, Whether the requests circuit breaker is closed (0) or open (1)
   rq_retry_open, Gauge, Whether the retry circuit breaker is closed (0) or open (1)

docs/root/intro/arch_overview/circuit_breaking.rst

@@ -9,6 +9,8 @@ mesh is that Envoy enforces circuit breaking limits at the network level as oppo
 configure and code each application independently. Envoy supports various types of fully distributed
 (not coordinated) circuit breaking:
 
+.. _arch_overview_circuit_break_cluster_maximum_connections:
+
 * **Cluster maximum connections**: The maximum number of connections that Envoy will establish to
   all hosts in an upstream cluster. In practice this is only applicable to HTTP/1.1 clusters since
   HTTP/2 uses a single connection to each host. If this circuit breaker overflows the :ref:`upstream_cx_overflow
@@ -34,6 +36,20 @@ configure and code each application independently. Envoy supports various types
   :ref:`upstream_rq_retry_overflow <config_cluster_manager_cluster_stats>` counter for the cluster
   will increment.
 
+  .. _arch_overview_circuit_break_cluster_maximum_connection_pools:
+
+* **Cluster maximum concurrent connection pools**: The maximum number of connection pools that can be
+  concurrently instantiated. Some features, such as the
+  :ref:`Original Src Listener Filter <arch_overview_ip_transparency_original_src_listener>`, can
+  create an unbounded number of connection pools. When a cluster has exhausted its concurrent
+  connection pools, it will attempt to reclaim an idle one. If it cannot, then the circuit breaker
+  will overflow. This differs from
+  :ref:`Cluster maximum connections <arch_overview_circuit_break_cluster_maximum_connections>` in that
+  connection pools never time out, whereas connections typically will. Connections automatically
+  clean up; connection pools do not. Note that in order for a connection pool to function it needs
+  at least one upstream connection, so this value should likely be no greater than
+  :ref:`Cluster maximum connections <arch_overview_circuit_break_cluster_maximum_connections>`.
+
 Each circuit breaking limit is :ref:`configurable <config_cluster_manager_cluster_circuit_breakers>`
 and tracked on a per upstream cluster and per priority basis. This allows different components of
 the distributed system to be tuned independently and have different limits. The live state of these

docs/root/intro/version_history.rst

@@ -69,6 +69,7 @@ Version history
 * upstream: added :ref:`degraded health value<arch_overview_load_balancing_degraded>` which allows
   routing to certain hosts only when there are insufficient healthy hosts available.
 * upstream: add cluster factory to allow creating and registering :ref:`custom cluster type<arch_overview_service_discovery_types_custom>`.
+* upstream: added a :ref:`circuit breaker <arch_overview_circuit_break_cluster_maximum_connection_pools>` to limit the number of concurrent connection pools in use.
 * tracing: added :ref:`verbose <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.tracing>` to support logging annotations on spans.
 * upstream: added support for host weighting and :ref:`locality weighting <arch_overview_load_balancing_locality_weighted_lb>` in the :ref:`ring hash load balancer <arch_overview_load_balancing_types_ring_hash>`, and added a :ref:`maximum_ring_size<envoy_api_field_Cluster.RingHashLbConfig.maximum_ring_size>` config parameter to strictly bound the ring size.
 * zookeeper: added a ZooKeeper proxy filter that parses ZooKeeper messages (requests/responses/events).

include/envoy/upstream/resource_manager.h

@@ -37,6 +37,11 @@ class Resource {
    */
   virtual void dec() PURE;
 
+  /**
+   * Decrement the resource count by a specific amount.
+   */
+  virtual void decBy(uint64_t amount) PURE;
+
   /**
    * @return the current maximum allowed number of this resource.
    */
@@ -73,6 +78,11 @@ class ResourceManager {
    * @return Resource& active retries.
    */
   virtual Resource& retries() PURE;
+
+  /**
+   * @return Resource& active connection pools.
+   */
+  virtual Resource& connectionPools() PURE;
 };
 
 } // namespace Upstream

include/envoy/upstream/upstream.h

@@ -537,10 +537,12 @@ class PrioritySet {
   OPEN_GAUGE      (rq_pending_open)                                                                \
   OPEN_GAUGE      (rq_open)                                                                        \
   OPEN_GAUGE      (rq_retry_open)                                                                  \
+  OPEN_GAUGE      (cx_pool_open)                                                                   \
   REMAINING_GAUGE (remaining_cx)                                                                   \
   REMAINING_GAUGE (remaining_pending)                                                              \
   REMAINING_GAUGE (remaining_rq)                                                                   \
-  REMAINING_GAUGE (remaining_retries)
+  REMAINING_GAUGE (remaining_retries)                                                              \
+  REMAINING_GAUGE (remaining_cx_pools)
 // clang-format on
 
 /**

source/common/upstream/BUILD

@@ -60,8 +60,7 @@ envoy_cc_library(
         "//source/common/protobuf:utility_lib",
         "//source/common/router:shadow_writer_lib",
         "//source/common/tcp:conn_pool_lib",
-        "//source/common/upstream:conn_pool_map",
-        "//source/common/upstream:conn_pool_map_impl_lib",
+        "//source/common/upstream:priority_conn_pool_map_impl_lib",
         "//source/common/upstream:upstream_lib",
         "@envoy_api//envoy/admin/v2alpha:config_dump_cc",
         "@envoy_api//envoy/api/v2/core:base_cc",
@@ -73,6 +72,8 @@ envoy_cc_library(
     hdrs = ["conn_pool_map.h"],
     deps = [
         "//include/envoy/event:dispatcher_interface",
+        "//include/envoy/upstream:resource_manager_interface",
+        "//include/envoy/upstream:upstream_interface",
         "//source/common/common:debug_recursion_checker_lib",
     ],
 )
@@ -85,6 +86,27 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "priority_conn_pool_map",
+    hdrs = ["priority_conn_pool_map.h"],
+    deps = [
+        ":conn_pool_map",
+        "//include/envoy/event:dispatcher_interface",
+        "//include/envoy/upstream:resource_manager_interface",
+        "//include/envoy/upstream:upstream_interface",
+        "//source/common/common:debug_recursion_checker_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "priority_conn_pool_map_impl_lib",
+    hdrs = ["priority_conn_pool_map_impl.h"],
+    deps = [
+        ":conn_pool_map_impl_lib",
+        ":priority_conn_pool_map",
+    ],
+)
+
 envoy_cc_library(
     name = "edf_scheduler_lib",
     hdrs = ["edf_scheduler.h"],

source/common/upstream/cluster_manager_impl.cc

@@ -31,10 +31,10 @@
 #include "common/router/shadow_writer_impl.h"
 #include "common/tcp/conn_pool.h"
 #include "common/upstream/cds_api_impl.h"
-#include "common/upstream/conn_pool_map_impl.h"
 #include "common/upstream/load_balancer_impl.h"
 #include "common/upstream/maglev_lb.h"
 #include "common/upstream/original_dst_cluster.h"
+#include "common/upstream/priority_conn_pool_map_impl.h"
 #include "common/upstream/ring_hash_lb.h"
 #include "common/upstream/subset_lb.h"
 
@@ -1043,7 +1043,7 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::getHttpConnPoolsContainer(
     if (!allocate) {
       return nullptr;
     }
-    ConnPoolsContainer container{thread_local_dispatcher_};
+    ConnPoolsContainer container{thread_local_dispatcher_, host};
     container_iter = host_http_conn_pool_map_.emplace(host, std::move(container)).first;
   }
 
@@ -1132,7 +1132,7 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::ClusterEntry::connPool(
   }
 
   // Inherit socket options from downstream connection, if set.
-  std::vector<uint8_t> hash_key = {uint8_t(protocol), uint8_t(priority)};
+  std::vector<uint8_t> hash_key = {uint8_t(protocol)};
 
   // Use downstream connection socket options for computing connection pool hash key, if any.
   // This allows socket options to control connection pooling so that connections with
@@ -1153,16 +1153,18 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::ClusterEntry::connPool(
 
   // Note: to simplify this, we assume that the factory is only called in the scope of this
   // function. Otherwise, we'd need to capture a few of these variables by value.
-  ConnPoolsContainer::ConnPools::OptPoolRef pool = container.pools_->getPool(hash_key, [&]() {
-    return parent_.parent_.factory_.allocateConnPool(
-        parent_.thread_local_dispatcher_, host, priority, protocol,
-        have_options ? context->downstreamConnection()->socketOptions() : nullptr);
-  });
-  // The Connection Pool tracking is a work in progress. We plan for it to eventually have the
-  // ability to fail, but until we add upper layer handling for failures, it should not. So, assert
-  // that we don't accidentally add conditions that could allow it to fail.
-  ASSERT(pool.has_value(), "Pool allocation should never fail");
-  return &(pool.value().get());
+  ConnPoolsContainer::ConnPools::OptPoolRef pool =
+      container.pools_->getPool(priority, hash_key, [&]() {
+        return parent_.parent_.factory_.allocateConnPool(
+            parent_.thread_local_dispatcher_, host, priority, protocol,
+            have_options ? context->downstreamConnection()->socketOptions() : nullptr);
+      });
+
+  if (pool.has_value()) {
+    return &(pool.value().get());
+  } else {
+    return nullptr;
+  }
 }
 
 Tcp::ConnectionPool::Instance*

source/common/upstream/cluster_manager_impl.h

@@ -22,8 +22,8 @@
 
 #include "common/config/grpc_mux_impl.h"
 #include "common/http/async_client_impl.h"
-#include "common/upstream/conn_pool_map.h"
 #include "common/upstream/load_stats_reporter.h"
+#include "common/upstream/priority_conn_pool_map.h"
 #include "common/upstream/upstream_impl.h"
 
 namespace Envoy {
@@ -236,10 +236,10 @@ class ClusterManagerImpl : public ClusterManager, Logger::Loggable<Logger::Id::u
    */
   struct ThreadLocalClusterManagerImpl : public ThreadLocal::ThreadLocalObject {
     struct ConnPoolsContainer {
-      ConnPoolsContainer(Event::Dispatcher& dispatcher)
-          : pools_{std::make_shared<ConnPools>(dispatcher, absl::nullopt)} {}
+      ConnPoolsContainer(Event::Dispatcher& dispatcher, const HostConstSharedPtr& host)
+          : pools_{std::make_shared<ConnPools>(dispatcher, host)} {}
 
-      typedef ConnPoolMap<std::vector<uint8_t>, Http::ConnectionPool::Instance> ConnPools;
+      typedef PriorityConnPoolMap<std::vector<uint8_t>, Http::ConnectionPool::Instance> ConnPools;
 
       // This is a shared_ptr so we can keep it alive while cleaning up.
       std::shared_ptr<ConnPools> pools_;

source/common/upstream/conn_pool_map.h

@@ -4,6 +4,8 @@
 #include <vector>
 
 #include "envoy/event/dispatcher.h"
+#include "envoy/upstream/resource_manager.h"
+#include "envoy/upstream/upstream.h"
 
 #include "common/common/debug_recursion_checker.h"
 
@@ -21,7 +23,8 @@ template <typename KEY_TYPE, typename POOL_TYPE> class ConnPoolMap {
   using DrainedCb = std::function<void()>;
   using OptPoolRef = absl::optional<std::reference_wrapper<POOL_TYPE>>;
 
-  ConnPoolMap(Event::Dispatcher& dispatcher, absl::optional<uint64_t> max_size);
+  ConnPoolMap(Event::Dispatcher& dispatcher, const HostConstSharedPtr& host,
+              ResourcePriority priority);
   ~ConnPoolMap();
   /**
    * Returns an existing pool for `key`, or creates a new one using `factory`. Note that it is
@@ -60,11 +63,17 @@ template <typename KEY_TYPE, typename POOL_TYPE> class ConnPoolMap {
    */
   bool freeOnePool();
 
+  /**
+   * Cleans up the active_pools_ map and updates resource tracking
+   **/
+  void clearActivePools();
+
   absl::flat_hash_map<KEY_TYPE, std::unique_ptr<POOL_TYPE>> active_pools_;
   Event::Dispatcher& thread_local_dispatcher_;
   std::vector<DrainedCb> cached_callbacks_;
   Common::DebugRecursionChecker recursion_checker_;
-  const absl::optional<uint64_t> max_size_;
+  const HostConstSharedPtr host_;
+  const ResourcePriority priority_;
 };
 
 } // namespace Upstream

source/common/upstream/conn_pool_map_impl.h

@@ -7,11 +7,17 @@ namespace Upstream {
 
 template <typename KEY_TYPE, typename POOL_TYPE>
 ConnPoolMap<KEY_TYPE, POOL_TYPE>::ConnPoolMap(Envoy::Event::Dispatcher& dispatcher,
-                                              absl::optional<uint64_t> max_size)
-    : thread_local_dispatcher_(dispatcher), max_size_(max_size) {}
-
-template <typename KEY_TYPE, typename POOL_TYPE>
-ConnPoolMap<KEY_TYPE, POOL_TYPE>::~ConnPoolMap() = default;
+                                              const HostConstSharedPtr& host,
+                                              ResourcePriority priority)
+    : thread_local_dispatcher_(dispatcher), host_(host), priority_(priority) {}
+
+template <typename KEY_TYPE, typename POOL_TYPE> ConnPoolMap<KEY_TYPE, POOL_TYPE>::~ConnPoolMap() {
+  // Clean up the pools to ensure resource tracking is kept up to date. Note that we do not call
+  // `clear()` here to avoid doing a deferred delete. This triggers some unwanted race conditions
+  // on shutdown where deleted resources end up putting stuff on the deferred delete list after the
+  // worker threads have shut down.
+  clearActivePools();
+}
 
 template <typename KEY_TYPE, typename POOL_TYPE>
 typename ConnPoolMap<KEY_TYPE, POOL_TYPE>::OptPoolRef
@@ -24,15 +30,19 @@ ConnPoolMap<KEY_TYPE, POOL_TYPE>::getPool(KEY_TYPE key, const PoolFactory& facto
   if (pool_iter != active_pools_.end()) {
     return std::ref(*(pool_iter->second));
   }
-
+  Resource& connPoolResource = host_->cluster().resourceManager(priority_).connectionPools();
   // We need a new pool. Check if we have room.
-  if (max_size_.has_value() && size() >= max_size_.value()) {
+  if (!connPoolResource.canCreate()) {
     // We're full. Try to free up a pool. If we can't, bail out.
     if (!freeOnePool()) {
+      // TODO(klarose): Add some explicit counters for failure cases here, similar to the other
+      // circuit breakers.
       return absl::nullopt;
     }
 
-    ASSERT(size() < max_size_.value(), "Freeing a pool should reduce the size to below the max.");
+    ASSERT(size() < connPoolResource.max(),
+           "Freeing a pool should reduce the size to below the max.");
+
     // TODO(klarose): Consider some simple hysteresis here. How can we prevent iterating over all
     // pools when we're at the limit every time we want to allocate a new one, even if most of the
     // pools are not busy, while balancing that with not unnecessarily freeing all pools? If we
@@ -42,6 +52,7 @@ ConnPoolMap<KEY_TYPE, POOL_TYPE>::getPool(KEY_TYPE key, const PoolFactory& facto
 
   // We have room for a new pool. Allocate one and let it know about any cached callbacks.
   auto new_pool = factory();
+  connPoolResource.inc();
   for (const auto& cb : cached_callbacks_) {
     new_pool->addDrainedCallback(cb);
   }
@@ -60,8 +71,7 @@ template <typename KEY_TYPE, typename POOL_TYPE> void ConnPoolMap<KEY_TYPE, POOL
   for (auto& pool_pair : active_pools_) {
     thread_local_dispatcher_.deferredDelete(std::move(pool_pair.second));
   }
-
-  active_pools_.clear();
+  clearActivePools();
 }
 
 template <typename KEY_TYPE, typename POOL_TYPE>
@@ -96,11 +106,17 @@ bool ConnPoolMap<KEY_TYPE, POOL_TYPE>::freeOnePool() {
   if (pool_iter != active_pools_.end()) {
     // We found one. Free it up, and let the caller know.
     active_pools_.erase(pool_iter);
+    host_->cluster().resourceManager(priority_).connectionPools().dec();
     return true;
   }
 
   return false;
 }
 
+template <typename KEY_TYPE, typename POOL_TYPE>
+void ConnPoolMap<KEY_TYPE, POOL_TYPE>::clearActivePools() {
+  host_->cluster().resourceManager(priority_).connectionPools().decBy(active_pools_.size());
+  active_pools_.clear();
+}
 } // namespace Upstream
 } // namespace Envoy