Add option for merging cluster updates

api/envoy/api/v2/cds.proto

@@ -432,6 +432,13 @@ message Cluster {
       ZoneAwareLbConfig zone_aware_lb_config = 2;
       LocalityWeightedLbConfig locality_weighted_lb_config = 3;
     }
+    // If set, membership and healthcheck updates that happen within this duration will be coalesced
+    // and delivered in one shot when the duration expires. The start of the duration is when the
+    // first update happens. This is useful for big clusters, with potentially noisy deploys that
+    // might trigger excessive CPU usage due to a constant stream of healthcheck state changes or
+    // membership updates. By default, this is not configured and updates apply immediately.
+    // Also, the first set of updates to be seen apply immediately as well (e.g.: a new cluster).
+    google.protobuf.Duration update_merge_window = 4;
   }
 
   // Common configuration for all load balancer implementations.

docs/root/intro/version_history.rst

@@ -48,6 +48,8 @@ Version history
 * upstream: added configuration option to the subset load balancer to take locality weights into account when
   selecting a host from a subset.
 * access log: added RESPONSE_DURATION and RESPONSE_TX_DURATION.
+* cluster: added :ref:`option <envoy_api_field_Cluster.CommonLbConfig.update_merge_window>` to coalesce updates
+  within the given duration.
 
 1.7.0
 ===============

source/common/upstream/cluster_manager_impl.cc

