dynamic_forward_proxy: DNS Cache circuit breaker

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

@@ -18,9 +18,16 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 
 // [#protodoc-title: Dynamic forward proxy common configuration]
 
+// Configuration of circuit breakers for resolver.
+message DnsCacheCircuitBreakers {
+  // 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;
+}
+
 // 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 +90,7 @@ 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;
+
+  // The config of circuit breakers for resolver. It provides a configurable threshold.
+  DnsCacheCircuitBreakers dns_cache_circuit_breaker = 7;
 }

docs/root/version_history/current.rst

@@ -8,6 +8,7 @@ Changes
 * access loggers: applied existing buffer limits to the non-google gRPC access logs, as well as :ref:`stats <config_access_log_stats>` for logged / dropped logs.
 * access loggers: extened specifier for FilterStateFormatter to output :ref:`unstructured log string <config_access_log_format_filter_state>`.
 * dynamic forward proxy: added :ref:`SNI based dynamic forward proxy <config_network_filters_sni_dynamic_forward_proxy>` support.
+* dynamic forward proxy: added configurable circuit breakers for resolver on DNS cache.
 * fault: added support for controlling the percentage of requests that abort, delay and response rate limits faults
   are applied to using :ref:`HTTP headers <config_http_filters_fault_injection_http_header>` to the HTTP fault filter.
 * fault: added support for specifying grpc_status code in abort faults using

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,16 @@ 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 = [
+        "//include/envoy/stats:stats_interface",
+        "//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 A pointer to resource manager for dns cache.
+   */
+  virtual DnsCacheResourceManagerPtr& 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,7 +16,7 @@ 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())),
@@ -32,6 +32,12 @@ DnsCacheImpl::DnsCacheImpl(
       max_hosts_(PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, max_hosts, 1024)) {
   tls_slot_->set([](Event::Dispatcher&) { return std::make_shared<ThreadLocalHostInfo>(); });
   updateTlsHostsMap();
+
+  if (config.has_dns_cache_circuit_breaker()) {
+    auto cb_stats = generateDnsCacheCircuitBreakersStats(*scope_);
+    resource_manager_ = std::make_unique<DnsCacheResourceManager>(
+        cb_stats, loader, config.name(), config.dns_cache_circuit_breaker());
+  }
 }
 
 DnsCacheImpl::~DnsCacheImpl() {
@@ -46,6 +52,13 @@ DnsCacheImpl::~DnsCacheImpl() {
   }
 }
 
+DnsCacheCircuitBreakersStats
+DnsCacheImpl::generateDnsCacheCircuitBreakersStats(Stats::Scope& scope) {
+  std::string stat_prefix = "circuit_breakers";
+  return {ALL_DNS_CACHE_CIRCUIT_BREAKERS_STATS(POOL_GAUGE_PREFIX(scope, stat_prefix),
+                                               POOL_GAUGE_PREFIX(scope, stat_prefix))};
+}
+
 DnsCacheImpl::LoadDnsCacheEntryResult
 DnsCacheImpl::loadDnsCacheEntry(absl::string_view host, uint16_t default_port,
                                 LoadDnsCacheEntryCallbacks& callbacks) {

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;
+  DnsCacheResourceManagerPtr& dnsCacheResourceManager() override { return resource_manager_; }
 
 private:
   using TlsHostMap = absl::flat_hash_map<std::string, DnsHostInfoSharedPtr>;
@@ -121,6 +122,7 @@ class DnsCacheImpl : public DnsCache, Logger::Loggable<Logger::Id::forward_proxy
     UpdateCallbacks& callbacks_;
   };
 
+  DnsCacheCircuitBreakersStats generateDnsCacheCircuitBreakersStats(Stats::Scope& scope);
   void startCacheLoad(const std::string& host, uint16_t default_port);
   void startResolve(const std::string& host, PrimaryHostInfo& host_info);
   void finishResolve(const std::string& host, Network::DnsResolver::ResolutionStatus status,
@@ -138,6 +140,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_;
+  DnsCacheResourceManagerPtr 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,22 @@
+#include "extensions/common/dynamic_forward_proxy/dns_cache_resource_manager.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Common {
+namespace DynamicForwardProxy {
+
+DnsCacheResourceManager::DnsCacheResourceManager(
+    DnsCacheCircuitBreakersStats& cb_stats, Runtime::Loader& loader, const std::string& config_name,
+    const envoy::extensions::common::dynamic_forward_proxy::v3::DnsCacheCircuitBreakers&
+        cb_config) {
+  const auto& max_requests = cb_config.max_pending_requests().value();
+  const auto runtime_key = fmt::format("dns_cache.{}.circuit_breakers", config_name);
+
+  pending_requests_ =
+      std::make_unique<DnsResource>(max_requests, loader, runtime_key, cb_stats.rq_pending_opening_,
+                                    cb_stats.rq_pending_remaining_);
+}
+} // namespace DynamicForwardProxy
+} // namespace Common
+} // namespace Extensions
+} // namespace Envoy

