make grpcmux pause/resume come in pair by returning a RAII obj which resumes requests on destruction.

include/envoy/config/BUILD

@@ -35,6 +35,7 @@ envoy_cc_library(
     deps = [
         ":subscription_interface",
         "//include/envoy/stats:stats_macros",
+        "//source/common/common:cleanup_lib",
         "//source/common/protobuf",
     ],
 )

include/envoy/config/grpc_mux.h

@@ -5,11 +5,13 @@
 #include "envoy/config/subscription.h"
 #include "envoy/stats/stats_macros.h"
 
+#include "common/common/cleanup.h"
 #include "common/protobuf/protobuf.h"
 
 namespace Envoy {
 namespace Config {
 
+using ScopedResume = std::unique_ptr<Cleanup>;
 /**
  * All control plane related stats. @see stats_macros.h
  */
@@ -62,32 +64,23 @@ class GrpcMux {
    * requests may later be resumed with resume().
    * @param type_url type URL corresponding to xDS API, e.g.
    * type.googleapis.com/envoy.api.v2.Cluster.
+   *
+   * @return a ScopedResume object, which when destructed, resumes the paused discovery requests.
+   * A discovery request will be sent if one would have been sent during the pause.
    */
-  virtual void pause(const std::string& type_url) PURE;
+  ABSL_MUST_USE_RESULT virtual ScopedResume pause(const std::string& type_url) PURE;
 
   /**
    * Pause discovery requests for given API types. This is useful when we're processing an update
    * for LDS or CDS and don't want a flood of updates for RDS or EDS respectively. Discovery
    * requests may later be resumed with resume().
    * @param type_urls type URLs corresponding to xDS API, e.g.
    * type.googleapis.com/envoy.api.v2.Cluster.
+   *
+   * @return a ScopedResume object, which when destructed, resumes the paused discovery requests.
+   * A discovery request will be sent if one would have been sent during the pause.
    */
-  virtual void pause(const std::vector<std::string> type_urls) PURE;
-
-  /**
-   * Resume discovery requests for a given API type. This will send a discovery request if one would
-   * have been sent during the pause.
-   * @param type_url type URL corresponding to xDS API e.g. type.googleapis.com/envoy.api.v2.Cluster
-   */
-  virtual void resume(const std::string& type_url) PURE;
-
-  /**
-   * Resume discovery requests for given API types. This will send a discovery request if one would
-   * have been sent during the pause.
-   * @param type_urls type URLs corresponding to xDS API e.g.
-   * type.googleapis.com/envoy.api.v2.Cluster
-   */
-  virtual void resume(const std::vector<std::string> type_urls) PURE;
+  ABSL_MUST_USE_RESULT virtual ScopedResume pause(const std::vector<std::string> type_urls) PURE;
 
   /**
    * Retrieves the current pause state as set by pause()/resume().

source/common/config/BUILD

@@ -139,6 +139,7 @@ envoy_cc_library(
         "//include/envoy/config:grpc_mux_interface",
         "//include/envoy/config:subscription_interface",
         "//include/envoy/upstream:cluster_manager_interface",
+        "//source/common/common:cleanup_lib",
         "//source/common/common:minimal_logger_lib",
         "//source/common/common:utility_lib",
         "//source/common/memory:utils_lib",

source/common/config/grpc_mux_impl.cc

@@ -99,37 +99,32 @@ GrpcMuxWatchPtr GrpcMuxImpl::addWatch(const std::string& type_url,
   return watch;
 }
 
-void GrpcMuxImpl::pause(const std::string& type_url) {
-  ENVOY_LOG(debug, "Pausing discovery requests for {}", type_url);
-  ApiState& api_state = api_state_[type_url];
-  ASSERT(!api_state.paused_);
-  ASSERT(!api_state.pending_);
-  api_state.paused_ = true;
+ScopedResume GrpcMuxImpl::pause(const std::string& type_url) {
+  return pause(std::vector<std::string>{type_url});
 }
 
-void GrpcMuxImpl::pause(const std::vector<std::string> type_urls) {
+ScopedResume GrpcMuxImpl::pause(const std::vector<std::string> type_urls) {
   for (const auto& type_url : type_urls) {
-    pause(type_url);
-  }
-}
-
-void GrpcMuxImpl::resume(const std::string& type_url) {
-  ENVOY_LOG(debug, "Resuming discovery requests for {}", type_url);
-  ApiState& api_state = api_state_[type_url];
-  ASSERT(api_state.paused_);
-  api_state.paused_ = false;
-
-  if (api_state.pending_) {
-    ASSERT(api_state.subscribed_);
-    queueDiscoveryRequest(type_url);
-    api_state.pending_ = false;
+    ENVOY_LOG(debug, "Pausing discovery requests for {}", type_url);
+    ApiState& api_state = api_state_[type_url];
+    ASSERT(!api_state.paused_);
+    ASSERT(!api_state.pending_);
+    api_state.paused_ = true;
   }
-}
+  return std::make_unique<Cleanup>([this, type_urls]() {
+    for (const auto& type_url : type_urls) {
+      ENVOY_LOG(debug, "Resuming discovery requests for {}", type_url);
+      ApiState& api_state = api_state_[type_url];
+      ASSERT(api_state.paused_);
+      api_state.paused_ = false;
 
-void GrpcMuxImpl::resume(const std::vector<std::string> type_urls) {
-  for (const auto& type_url : type_urls) {
-    resume(type_url);
-  }
+      if (api_state.pending_) {
+        ASSERT(api_state.subscribed_);
+        queueDiscoveryRequest(type_url);
+        api_state.pending_ = false;
+      }
+    }
+  });
 }
 
 bool GrpcMuxImpl::paused(const std::string& type_url) const {

source/common/config/grpc_mux_impl.h

@@ -20,7 +20,6 @@
 
 namespace Envoy {
 namespace Config {
-
 /**
  * ADS API implementation that fetches via gRPC.
  */
@@ -38,10 +37,8 @@ class GrpcMuxImpl : public GrpcMux,
   void start() override;
 
   // GrpcMux
-  void pause(const std::string& type_url) override;
-  void pause(const std::vector<std::string> type_urls) override;
-  void resume(const std::string& type_url) override;
-  void resume(const std::vector<std::string> type_urls) override;
+  ScopedResume pause(const std::string& type_url) override;
+  ScopedResume pause(const std::vector<std::string> type_urls) override;
   bool paused(const std::string& type_url) const override;
   bool paused(const std::vector<std::string> type_urls) const override;
 
@@ -147,10 +144,12 @@ class NullGrpcMuxImpl : public GrpcMux,
                         GrpcStreamCallbacks<envoy::service::discovery::v3::DiscoveryResponse> {
 public:
   void start() override {}
-  void pause(const std::string&) override {}
-  void pause(const std::vector<std::string>) override {}
-  void resume(const std::string&) override {}
-  void resume(const std::vector<std::string>) override {}
+  ScopedResume pause(const std::string&) override {
+    return std::make_unique<Cleanup>([] {});
+  }
+  ScopedResume pause(const std::vector<std::string>) override {
+    return std::make_unique<Cleanup>([] {});
+  }
   bool paused(const std::string&) const override { return false; }
   bool paused(const std::vector<std::string>) const override { return false; }
 

source/common/config/grpc_subscription_impl.cc

@@ -105,9 +105,7 @@ void GrpcSubscriptionImpl::onConfigUpdateFailed(ConfigUpdateFailureReason reason
   stats_.update_attempt_.inc();
 }
 
-void GrpcSubscriptionImpl::pause() { grpc_mux_->pause(type_url_); }
-
-void GrpcSubscriptionImpl::resume() { grpc_mux_->resume(type_url_); }
+ScopedResume GrpcSubscriptionImpl::pause() { return grpc_mux_->pause(type_url_); }
 
 void GrpcSubscriptionImpl::disableInitFetchTimeoutTimer() {
   if (init_fetch_timeout_timer_) {

source/common/config/grpc_subscription_impl.h

@@ -35,8 +35,7 @@ class GrpcSubscriptionImpl : public Subscription,
 
   GrpcMuxSharedPtr grpcMux() { return grpc_mux_; }
 
-  void pause();
-  void resume();
+  ScopedResume pause();
 
 private:
   void disableInitFetchTimeoutTimer();

source/common/config/new_grpc_mux_impl.cc

@@ -25,23 +25,21 @@ NewGrpcMuxImpl::NewGrpcMuxImpl(Grpc::RawAsyncClientPtr&& async_client,
                    rate_limit_settings),
       local_info_(local_info), transport_api_version_(transport_api_version) {}
 
-void NewGrpcMuxImpl::pause(const std::string& type_url) { pausable_ack_queue_.pause(type_url); }
-
-void NewGrpcMuxImpl::pause(const std::vector<std::string> type_urls) {
-  for (const auto& type_url : type_urls) {
-    pause(type_url);
-  }
-}
-
-void NewGrpcMuxImpl::resume(const std::string& type_url) {
-  pausable_ack_queue_.resume(type_url);
-  trySendDiscoveryRequests();
+ScopedResume NewGrpcMuxImpl::pause(const std::string& type_url) {
+  return pause(std::vector<std::string>{type_url});
 }
 
-void NewGrpcMuxImpl::resume(const std::vector<std::string> type_urls) {
+ScopedResume NewGrpcMuxImpl::pause(const std::vector<std::string> type_urls) {
   for (const auto& type_url : type_urls) {
-    resume(type_url);
+    pausable_ack_queue_.pause(type_url);
   }
+
+  return std::make_unique<Cleanup>([this, type_urls]() {
+    for (const auto& type_url : type_urls) {
+      pausable_ack_queue_.resume(type_url);
+      trySendDiscoveryRequests();
+    }
+  });
 }
 
 bool NewGrpcMuxImpl::paused(const std::string& type_url) const {

source/common/config/new_grpc_mux_impl.h

@@ -38,10 +38,8 @@ class NewGrpcMuxImpl
                            SubscriptionCallbacks& callbacks,
                            OpaqueResourceDecoder& resource_decoder) override;
 
-  void pause(const std::string& type_url) override;
-  void pause(const std::vector<std::string> type_urls) override;
-  void resume(const std::string& type_url) override;
-  void resume(const std::vector<std::string> type_urls) override;
+  ScopedResume pause(const std::string& type_url) override;
+  ScopedResume pause(const std::vector<std::string> type_urls) override;
   bool paused(const std::string& type_url) const override;
   bool paused(const std::vector<std::string> type_urls) const override;
 

source/common/router/scoped_rds.cc

@@ -28,6 +28,7 @@ using Envoy::Config::ConfigProvider;
 using Envoy::Config::ConfigProviderInstanceType;
 using Envoy::Config::ConfigProviderManager;
 using Envoy::Config::ConfigProviderPtr;
+using Envoy::Config::ScopedResume;
 
 namespace Envoy {
 namespace Router {
@@ -225,40 +226,41 @@ void ScopedRdsConfigSubscription::onConfigUpdate(
   // NOTE: deletes are done before adds/updates.
 
   absl::flat_hash_map<std::string, ScopedRouteInfoConstSharedPtr> to_be_removed_scopes;
+  // Destruction of resume_rds will lift the floodgate for new RDS subscriptions.
+  // Note in the case of partial acceptance, accepted RDS subscriptions should be started
+  // despite of any error.
+  ScopedResume resume_rds;
   // If new route config sources come after the local init manager's initialize() been
   // called, the init manager can't accept new targets. Instead we use a local override which will
   // start new subscriptions but not wait on them to be ready.
-  std::unique_ptr<Init::ManagerImpl> noop_init_manager;
-  // NOTE: This should be defined after noop_init_manager as it depends on the
-  // noop_init_manager.
-  std::unique_ptr<Cleanup> resume_rds;
+  std::unique_ptr<Init::ManagerImpl> srds_init_mgr;
+  // NOTE: This should be defined after srds_init_mgr and resume_rds, as it depends on the
+  // srds_init_mgr, and we want a single RDS discovery request to be sent to management
+  // server.
+  std::unique_ptr<Cleanup> srds_initialization_continuation;
+  ASSERT(localInitManager().state() > Init::Manager::State::Uninitialized);
+  const auto type_urls =
+      Envoy::Config::getAllVersionTypeUrls<envoy::config::route::v3::RouteConfiguration>();
+  // Pause RDS to not send a burst of RDS requests until we start all the new subscriptions.
+  // In the case that localInitManager is uninitialized, RDS is already paused
+  // either by Server init or LDS init.
+  if (factory_context_.clusterManager().adsMux()) {
+    resume_rds = factory_context_.clusterManager().adsMux()->pause(type_urls);
+  }
   // if local init manager is initialized, the parent init manager may have gone away.
   if (localInitManager().state() == Init::Manager::State::Initialized) {
-    const auto type_urls =
-        Envoy::Config::getAllVersionTypeUrls<envoy::config::route::v3::RouteConfiguration>();
-    noop_init_manager =
+    srds_init_mgr =
         std::make_unique<Init::ManagerImpl>(fmt::format("SRDS {}:{}", name_, version_info));
-    // Pause RDS to not send a burst of RDS requests until we start all the new subscriptions.
-    // In the case if factory_context_.init_manager() is uninitialized, RDS is already paused
-    // either by Server init or LDS init.
-    if (factory_context_.clusterManager().adsMux()) {
-      factory_context_.clusterManager().adsMux()->pause(type_urls);
-    }
-    resume_rds = std::make_unique<Cleanup>([this, &noop_init_manager, version_info, type_urls] {
-      // For new RDS subscriptions created after listener warming up, we don't wait for them to
-      // warm up.
-      Init::WatcherImpl noop_watcher(
-          // Note: we just throw it away.
-          fmt::format("SRDS ConfigUpdate watcher {}:{}", name_, version_info),
-          []() { /*Do nothing.*/ });
-      noop_init_manager->initialize(noop_watcher);
-      // New RDS subscriptions should have been created, now lift the floodgate.
-      // Note in the case of partial acceptance, accepted RDS subscriptions should be started
-      // despite of any error.
-      if (factory_context_.clusterManager().adsMux()) {
-        factory_context_.clusterManager().adsMux()->resume(type_urls);
-      }
-    });
+    srds_initialization_continuation =
+        std::make_unique<Cleanup>([this, &srds_init_mgr, version_info] {
+          // For new RDS subscriptions created after listener warming up, we don't wait for them to
+          // warm up.
+          Init::WatcherImpl noop_watcher(
+              // Note: we just throw it away.
+              fmt::format("SRDS ConfigUpdate watcher {}:{}", name_, version_info),
+              []() { /*Do nothing.*/ });
+          srds_init_mgr->initialize(noop_watcher);
+        });
   }
 
   std::vector<std::string> exception_msgs;
@@ -268,7 +270,7 @@ void ScopedRdsConfigSubscription::onConfigUpdate(
       to_be_removed_rds_providers = removeScopes(removed_resources, version_info);
   bool any_applied =
       addOrUpdateScopes(added_resources,
-                        (noop_init_manager == nullptr ? localInitManager() : *noop_init_manager),
+                        (srds_init_mgr == nullptr ? localInitManager() : *srds_init_mgr),
                         version_info, exception_msgs) ||
       !to_be_removed_rds_providers.empty();
   ConfigSubscriptionCommonBase::onConfigUpdate();

source/common/upstream/cds_api_impl.cc

@@ -52,13 +52,11 @@ void CdsApiImpl::onConfigUpdate(const std::vector<Config::DecodedResourceRef>& r
 void CdsApiImpl::onConfigUpdate(const std::vector<Config::DecodedResourceRef>& added_resources,
                                 const Protobuf::RepeatedPtrField<std::string>& removed_resources,
                                 const std::string& system_version_info) {
-  std::unique_ptr<Cleanup> maybe_eds_resume;
+  Config::ScopedResume maybe_resume_eds;
   if (cm_.adsMux()) {
     const auto type_urls =
         Config::getAllVersionTypeUrls<envoy::config::endpoint::v3::ClusterLoadAssignment>();
-    cm_.adsMux()->pause(type_urls);
-    maybe_eds_resume =
-        std::make_unique<Cleanup>([this, type_urls] { cm_.adsMux()->resume(type_urls); });
+    maybe_resume_eds = cm_.adsMux()->pause(type_urls);
   }
 
   ENVOY_LOG(info, "cds: add {} cluster(s), remove {} cluster(s)", added_resources.size(),

source/common/upstream/cluster_manager_impl.cc

@@ -155,14 +155,12 @@ void ClusterManagerInitHelper::maybeFinishInitialize() {
       // If the first CDS response doesn't have any primary cluster, ClusterLoadAssignment
       // should be already paused by CdsApiImpl::onConfigUpdate(). Need to check that to
       // avoid double pause ClusterLoadAssignment.
-      std::unique_ptr<Cleanup> maybe_eds_resume;
+      Config::ScopedResume maybe_resume_eds;
       if (cm_.adsMux()) {
         const auto type_urls =
             Config::getAllVersionTypeUrls<envoy::config::endpoint::v3::ClusterLoadAssignment>();
         if (!cm_.adsMux()->paused(type_urls)) {
-          cm_.adsMux()->pause(type_urls);
-          maybe_eds_resume =
-              std::make_unique<Cleanup>([this, type_urls] { cm_.adsMux()->resume(type_urls); });
+          maybe_resume_eds = cm_.adsMux()->pause(type_urls);
         }
       }
       initializeSecondaryClusters();
@@ -805,9 +803,10 @@ void ClusterManagerImpl::updateClusterCounts() {
     const auto type_urls = Config::getAllVersionTypeUrls<envoy::config::cluster::v3::Cluster>();
     const uint64_t previous_warming = cm_stats_.warming_clusters_.value();
     if (previous_warming == 0 && !warming_clusters_.empty()) {
-      ads_mux_->pause(type_urls);
+      resume_cds_ = ads_mux_->pause(type_urls);
     } else if (previous_warming > 0 && warming_clusters_.empty()) {
-      ads_mux_->resume(type_urls);
+      ASSERT(resume_cds_ != nullptr);
+      resume_cds_.reset();
     }
   }
   cm_stats_.active_clusters_.set(active_clusters_.size());

source/common/upstream/cluster_manager_impl.h

@@ -238,6 +238,9 @@ class ClusterManagerImpl : public ClusterManager, Logger::Loggable<Logger::Id::u
   Http::AsyncClient& httpAsyncClientForCluster(const std::string& cluster) override;
   bool removeCluster(const std::string& cluster) override;
   void shutdown() override {
+    if (resume_cds_ != nullptr) {
+      resume_cds_->cancel();
+    }
     // Make sure we destroy all potential outgoing connections before this returns.
     cds_api_.reset();
     ads_mux_.reset();
@@ -498,6 +501,8 @@ class ClusterManagerImpl : public ClusterManager, Logger::Loggable<Logger::Id::u
   ClusterManagerStats cm_stats_;
   ClusterManagerInitHelper init_helper_;
   Config::GrpcMuxSharedPtr ads_mux_;
+  // Temporarily saved resume cds callback from updateClusterCounts invocation.
+  Config::ScopedResume resume_cds_;
   LoadStatsReporterPtr load_stats_reporter_;
   // The name of the local cluster of this Envoy instance if defined.
   absl::optional<std::string> local_cluster_name_;