@@ -180,7 +180,7 @@ ClusterManagerImpl::ClusterManagerImpl(const envoy::config::bootstrap::v2::Boots
       init_helper_([this](Cluster& cluster) { onClusterInit(cluster); }),
       config_tracker_entry_(
           admin.getConfigTracker().add("clusters", [this] { return dumpClusterConfigs(); })),
-      system_time_source_(system_time_source) {
+      system_time_source_(system_time_source), dispatcher_(main_thread_dispatcher) {
   async_client_manager_ = std::make_unique<Grpc::AsyncClientManagerImpl>(*this, tls);
   const auto& cm_config = bootstrap.cluster_manager();
   if (cm_config.has_outlier_detection()) {
@@ -330,7 +330,17 @@ void ClusterManagerImpl::onClusterInit(Cluster& cluster) {
                                                            const HostVector& hosts_removed) {
     // This fires when a cluster is about to have an updated member set. We need to send this
     // out to all of the thread local configurations.
-    postThreadLocalClusterUpdate(cluster, priority, hosts_added, hosts_removed);
+
+    // Should we coalesce updates?
+    bool scheduled = false;
+    if (cluster.info()->lbConfig().has_update_merge_window()) {
+      scheduled = scheduleUpdate(cluster, priority, hosts_added, hosts_removed);
+    }
+
+    // If an update was not scheduled, deliver it immediately.
+    if (!scheduled) {
+      postThreadLocalClusterUpdate(cluster, priority, hosts_added, hosts_removed);
+    }
   });
 
   // Finally, if the cluster has any hosts, post updates cross-thread so the per-thread load
@@ -343,6 +353,80 @@ void ClusterManagerImpl::onClusterInit(Cluster& cluster) {
   }
 }
 
+bool ClusterManagerImpl::scheduleUpdate(const Cluster& cluster, uint32_t priority,
+                                        const HostVector& hosts_added,
+                                        const HostVector& hosts_removed) {
+  const auto& update_merge_window = cluster.info()->lbConfig().update_merge_window();
+  const auto timeout = DurationUtil::durationToMilliseconds(update_merge_window);
+  PendingUpdatesByPriorityMapPtr updates_by_prio;
+  PendingUpdatesPtr updates;
+
+  // Find pending updates for this cluster.
+  auto updates_by_prio_it = updates_map_.find(cluster.info()->name());
+  if (updates_by_prio_it != updates_map_.end()) {
+    updates_by_prio = updates_by_prio_it->second;
+  } else {
+    updates_by_prio = std::make_shared<PendingUpdatesByPriorityMap>();
+    updates_map_[cluster.info()->name()] = updates_by_prio;
+  }
+
+  // Find pending updates for this priority.
+  auto updates_it = updates_by_prio->find(priority);
+  if (updates_it != updates_by_prio->end()) {
+    updates = updates_it->second;
+  } else {
+    updates = std::make_shared<PendingUpdates>();
+    (*updates_by_prio)[priority] = updates;
+  }
+
+  // Has an update_merge_window gone by since the last update? If so, don't schedule
+  // the update so it can be applied immediately.
+  auto delta = std::chrono::steady_clock::now() - updates->last_updated;
+  uint64_t delta_ms = std::chrono::duration_cast<std::chrono::milliseconds>(delta).count();
+  if (delta_ms > timeout) {
+    updates->last_updated = std::chrono::steady_clock::now();
+    return false;
+  }
+
+  // Record the updates — preserving order — that should be applied when the timer fires.
+  for (auto host : hosts_added) {
+    if (updates->added_seen.count(host) != 1) {
+      updates->added.push_back(host);
+    }
+  }
+
+  for (auto host : hosts_removed) {
+    if (updates->removed_seen.count(host) != 1) {
+      updates->removed.push_back(host);
+    }
+  }
+
+  // If there's no timer, create one.
+  if (updates->timer == nullptr) {
+    updates->timer = dispatcher_.createTimer([this, &cluster, priority, &updates]() -> void {
+      applyUpdates(cluster, priority, updates);
+    });
+    updates->timer->enableTimer(std::chrono::milliseconds(timeout));
+  }
+
+  return true;
+}
+
+void ClusterManagerImpl::applyUpdates(const Cluster& cluster, uint32_t priority,
+                                      PendingUpdatesPtr updates) {
+  // Deliver pending updates.
+  postThreadLocalClusterUpdate(cluster, priority, updates->added, updates->removed);
+  cm_stats_.coalesced_updates_.inc();
+
+  // Reset everything.
+  updates->timer = nullptr;
+  updates->added.clear();
+  updates->removed.clear();
+  updates->added_seen.clear();
+  updates->removed_seen.clear();
+  updates->last_updated = std::chrono::steady_clock::now();
+}
+
 bool ClusterManagerImpl::addOrUpdateCluster(const envoy::api::v2::Cluster& cluster,
                                             const std::string& version_info) {
   // First we need to see if this new config is new or an update to an existing dynamic cluster.

source/common/upstream/cluster_manager_impl.h

@@ -138,6 +138,7 @@ class ClusterManagerInitHelper : Logger::Loggable<Logger::Id::upstream> {
   COUNTER(cluster_added)                                                                           \
   COUNTER(cluster_modified)                                                                        \
   COUNTER(cluster_removed)                                                                         \
+  COUNTER(coalesced_updates)                                                                       \
   GAUGE  (active_clusters)                                                                         \
   GAUGE  (warming_clusters)
 // clang-format on
@@ -361,6 +362,23 @@ class ClusterManagerImpl : public ClusterManager, Logger::Loggable<Logger::Id::u
   // This map is ordered so that config dumping is consistent.
   typedef std::map<std::string, ClusterDataPtr> ClusterMap;
 
+  struct PendingUpdates {
+    PendingUpdates() {}
+    Event::TimerPtr timer;
+    HostVector added;
+    HostVector removed;
+    std::unordered_set<HostSharedPtr> added_seen;
+    std::unordered_set<HostSharedPtr> removed_seen;
+    MonotonicTime last_updated;
+  };
+  typedef std::shared_ptr<PendingUpdates> PendingUpdatesPtr;
+  typedef std::unordered_map<uint32_t, PendingUpdatesPtr> PendingUpdatesByPriorityMap;
+  typedef std::shared_ptr<PendingUpdatesByPriorityMap> PendingUpdatesByPriorityMapPtr;
+  typedef std::unordered_map<std::string, PendingUpdatesByPriorityMapPtr> ClusterUpdatesMap;
+
+  void applyUpdates(const Cluster& cluster, uint32_t priority, PendingUpdatesPtr updates);
+  bool scheduleUpdate(const Cluster& cluster, uint32_t priority, const HostVector& hosts_added,
+                      const HostVector& hosts_removed);
   void createOrUpdateThreadLocalCluster(ClusterData& cluster);
   ProtobufTypes::MessagePtr dumpClusterConfigs();
   static ClusterManagerStats generateStats(Stats::Scope& scope);
@@ -394,6 +412,8 @@ class ClusterManagerImpl : public ClusterManager, Logger::Loggable<Logger::Id::u
   Grpc::AsyncClientManagerPtr async_client_manager_;
   Server::ConfigTracker::EntryOwnerPtr config_tracker_entry_;
   SystemTimeSource& system_time_source_;
+  ClusterUpdatesMap updates_map_;
+  Event::Dispatcher& dispatcher_;
 };
 
 } // namespace Upstream

test/common/upstream/cluster_manager_impl_test.cc

@@ -1620,6 +1620,67 @@ TEST_F(ClusterManagerImplTest, OriginalDstInitialization) {
   factory_.tls_.shutdownThread();
 }
 
+TEST_F(ClusterManagerImplTest, CoalescedUpdates) {
+  const std::string yaml = R"EOF(
+  static_resources:
+    clusters:
+    - name: cluster_1
+      connect_timeout: 0.250s
+      type: STATIC
+      lb_policy: ROUND_ROBIN
+      hosts:
+      - socket_address:
+          address: "127.0.0.1"
+          port_value: 11001
+      - socket_address:
+          address: "127.0.0.1"
+          port_value: 11002
+      common_lb_config:
+        update_merge_window: 3s
+  )EOF";
+
+  create(parseBootstrapFromV2Yaml(yaml));
+  EXPECT_FALSE(cluster_manager_->get("cluster_1")->info()->addedViaApi());
+
+  // Remove each host, sequentially.
+  const Cluster& cluster = cluster_manager_->clusters().begin()->second;
+
+  HostVectorSharedPtr hosts(
+      new HostVector(cluster.prioritySet().hostSetsPerPriority()[0]->hosts()));
+  HostsPerLocalitySharedPtr hosts_per_locality = std::make_shared<HostsPerLocalityImpl>();
+  HostVector hosts_added{};
+  HostVector hosts_removed_0{(*hosts)[0]};
+  HostVector hosts_removed_1{(*hosts)[1]};
+
+  // No timer needs to be mocked at this point, since the first update should be applied
+  // immediately.
+  cluster.prioritySet().hostSetsPerPriority()[0]->updateHosts(
+      hosts, hosts, hosts_per_locality, hosts_per_locality, {}, hosts_added, hosts_removed_0);
+
+  // Now we need the timer to be ready, since createTimer() will be call given that this update
+  // _will_ be delayed.
+  Event::MockTimer* timer = new NiceMock<Event::MockTimer>(&factory_.dispatcher_);
+  cluster.prioritySet().hostSetsPerPriority()[0]->updateHosts(
+      hosts, hosts, hosts_per_locality, hosts_per_locality, {}, hosts_added, hosts_removed_1);
+
+  // Ensure the coalesced updates were applied.
+  timer->callback_();
+  EXPECT_EQ(1, factory_.stats_.counter("cluster_manager.coalesced_updates").value());
+
+  // Prepare a new timer for the next round of updates.
+  timer = new NiceMock<Event::MockTimer>(&factory_.dispatcher_);
+
+  // Add them back.
+  cluster.prioritySet().hostSetsPerPriority()[0]->updateHosts(
+      hosts, hosts, hosts_per_locality, hosts_per_locality, {}, hosts_removed_0, hosts_added);
+  cluster.prioritySet().hostSetsPerPriority()[0]->updateHosts(
+      hosts, hosts, hosts_per_locality, hosts_per_locality, {}, hosts_removed_1, hosts_added);
+
+  // Ensure the coalesced updates were applied again.
+  timer->callback_();
+  EXPECT_EQ(2, factory_.stats_.counter("cluster_manager.coalesced_updates").value());
+}
+
 class ClusterManagerInitHelperTest : public testing::Test {
 public:
   MOCK_METHOD1(onClusterInit, void(Cluster& cluster));