source/extensions/common/dynamic_forward_proxy/dns_cache_resource_manager.h

@@ -0,0 +1,97 @@
+#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/stats/scope.h"
+#include "envoy/stats/stats_macros.h"
+#include "envoy/upstream/resource_manager.h"
+
+#include "common/common/assert.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Common {
+namespace DynamicForwardProxy {
+
+#define ALL_DNS_CACHE_CIRCUIT_BREAKERS_STATS(OPEN_GAUGE, REMAINING_GAUGE)                          \
+  OPEN_GAUGE(rq_pending_opening, Accumulate)                                                       \
+  REMAINING_GAUGE(rq_pending_remaining, Accumulate)
+
+struct DnsCacheCircuitBreakersStats {
+  ALL_DNS_CACHE_CIRCUIT_BREAKERS_STATS(GENERATE_GAUGE_STRUCT, GENERATE_GAUGE_STRUCT)
+};
+
+class DnsResource : public Envoy::Upstream::Resource {
+public:
+  DnsResource(uint64_t max, Runtime::Loader& runtime, const std::string& runtime_key,
+              Stats::Gauge& opening, Stats::Gauge& remaining)
+      : max_(max), runtime_(runtime), opening_(opening), remaining_(remaining),
+        runtime_key_(runtime_key) {
+    remaining_.set(max);
+  }
+  ~DnsResource() override { ASSERT(current_ == 0); }
+
+  // Envoy::Upstream::Resource
+  bool canCreate() override { return current_ < max(); }
+  void inc() override {
+    current_++;
+    std::cout << "inc" << std::endl;
+    std::cout << max_ << std::endl;
+    std::cout << current_ << std::endl;
+    remaining_.set(max() > current_ ? max() - current_ : 0);
+    opening_.set(max() > current_ ? 0 : 1);
+  }
+
+  void dec() override { decBy(1); }
+
+  void decBy(uint64_t amount) override {
+    ASSERT(current_ >= amount);
+    current_ -= amount;
+    std::cout << "dec" << std::endl;
+    std::cout << max_ << std::endl;
+    std::cout << current_ << std::endl;
+    remaining_.set(max() > current_ ? max() - current_ : 0);
+    opening_.set(max() > current_ ? 0 : 1);
+  }
+
+  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_;
+  Stats::Gauge& opening_;
+  Stats::Gauge& remaining_;
+  const std::string runtime_key_;
+};
+
+class DnsCacheResourceManager : public Envoy::Upstream::ResourceManager {
+public:
+  DnsCacheResourceManager(
+      DnsCacheCircuitBreakersStats& cb_stats, Runtime::Loader& loader,
+      const std::string& config_name,
+      const envoy::extensions::common::dynamic_forward_proxy::v3::DnsCacheCircuitBreakers&
+          cb_config);
+
+  // Envoy::Upstream::ResourceManager
+  Envoy::Upstream::Resource& pendingRequests() override { return *pending_requests_; }
+  Envoy::Upstream::Resource& connections() override { NOT_REACHED_GCOVR_EXCL_LINE; }
+  Envoy::Upstream::Resource& requests() override { NOT_REACHED_GCOVR_EXCL_LINE; }
+  Envoy::Upstream::Resource& retries() override { NOT_REACHED_GCOVR_EXCL_LINE; }
+  Envoy::Upstream::Resource& connectionPools() override { NOT_REACHED_GCOVR_EXCL_LINE; }
+
+private:
+  std::unique_ptr<DnsResource> pending_requests_;
+};
+
+using DnsCacheResourceManagerPtr = std::unique_ptr<DnsCacheResourceManager>;
+
+} // 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 {