dynamic_forward_proxy: DNS Cache circuit breaker

api/envoy/config/cluster/v3/circuit_breaker.proto

@@ -103,3 +103,13 @@ message CircuitBreakers {
   // are used.
   repeated Thresholds thresholds = 1;
 }
+
+message DnsCacheCircuitBreaker {
+  message DnsCacheThreshold {
+    // The maximum number of pending requests that Envoy will allow to the
+    // resolver. If not specified, the default is 1024.
+    google.protobuf.UInt32Value max_pending_requests = 1;
+  }
+
+  DnsCacheThreshold threshold = 1;
+}

api/envoy/extensions/common/dynamic_forward_proxy/v3/dns_cache.proto

@@ -2,6 +2,7 @@ syntax = "proto3";
 
 package envoy.extensions.common.dynamic_forward_proxy.v3;
 
+import "envoy/config/cluster/v3/circuit_breaker.proto";
 import "envoy/config/cluster/v3/cluster.proto";
 
 import "google/protobuf/duration.proto";
@@ -20,7 +21,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 
 // Configuration for the dynamic forward proxy DNS cache. See the :ref:`architecture overview
 // <arch_overview_http_dynamic_forward_proxy>` for more information.
-// [#next-free-field: 7]
+// [#next-free-field: 8]
 message DnsCacheConfig {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.common.dynamic_forward_proxy.v2alpha.DnsCacheConfig";
@@ -83,4 +84,6 @@ message DnsCacheConfig {
   // this is used as the cache's DNS refresh rate when DNS requests are failing. If this setting is
   // not specified, the failure refresh rate defaults to the dns_refresh_rate.
   config.cluster.v3.Cluster.RefreshRate dns_failure_refresh_rate = 6;
+
+  config.cluster.v3.DnsCacheCircuitBreaker dns_cache_circuit_breaker = 7;
 }

source/extensions/clusters/dynamic_forward_proxy/cluster.cc

@@ -198,7 +198,7 @@ ClusterFactory::createClusterWithConfig(
     Stats::ScopePtr&& stats_scope) {
   Extensions::Common::DynamicForwardProxy::DnsCacheManagerFactoryImpl cache_manager_factory(
       context.singletonManager(), context.dispatcher(), context.tls(), context.random(),
-      context.stats());
+      context.runtime(), context.stats());
   envoy::config::cluster::v3::Cluster cluster_config = cluster;
   if (cluster_config.has_upstream_http_protocol_options()) {
     if (!cluster_config.upstream_http_protocol_options().auto_sni() ||

source/extensions/common/dynamic_forward_proxy/BUILD

@@ -12,6 +12,7 @@ envoy_cc_library(
     name = "dns_cache_interface",
     hdrs = ["dns_cache.h"],
     deps = [
+        ":dns_cache_resource_manager",
         "//include/envoy/common:backoff_strategy_interface",
         "//include/envoy/event:dispatcher_interface",
         "//include/envoy/singleton:manager_interface",
@@ -46,3 +47,15 @@ envoy_cc_library(
         "@envoy_api//envoy/extensions/common/dynamic_forward_proxy/v3:pkg_cc_proto",
     ],
 )
+
+envoy_cc_library(
+    name = "dns_cache_resource_manager",
+    srcs = ["dns_cache_resource_manager.cc"],
+    hdrs = ["dns_cache_resource_manager.h"],
+    deps = [
+        "//source/common/common:assert_lib",
+        "//source/common/runtime:runtime_lib",
+        "//source/common/upstream:resource_manager_lib",
+        "@envoy_api//envoy/extensions/common/dynamic_forward_proxy/v3:pkg_cc_proto",
+    ],
+)

source/extensions/common/dynamic_forward_proxy/dns_cache.h

@@ -5,6 +5,8 @@
 #include "envoy/singleton/manager.h"
 #include "envoy/thread_local/thread_local.h"
 
+#include "extensions/common/dynamic_forward_proxy/dns_cache_resource_manager.h"
+
 namespace Envoy {
 namespace Extensions {
 namespace Common {
@@ -148,6 +150,11 @@ class DnsCache {
    * @return all hosts currently stored in the cache.
    */
   virtual absl::flat_hash_map<std::string, DnsHostInfoSharedPtr> hosts() PURE;
+
+  /**
+   * @return resource manager for dns cache.
+   */
+  virtual DnsCacheResourceManager& dnsCacheResourceManager() PURE;
 };
 
 using DnsCacheSharedPtr = std::shared_ptr<DnsCache>;
@@ -176,7 +183,7 @@ using DnsCacheManagerSharedPtr = std::shared_ptr<DnsCacheManager>;
 DnsCacheManagerSharedPtr getCacheManager(Singleton::Manager& manager,
                                          Event::Dispatcher& main_thread_dispatcher,
                                          ThreadLocal::SlotAllocator& tls,
-                                         Runtime::RandomGenerator& random,
+                                         Runtime::RandomGenerator& random, Runtime::Loader& loader,
                                          Stats::Scope& root_scope);
 
 /**

source/extensions/common/dynamic_forward_proxy/dns_cache_impl.cc

@@ -16,13 +16,14 @@ namespace DynamicForwardProxy {
 
 DnsCacheImpl::DnsCacheImpl(
     Event::Dispatcher& main_thread_dispatcher, ThreadLocal::SlotAllocator& tls,
-    Runtime::RandomGenerator& random, Stats::Scope& root_scope,
+    Runtime::RandomGenerator& random, Runtime::Loader& loader, Stats::Scope& root_scope,
     const envoy::extensions::common::dynamic_forward_proxy::v3::DnsCacheConfig& config)
     : main_thread_dispatcher_(main_thread_dispatcher),
       dns_lookup_family_(Upstream::getDnsLookupFamilyFromEnum(config.dns_lookup_family())),
       resolver_(main_thread_dispatcher.createDnsResolver({}, false)), tls_slot_(tls.allocateSlot()),
       scope_(root_scope.createScope(fmt::format("dns_cache.{}.", config.name()))),
       stats_{ALL_DNS_CACHE_STATS(POOL_COUNTER(*scope_), POOL_GAUGE(*scope_))},
+      resource_manager_(loader, config),
       refresh_interval_(PROTOBUF_GET_MS_OR_DEFAULT(config, dns_refresh_rate, 60000)),
       failure_backoff_strategy_(
           Config::Utility::prepareDnsRefreshStrategy<

source/extensions/common/dynamic_forward_proxy/dns_cache_impl.h

@@ -39,7 +39,7 @@ struct DnsCacheStats {
 class DnsCacheImpl : public DnsCache, Logger::Loggable<Logger::Id::forward_proxy> {
 public:
   DnsCacheImpl(Event::Dispatcher& main_thread_dispatcher, ThreadLocal::SlotAllocator& tls,
-               Runtime::RandomGenerator& random, Stats::Scope& root_scope,
+               Runtime::RandomGenerator& random, Runtime::Loader& loader, Stats::Scope& root_scope,
                const envoy::extensions::common::dynamic_forward_proxy::v3::DnsCacheConfig& config);
   ~DnsCacheImpl() override;
 
@@ -48,6 +48,7 @@ class DnsCacheImpl : public DnsCache, Logger::Loggable<Logger::Id::forward_proxy
                                             LoadDnsCacheEntryCallbacks& callbacks) override;
   AddUpdateCallbacksHandlePtr addUpdateCallbacks(UpdateCallbacks& callbacks) override;
   absl::flat_hash_map<std::string, DnsHostInfoSharedPtr> hosts() override;
+  DnsCacheResourceManager& dnsCacheResourceManager() override { return resource_manager_; }
 
 private:
   using TlsHostMap = absl::flat_hash_map<std::string, DnsHostInfoSharedPtr>;
@@ -138,6 +139,7 @@ class DnsCacheImpl : public DnsCache, Logger::Loggable<Logger::Id::forward_proxy
   DnsCacheStats stats_;
   std::list<AddUpdateCallbacksHandleImpl*> update_callbacks_;
   absl::flat_hash_map<std::string, PrimaryHostInfoPtr> primary_hosts_;
+  DnsCacheResourceManager resource_manager_;
   const std::chrono::milliseconds refresh_interval_;
   const BackOffStrategyPtr failure_backoff_strategy_;
   const std::chrono::milliseconds host_ttl_;

source/extensions/common/dynamic_forward_proxy/dns_cache_manager_impl.cc

@@ -27,21 +27,21 @@ DnsCacheSharedPtr DnsCacheManagerImpl::getCache(
     return existing_cache->second.cache_;
   }
 
-  DnsCacheSharedPtr new_cache =
-      std::make_shared<DnsCacheImpl>(main_thread_dispatcher_, tls_, random_, root_scope_, config);
+  DnsCacheSharedPtr new_cache = std::make_shared<DnsCacheImpl>(
+      main_thread_dispatcher_, tls_, random_, loader_, root_scope_, config);
   caches_.emplace(config.name(), ActiveCache{config, new_cache});
   return new_cache;
 }
 
 DnsCacheManagerSharedPtr getCacheManager(Singleton::Manager& singleton_manager,
                                          Event::Dispatcher& main_thread_dispatcher,
                                          ThreadLocal::SlotAllocator& tls,
-                                         Runtime::RandomGenerator& random,
+                                         Runtime::RandomGenerator& random, Runtime::Loader& loader,
                                          Stats::Scope& root_scope) {
   return singleton_manager.getTyped<DnsCacheManager>(
       SINGLETON_MANAGER_REGISTERED_NAME(dns_cache_manager),
-      [&main_thread_dispatcher, &tls, &random, &root_scope] {
-        return std::make_shared<DnsCacheManagerImpl>(main_thread_dispatcher, tls, random,
+      [&main_thread_dispatcher, &tls, &random, &loader, &root_scope] {
+        return std::make_shared<DnsCacheManagerImpl>(main_thread_dispatcher, tls, random, loader,
                                                      root_scope);
       });
 }

source/extensions/common/dynamic_forward_proxy/dns_cache_manager_impl.h

@@ -14,9 +14,10 @@ namespace DynamicForwardProxy {
 class DnsCacheManagerImpl : public DnsCacheManager, public Singleton::Instance {
 public:
   DnsCacheManagerImpl(Event::Dispatcher& main_thread_dispatcher, ThreadLocal::SlotAllocator& tls,
-                      Runtime::RandomGenerator& random, Stats::Scope& root_scope)
+                      Runtime::RandomGenerator& random, Runtime::Loader& loader,
+                      Stats::Scope& root_scope)
       : main_thread_dispatcher_(main_thread_dispatcher), tls_(tls), random_(random),
-        root_scope_(root_scope) {}
+        loader_(loader), root_scope_(root_scope) {}
 
   // DnsCacheManager
   DnsCacheSharedPtr getCache(
@@ -35,6 +36,7 @@ class DnsCacheManagerImpl : public DnsCacheManager, public Singleton::Instance {
   Event::Dispatcher& main_thread_dispatcher_;
   ThreadLocal::SlotAllocator& tls_;
   Runtime::RandomGenerator& random_;
+  Runtime::Loader& loader_;
   Stats::Scope& root_scope_;
   absl::flat_hash_map<std::string, ActiveCache> caches_;
 };
@@ -43,19 +45,20 @@ class DnsCacheManagerFactoryImpl : public DnsCacheManagerFactory {
 public:
   DnsCacheManagerFactoryImpl(Singleton::Manager& singleton_manager, Event::Dispatcher& dispatcher,
                              ThreadLocal::SlotAllocator& tls, Runtime::RandomGenerator& random,
-                             Stats::Scope& root_scope)
+                             Runtime::Loader& loader, Stats::Scope& root_scope)
       : singleton_manager_(singleton_manager), dispatcher_(dispatcher), tls_(tls), random_(random),
-        root_scope_(root_scope) {}
+        loader_(loader), root_scope_(root_scope) {}
 
   DnsCacheManagerSharedPtr get() override {
-    return getCacheManager(singleton_manager_, dispatcher_, tls_, random_, root_scope_);
+    return getCacheManager(singleton_manager_, dispatcher_, tls_, random_, loader_, root_scope_);
   }
 
 private:
   Singleton::Manager& singleton_manager_;
   Event::Dispatcher& dispatcher_;
   ThreadLocal::SlotAllocator& tls_;
   Runtime::RandomGenerator& random_;
+  Runtime::Loader& loader_;
   Stats::Scope& root_scope_;
 };
 

source/extensions/common/dynamic_forward_proxy/dns_cache_resource_manager.cc

@@ -0,0 +1,24 @@
+#include "extensions/common/dynamic_forward_proxy/dns_cache_resource_manager.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Common {
+namespace DynamicForwardProxy {
+
+DnsCacheResourceManager::DnsCacheResourceManager(
+    Runtime::Loader& loader,
+    const envoy::extensions::common::dynamic_forward_proxy::v3::DnsCacheConfig& config) {
+  uint32_t max_requests = 1024;
+
+  if (config.has_dns_failure_refresh_rate()) {
+    const auto& threshold = config.dns_cache_circuit_breaker().threshold();
+    max_requests = threshold.max_pending_requests().value();
+  }
+  const auto runtime_key = fmt::format("circuit_breaker.dns_cache.{}", config.name());
+
+  pending_requests_ = std::make_unique<DnsResource>(max_requests, loader, runtime_key);
+}
+} // namespace DynamicForwardProxy
+} // namespace Common
+} // namespace Extensions
+} // namespace Envoy

source/extensions/common/dynamic_forward_proxy/dns_cache_resource_manager.h

@@ -0,0 +1,56 @@
+#pragma once
+
+#include <atomic>
+#include <memory>
+#include <string>
+
+#include "envoy/extensions/common/dynamic_forward_proxy/v3/dns_cache.pb.h"
+#include "envoy/runtime/runtime.h"
+#include "envoy/upstream/resource_manager.h"
+
+#include "common/common/assert.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Common {
+namespace DynamicForwardProxy {
+
+class DnsResource : public Envoy::Upstream::Resource {
+public:
+  DnsResource(uint64_t max, Runtime::Loader& runtime, const std::string& runtime_key)
+      : max_(max), runtime_(runtime), runtime_key_(runtime_key) {}
+  ~DnsResource() override { ASSERT(current_ == 0); }
+
+  // Envoy::Upstream::Resource
+  bool canCreate() override { return current_ < max(); }
+  void inc() override { current_++; }
+  void dec() override { decBy(1); }
+  void decBy(uint64_t amount) override {
+    ASSERT(current_ >= amount);
+    current_ -= amount;
+  }
+  uint64_t max() override { return runtime_.snapshot().getInteger(runtime_key_, max_); }
+  uint64_t count() const override { return current_.load(); }
+
+private:
+  uint64_t max_;
+  std::atomic<uint64_t> current_{};
+  Runtime::Loader& runtime_;
+  const std::string runtime_key_;
+};
+
+class DnsCacheResourceManager {
+public:
+  DnsCacheResourceManager(
+      Runtime::Loader& loader,
+      const envoy::extensions::common::dynamic_forward_proxy::v3::DnsCacheConfig& config);
+  DnsResource& pendingRequests() { return *pending_requests_; }
+
+private:
+  std::unique_ptr<DnsResource> pending_requests_;
+};
+
+} // namespace DynamicForwardProxy
+} // namespace Common
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/http/dynamic_forward_proxy/config.cc

@@ -16,7 +16,7 @@ Http::FilterFactoryCb DynamicForwardProxyFilterFactory::createFilterFactoryFromP
     const std::string&, Server::Configuration::FactoryContext& context) {
   Extensions::Common::DynamicForwardProxy::DnsCacheManagerFactoryImpl cache_manager_factory(
       context.singletonManager(), context.dispatcher(), context.threadLocal(), context.random(),
-      context.scope());
+      context.runtime(), context.scope());
   ProxyFilterConfigSharedPtr filter_config(std::make_shared<ProxyFilterConfig>(
       proto_config, cache_manager_factory, context.clusterManager()));
   return [filter_config](Http::FilterChainFactoryCallbacks& callbacks) -> void {