CM: Add on-demand cluster discovery functionality

include/envoy/upstream/cluster_manager.h

@@ -39,7 +39,7 @@ namespace Envoy {
 namespace Upstream {
 
 /**
- * ClusterUpdateCallbacks provide a way to exposes Cluster lifecycle events in the
+ * ClusterUpdateCallbacks provide a way to expose Cluster lifecycle events in the
  * ClusterManager.
  */
 class ClusterUpdateCallbacks {
@@ -72,6 +72,63 @@ class ClusterUpdateCallbacksHandle {
 
 using ClusterUpdateCallbacksHandlePtr = std::unique_ptr<ClusterUpdateCallbacksHandle>;
 
+/**
+ * Status enum for the result of an attempted cluster discovery.
+ */
+enum class ClusterDiscoveryStatus {
+  /**
+   * Cluster was not found during the discovery process.
+   */
+  Missing,
+  /**
+   * Cluster found and currently available through ClusterManager.
+   */
+  Available,
+};
+
+/**
+ * ClusterDiscoveryCallback is a callback called at the end of the on-demand cluster discovery
+ * process. The status of the discovery is sent as a parameter.
+ */
+using ClusterDiscoveryCallback = std::function<void(ClusterDiscoveryStatus)>;
+using ClusterDiscoveryCallbackWeakPtr = std::weak_ptr<ClusterDiscoveryCallback>;
+using ClusterDiscoveryCallbackSharedPtr = std::shared_ptr<ClusterDiscoveryCallback>;
+
+/**
+ * ClusterDiscoveryCallbackHandle is a RAII wrapper for a ClusterDiscoveryCallback. Deleting the
+ * ClusterDiscoveryCallbackHandle will remove the callbacks from ClusterManager.
+ */
+class ClusterDiscoveryCallbackHandle {
+public:
+  virtual ~ClusterDiscoveryCallbackHandle() = default;
+};
+
+using ClusterDiscoveryCallbackHandlePtr = std::unique_ptr<ClusterDiscoveryCallbackHandle>;
+
+class OdCdsApiHandle {
+public:
+  virtual ~OdCdsApiHandle() = default;
+
+  /**
+   * Request an on-demand discovery of a cluster with a passed name. This ODCDS may be used to
+   * perform the discovery process in the main thread if there is no discovery going on for this
+   * cluster. The passed callback will be invoked when the cluster is added and warmed up. It is
+   * expected that the callback will be destroyed when it is invoked. To cancel the discovery,
+   * destroy the returned handle and the callback.
+   *
+   * This function is thread-safe.
+   *
+   * @param name is the name of the cluster to be discovered.
+   * @param callback will be called when the discovery is finished.
+   * @return ClusterDiscoveryCallbackHandlePtr the discovery process handle.
+   */
+  virtual ClusterDiscoveryCallbackHandlePtr
+  requestOnDemandClusterDiscovery(const std::string& name,
+                                  ClusterDiscoveryCallbackWeakPtr callback) PURE;
+};
+
+using OdCdsApiHandleSharedPtr = std::shared_ptr<OdCdsApiHandle>;
+
 class ClusterManagerFactory;
 
 // These are per-cluster per-thread, so not "global" stats.
@@ -309,6 +366,19 @@ class ClusterManager {
   virtual const ClusterRequestResponseSizeStatNames&
   clusterRequestResponseSizeStatNames() const PURE;
   virtual const ClusterTimeoutBudgetStatNames& clusterTimeoutBudgetStatNames() const PURE;
+
+  /**
+   * Allocates an on-demand CDS API provider from configuration proto or locator.
+   *
+   * @param odcds_config is a configuration proto. Used when odcds_resources_locator is a nullptr.
+   * @param odcds_resources_locator is a locator for ODCDS.
+   * @param validation_visitor
+   * @return OdCdsApiHandleSharedPtr the ODCDS handle.
+   */
+  virtual OdCdsApiHandleSharedPtr
+  allocateOdCdsApi(const envoy::config::core::v3::ConfigSource& odcds_config,
+                   const xds::core::v3::ResourceLocator* odcds_resources_locator,
+                   ProtobufMessage::ValidationVisitor& validation_visitor) PURE;
 };
 
 using ClusterManagerPtr = std::unique_ptr<ClusterManager>;

source/common/upstream/BUILD

@@ -44,6 +44,25 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "od_cds_api_lib",
+    srcs = ["od_cds_api_impl.cc"],
+    hdrs = ["od_cds_api_impl.h"],
+    deps = [
+        ":cds_api_helper_lib",
+        "//include/envoy/config:subscription_interface",
+        "//include/envoy/protobuf:message_validator_interface",
+        "//include/envoy/stats:stats_interface",
+        "//include/envoy/upstream:cluster_manager_interface",
+        "//source/common/common:minimal_logger_lib",
+        "//source/common/config:subscription_base_interface",
+        "//source/common/grpc:common_lib",
+        "//source/common/protobuf",
+        "@envoy_api//envoy/config/cluster/v3:pkg_cc_proto",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
+    ],
+)
+
 envoy_cc_library(
     name = "cluster_manager_lib",
     srcs = ["cluster_manager_impl.cc"],
@@ -52,6 +71,7 @@ envoy_cc_library(
         ":cds_api_lib",
         ":load_balancer_lib",
         ":load_stats_reporter_lib",
+        ":od_cds_api_lib",
         ":ring_hash_lb_lib",
         ":subset_lb_lib",
         "//include/envoy/api:api_interface",

source/common/upstream/cds_api_helper.h

@@ -37,7 +37,7 @@ class CdsApiHelper : Logger::Loggable<Logger::Id::upstream> {
 
 private:
   ClusterManager& cm_;
-  std::string name_;
+  const std::string name_;
   std::string system_version_info_;
 };
 

source/common/upstream/cluster_manager_impl.cc

@@ -957,6 +957,7 @@ void ClusterManagerImpl::postThreadLocalClusterUpdate(ClusterManagerCluster& cm_
     per_priority.overprovisioning_factor_ = host_set->overprovisioningFactor();
   }
 
+  pending_cluster_creations_.erase(cm_cluster.cluster().info()->name());
   tls_.runOnAllThreads(
       [info = cm_cluster.cluster().info(), params = std::move(params), add_or_update_cluster,
        load_balancer_factory](OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {
@@ -1027,6 +1028,198 @@ ClusterManagerImpl::addThreadLocalClusterUpdateCallbacks(ClusterUpdateCallbacks&
   return std::make_unique<ClusterUpdateCallbacksHandleImpl>(cb, cluster_manager.update_callbacks_);
 }
 
+namespace {
+
+using ClusterAddedCb = std::function<void(ThreadLocalCluster&)>;
+
+class ClusterCallbacks : public ClusterUpdateCallbacks {
+public:
+  ClusterCallbacks(ClusterAddedCb cb) : cb_(std::move(cb)) {}
+
+  void onClusterAddOrUpdate(ThreadLocalCluster& cluster) override { cb_(cluster); };
+
+  void onClusterRemoval(const std::string&) override {}
+
+private:
+  ClusterAddedCb cb_;
+};
+
+} // namespace
+
+ClusterManagerImpl::ClusterDiscoveryManager::ClusterDiscoveryManager(
+    ThreadLocalClusterManagerImpl& parent)
+    : parent_(parent) {}
+
+void ClusterManagerImpl::ClusterDiscoveryManager::ensureCallbacksAreInstalled() {
+  if (callbacks_handle_) {
+    return;
+  }
+  auto cb = ClusterAddedCb([this](ThreadLocalCluster& cluster) {
+    processClusterName(cluster.info()->name(), ClusterDiscoveryStatus::Available);
+  });
+  callbacks_ = std::make_unique<ClusterCallbacks>(cb);
+  callbacks_handle_ = parent_.parent_.addThreadLocalClusterUpdateCallbacks(*callbacks_);
+}
+
+void ClusterManagerImpl::ClusterDiscoveryManager::processClusterName(
+    const std::string& name, ClusterDiscoveryStatus cluster_status) {
+  auto map_node_handle = pending_clusters_.extract(name);
+  if (map_node_handle.empty()) {
+    return;
+  }
+  for (const auto& weak_callback : map_node_handle.mapped()) {
+    auto callback = weak_callback.lock();
+    if (callback != nullptr) {
+      (*callback)(cluster_status);
+    }
+  }
+  maybePostResetCallbacks();
+}
+
+ClusterManagerImpl::ClusterDiscoveryManager::Pair
+ClusterManagerImpl::ClusterDiscoveryManager::addCallback(
+    const std::string& name, ClusterDiscoveryCallbackWeakPtr weak_callback) {
+  auto& callbacks_deque = pending_clusters_[name];
+  callbacks_deque.emplace_back(weak_callback);
+  auto handle =
+      std::make_unique<ClusterDiscoveryCallbackHandleImpl>(*this, std::move(weak_callback), name);
+  auto discovery_in_progress = (callbacks_deque.size() > 1);
+  return {std::move(handle), discovery_in_progress};
+}
+
+void ClusterManagerImpl::ClusterDiscoveryManager::erase(const std::string& name,
+                                                        ClusterDiscoveryCallbackWeakPtr cb) {
+  const bool drop_deque = eraseFromDeque(name, cb);
+  if (drop_deque) {
+    pending_clusters_.erase(name);
+  }
+  maybePostResetCallbacks();
+}
+
+bool ClusterManagerImpl::ClusterDiscoveryManager::eraseFromDeque(
+    const std::string& name, ClusterDiscoveryCallbackWeakPtr cb) {
+  auto it = pending_clusters_.find(name);
+  if (it == pending_clusters_.end()) {
+    return false;
+  }
+  auto& deque = it->second;
+  // Could use std::erase_if, but it's only in C++20.
+  auto it2 = std::remove_if(deque.begin(), deque.end(),
+                            [cb](ClusterDiscoveryCallbackWeakPtr const& weak_ptr) {
+                              if (cb.owner_before(weak_ptr)) {
+                                return false;
+                              }
+                              if (weak_ptr.owner_before(cb)) {
+                                return false;
+                              }
+                              return true;
+                            });
+  deque.erase(it2, deque.end());
+  return deque.empty();
+}
+
+void ClusterManagerImpl::ClusterDiscoveryManager::maybePostResetCallbacks() {
+  if (!callbacks_handle_cleanup_posted_ && pending_clusters_.empty()) {
+    parent_.thread_local_dispatcher_.post([this] {
+      // Something might got added in the meantime, so check the
+      // map again.
+      if (pending_clusters_.empty()) {
+        callbacks_handle_.reset();
+        callbacks_.reset();
+      }
+      callbacks_handle_cleanup_posted_ = false;
+    });
+    callbacks_handle_cleanup_posted_ = true;
+  }
+}
+
+OdCdsApiHandleSharedPtr
+ClusterManagerImpl::allocateOdCdsApi(const envoy::config::core::v3::ConfigSource& odcds_config,
+                                     const xds::core::v3::ResourceLocator* odcds_resources_locator,
+                                     ProtobufMessage::ValidationVisitor& validation_visitor) {
+  // TODO(krnowak): Instead of creating a new handle every time, store the handles internally and
+  // return an already existing one if the config or locator matches. Note that this may need a way
+  // to clean up the unused handles, so we can close the unnecessary connections.
+  auto odcds = OdCdsApiImpl::create(odcds_config, odcds_resources_locator, *this, stats_,
+                                    validation_visitor);
+  return OdCdsApiHandleImpl::create(*this, std::move(odcds));
+}
+
+ClusterDiscoveryCallbackHandlePtr
+ClusterManagerImpl::requestOnDemandClusterDiscovery(OdCdsApiHandleImplSharedPtr odcds_handle,
+                                                    const std::string& name,
+                                                    ClusterDiscoveryCallbackWeakPtr weak_callback) {
+  ThreadLocalClusterManagerImpl& cluster_manager = *tls_;
+
+  cluster_manager.cdm_.ensureCallbacksAreInstalled();
+
+  auto [handle, discovery_in_progress] = cluster_manager.cdm_.addCallback(name, weak_callback);
+  if (discovery_in_progress) {
+    // We have already started a discovery process for a cluster with
+    // this name, so nothing more left to do here.
+    return std::move(handle);
+  }
+  dispatcher_.post([this, odcds_handle = std::move(odcds_handle), weak_callback, name,
+                    &thread_local_dispatcher = cluster_manager.thread_local_dispatcher_] {
+    // Check for the cluster here too. It might have been added
+    // between the time when this callback was posted and when it is
+    // being executed.
+    if (getThreadLocalCluster(name) != nullptr) {
+      if (weak_callback.expired()) {
+        // Not only the cluster was added, but it was also already
+        // handled, so don't bother with posting the callback back to
+        // the worker thread.
+        return;
+      }
+      thread_local_dispatcher.post([weak_callback] {
+        if (auto callback = weak_callback.lock(); callback != nullptr) {
+          // If this gets called here, it means that we requested a
+          // discovery of a cluster without checking if that cluster
+          // is already known by cluster manager.
+          (*callback)(ClusterDiscoveryStatus::Available);
+        }
+      });
+      return;
+    }
+
+    auto it = pending_cluster_creations_.find(name);
+    if (it != pending_cluster_creations_.end()) {
+      // We already began the discovery process for this cluster,
+      // nothing to do.
+      return;
+    }
+    auto& odcds = odcds_handle->getOdCds();
+    odcds.updateOnDemand(name);
+    auto timer_cb = Event::TimerCb([this, name] { notifyExpiredDiscovery(name); });
+    auto timer = dispatcher_.createTimer(timer_cb);
+    timer->enableTimer(std::chrono::milliseconds(5000));
+    // Keep odcds alive for the duration of the discovery process.
+    pending_cluster_creations_.insert(
+        {std::move(name), ClusterCreation{std::move(odcds_handle), std::move(timer)}});
+  });
+
+  return std::move(handle);
+}
+
+void ClusterManagerImpl::notifyExpiredDiscovery(const std::string& name) {
+  auto map_node_handle = pending_cluster_creations_.extract(name);
+  if (map_node_handle.empty()) {
+    return;
+  }
+  // Defer destroying the timer, so it's not destroyed during its
+  // callback. TimerPtr is a unique_ptr, which is not copyable, but
+  // std::function is copyable, we turn a move-only unique_ptr into a
+  // copyable shared_ptr and pass that to the std::function.
+  dispatcher_.post([timer = std::shared_ptr<Event::Timer>(
+                        std::move(map_node_handle.mapped().expiration_timer_))] {});
+  tls_.runOnAllThreads([name](OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {
+    if (!cluster_manager.has_value()) {
+      return;
+    }
+    cluster_manager->cdm_.processClusterName(name, ClusterDiscoveryStatus::Missing);
+  });
+}
+
 ProtobufTypes::MessagePtr ClusterManagerImpl::dumpClusterConfigs() {
   auto config_dump = std::make_unique<envoy::admin::v3::ClustersConfigDump>();
   config_dump->set_version_info(cds_api_ != nullptr ? cds_api_->versionInfo() : "");
@@ -1061,7 +1254,7 @@ ProtobufTypes::MessagePtr ClusterManagerImpl::dumpClusterConfigs() {
 ClusterManagerImpl::ThreadLocalClusterManagerImpl::ThreadLocalClusterManagerImpl(
     ClusterManagerImpl& parent, Event::Dispatcher& dispatcher,
     const absl::optional<LocalClusterParams>& local_cluster_params)
-    : parent_(parent), thread_local_dispatcher_(dispatcher) {
+    : parent_(parent), thread_local_dispatcher_(dispatcher), cdm_(*this) {
   // If local cluster is defined then we need to initialize it first.
   if (local_cluster_params.has_value()) {
     const auto& local_cluster_name = local_cluster_params->info_->name();