cluster: copy active cluster hosts during warming

include/envoy/upstream/upstream.h

@@ -354,6 +354,11 @@ class PrioritySet {
    */
   virtual const std::vector<HostSetPtr>& hostSetsPerPriority() const PURE;
 
+  /**
+   * @return true if the priority set does have any hosts in any priorities.
+   */
+  virtual bool empty() const PURE;
+
   /**
    * Parameter class for updateHosts.
    */
@@ -771,6 +776,11 @@ class Cluster {
    * @return the const PrioritySet for the cluster.
    */
   virtual const PrioritySet& prioritySet() const PURE;
+
+  /**
+   * @return true, if this cluster is initialized by empty config update.
+   */
+  virtual bool initializedByEmptyConfig() const PURE;
 };
 
 typedef std::shared_ptr<Cluster> ClusterSharedPtr;

source/common/upstream/cluster_manager_impl.cc

@@ -473,18 +473,55 @@ bool ClusterManagerImpl::addOrUpdateCluster(const envoy::api::v2::Cluster& clust
     cluster_warming_cb(cluster_name, ClusterWarmingState::Starting);
     cluster_entry->cluster_->initialize([this, cluster_name, cluster_warming_cb] {
       auto warming_it = warming_clusters_.find(cluster_name);
-      auto& cluster_entry = *warming_it->second;
+      auto& warming_cluster_entry = *warming_it->second;
 
       // If the cluster is being updated, we need to cancel any pending merged updates.
       // Otherwise, applyUpdates() will fire with a dangling cluster reference.
       updates_map_.erase(cluster_name);
 
+      // If management server sends a EDS response, for any other cluster grpc_mux_impl calls
+      // onConfigUpdate on this cluster with empty resources.
+      // See
+      // https://github.com/envoyproxy/envoy/blob/master/source/common/config/grpc_mux_impl.cc#L161
+      // for more details. If the cluster is in fully initialized state, that would just increment
+      // update_empty stat. However, if the cluster is in warming state the initialization call back
+      // would be triggered and warming cluster would not have any hosts. So if onConfigUpdate was
+      // triggered by an EDS update that had no references to this cluster and active cluster has
+      // some hosts, copy the active cluster priority set to the warming cluster to prevent the
+      // hosts from being cleared after warming.
+      // See https://github.com/envoyproxy/envoy/issues/5168 for more context.
+      // This also ensures that we adhere to the clause "When a requested resource is missing in a
+      // RDS or EDS update, Envoy will retain the last known value for this resource." as documented
+      // in https://github.com/envoyproxy/data-plane-api/blob/master/XDS_PROTOCOL.md.
+      const auto active_it = active_clusters_.find(cluster_name);
+      if (active_it != active_clusters_.end()) {
+        const auto& active_cluster_entry = *active_it->second;
+        if (warming_cluster_entry.cluster_->initializedByEmptyConfig() &&
+            !active_cluster_entry.cluster_->prioritySet().empty()) {
+          ENVOY_LOG(debug, "copying host set from active cluster {} to warming cluster",
+                    cluster_name);
+          const auto& active_host_sets =
+              active_cluster_entry.cluster_->prioritySet().hostSetsPerPriority();
+          for (size_t priority = 0; priority < active_host_sets.size(); ++priority) {
+            const auto& active_host_set = active_host_sets[priority];
+            // TODO(ramaraochavali): Can we skip these copies by exporting out const shared_ptr from
+            // HostSet?
+            HostVectorConstSharedPtr hosts_copy(new HostVector(active_host_set->hosts()));
+            HostsPerLocalityConstSharedPtr hosts_per_locality_copy =
+                active_host_set->hostsPerLocality().clone();
+            warming_cluster_entry.cluster_->prioritySet().updateHosts(
+                priority, HostSetImpl::partitionHosts(hosts_copy, hosts_per_locality_copy),
+                active_host_set->localityWeights(), {}, {},
+                active_host_set->overprovisioningFactor());
+          }
+        }
+      }
       active_clusters_[cluster_name] = std::move(warming_it->second);
       warming_clusters_.erase(warming_it);
 
       ENVOY_LOG(info, "warming cluster {} complete", cluster_name);
-      createOrUpdateThreadLocalCluster(cluster_entry);
-      onClusterInit(*cluster_entry.cluster_);
+      createOrUpdateThreadLocalCluster(warming_cluster_entry);
+      onClusterInit(*warming_cluster_entry.cluster_);
       cluster_warming_cb(cluster_name, ClusterWarmingState::Finished);
       updateGauges();
     });

source/common/upstream/eds.cc

@@ -47,7 +47,7 @@ void EdsClusterImpl::onConfigUpdate(const ResourceVector& resources, const std::
   if (resources.empty()) {
     ENVOY_LOG(debug, "Missing ClusterLoadAssignment for {} in onConfigUpdate()", cluster_name_);
     info_->stats().update_empty_.inc();
-    onPreInitComplete();
+    onPreInitComplete(true);
     return;
   }
   if (resources.size() != 1) {
@@ -120,7 +120,7 @@ void EdsClusterImpl::onConfigUpdate(const ResourceVector& resources, const std::
 
   // If we didn't setup to initialize when our first round of health checking is complete, just
   // do it now.
-  onPreInitComplete();
+  onPreInitComplete(false);
 }
 
 bool EdsClusterImpl::updateHostsPerLocality(
@@ -163,7 +163,7 @@ bool EdsClusterImpl::updateHostsPerLocality(
 void EdsClusterImpl::onConfigUpdateFailed(const EnvoyException* e) {
   UNREFERENCED_PARAMETER(e);
   // We need to allow server startup to continue, even if we have a bad config.
-  onPreInitComplete();
+  onPreInitComplete(false);
 }
 
 } // namespace Upstream

source/common/upstream/health_discovery_service.h

@@ -55,6 +55,7 @@ class HdsCluster : public Cluster, Logger::Loggable<Logger::Id::upstream> {
   Outlier::Detector* outlierDetector() override { return outlier_detector_.get(); }
   const Outlier::Detector* outlierDetector() const override { return outlier_detector_.get(); }
   void initialize(std::function<void()> callback) override;
+  bool initializedByEmptyConfig() const override { return false; }
 
   // Creates and starts healthcheckers to its endpoints
   void startHealthchecks(AccessLog::AccessLogManager& access_log_manager, Runtime::Loader& runtime,

source/common/upstream/logical_dns_cluster.cc

@@ -134,7 +134,7 @@ void LogicalDnsCluster::startResolve() {
           }
         }
 
-        onPreInitComplete();
+        onPreInitComplete(false);
         resolve_timer_->enableTimer(dns_refresh_rate_ms_);
       });
 }

source/common/upstream/original_dst_cluster.h

@@ -109,7 +109,7 @@ class OriginalDstCluster : public ClusterImplBase {
   void cleanup();
 
   // ClusterImplBase
-  void startPreInit() override { onPreInitComplete(); }
+  void startPreInit() override { onPreInitComplete(false); }
 
   Event::Dispatcher& dispatcher_;
   const std::chrono::milliseconds cleanup_interval_ms_;

source/common/upstream/subset_lb.h

@@ -68,7 +68,7 @@ class SubsetLoadBalancer : public LoadBalancer, Logger::Loggable<Logger::Id::ups
 
     void update(uint32_t priority, const HostVector& hosts_added, const HostVector& hosts_removed);
 
-    bool empty() { return empty_; }
+    bool empty() const override { return empty_; }
 
     const HostSubsetImpl* getOrCreateHostSubset(uint32_t priority) {
       return reinterpret_cast<const HostSubsetImpl*>(&getOrCreateHostSet(priority));

source/common/upstream/upstream_impl.cc

@@ -741,13 +741,13 @@ void ClusterImplBase::initialize(std::function<void()> callback) {
   startPreInit();
 }
 
-void ClusterImplBase::onPreInitComplete() {
+void ClusterImplBase::onPreInitComplete(const bool empty_update) {
   // Protect against multiple calls.
   if (initialization_started_) {
     return;
   }
   initialization_started_ = true;
-
+  empty_update_ = empty_update;
   ENVOY_LOG(debug, "initializing secondary cluster {} completed", info()->name());
   init_manager_.initialize([this]() { onInitDone(); });
 }
@@ -1059,7 +1059,7 @@ void StaticClusterImpl::startPreInit() {
   }
   priority_state_manager_.reset();
 
-  onPreInitComplete();
+  onPreInitComplete(false);
 }
 
 bool BaseDynamicClusterImpl::updateDynamicHostList(const HostVector& new_hosts,
@@ -1372,7 +1372,7 @@ void StrictDnsClusterImpl::ResolveTarget::startResolve() {
         // multiple DNS names, this will return initialized after a single DNS resolution
         // completes. This is not perfect but is easier to code and unclear if the extra
         // complexity is needed so will start with this.
-        parent_.onPreInitComplete();
+        parent_.onPreInitComplete(false);
         resolve_timer_->enableTimer(parent_.dns_refresh_rate_ms_);
       });
 }

source/common/upstream/upstream_impl.h

@@ -413,6 +413,15 @@ class PrioritySetImpl : public PrioritySet {
                    const HostVector& hosts_removed,
                    absl::optional<uint32_t> overprovisioning_factor = absl::nullopt) override;
 
+  bool empty() const override {
+    for (auto const& host_set : host_sets_) {
+      if (!host_set->hosts().empty()) {
+        return false;
+      }
+    }
+    return true;
+  }
+
 protected:
   // Allows subclasses of PrioritySetImpl to create their own type of HostSetImpl.
   virtual HostSetImplPtr createHostSet(uint32_t priority,
@@ -578,6 +587,7 @@ class ClusterImplBase : public Cluster, protected Logger::Loggable<Logger::Id::u
   // Upstream::Cluster
   PrioritySet& prioritySet() override { return priority_set_; }
   const PrioritySet& prioritySet() const override { return priority_set_; }
+  bool initializedByEmptyConfig() const override { return empty_update_; }
 
   /**
    * Optionally set the health checker for the primary cluster. This is done after cluster
@@ -627,8 +637,10 @@ class ClusterImplBase : public Cluster, protected Logger::Loggable<Logger::Id::u
    * Called by every concrete cluster when pre-init is complete. At this point,
    * shared init starts init_manager_ initialization and determines if there
    * is an initial health check pass needed, etc.
+   *
+   * @param empty_update indicates that onPreInitComplete is triggered via empty update.
    */
-  void onPreInitComplete();
+  void onPreInitComplete(const bool empty_update);
 
   /**
    * Called by every concrete cluster after all targets registered at init manager are
@@ -654,6 +666,7 @@ class ClusterImplBase : public Cluster, protected Logger::Loggable<Logger::Id::u
   bool initialization_started_{};
   std::function<void()> initialization_complete_callback_;
   uint64_t pending_initialize_health_checks_{};
+  bool empty_update_{};
 };
 
 /**