CM: Add on-demand cluster discovery functionality

docs/root/api-docs/xds_protocol.rst

@@ -427,13 +427,18 @@ names becomes empty, that means that the client is no longer interested in any r
 specified type.
 
 For :ref:`Listener <envoy_v3_api_msg_config.listener.v3.Listener>` and :ref:`Cluster <envoy_v3_api_msg_config.cluster.v3.Cluster>` resource
-types, there is also a "wildcard" mode, which is triggered when the initial request on the stream
-for that resource type contains no resource names. In this case, the server should use
+types, there is also a "wildcard" mode, which comes in two flavors. First flavor, implicit, is triggered when the initial request on the stream
+for that resource type contains no resource names. Second flavor, explicit, is triggered when a request
+(not necessarily an initial one on the stream) for that resource type contains (among other names) a special name "*".
+For wildcard requests, the server should use
 site-specific business logic to determine the full set of resources that the client is interested
-in, typically based on the client's :ref:`node <envoy_v3_api_msg_config.core.v3.Node>` identification. Note
-that once a stream has entered wildcard mode for a given resource type, there is no way to change
-the stream out of wildcard mode; resource names specified in any subsequent request on the stream
-will be ignored.
+in, typically based on the client's :ref:`node <envoy_v3_api_msg_config.core.v3.Node>` identification.
+The client can opt out from the wildcard mode by unsubscribing from the "*" resource name. Note that
+opting back into the wildcard mode can only be done with a request containing the "*" resource name,
+thus switching into the explicit wildcard mode. Also note that if client wants to express
+an interest in a resource name after sending an empty initial request on the stream, the client
+needs to switch to the explicit wildcard mode, otherwise resource names specified in subsequent
+request on the stream will be ignored.
 
 Client Behavior
 """""""""""""""
@@ -535,6 +540,8 @@ being requested by the client, and if one of those resources springs into existe
 server must send an update to the client informing it of the new resource. Clients that initially
 see a resource that does not exist must be prepared for the resource to be created at any time.
 
+.. _xds_protocol_unsubscribing_from_resources:
+
 Unsubscribing From Resources
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
@@ -553,7 +560,10 @@ Note that for :ref:`Listener <envoy_v3_api_msg_config.listener.v3.Listener>` and
 resource types where the stream is in "wildcard" mode (see :ref:`How the client specifies what
 resources to return <xds_protocol_resource_hints>` for details), the set of resources being
 subscribed to is determined by the server instead of the client, so there is no mechanism
-for the client to unsubscribe from resources.
+for the client to unsubscribe from resources. The only resources that the client could unsubscribe
+from are the resources that the client explicitly expressed the interest in before. Note that
+the server may still send the resource to the client if the resource was also a part of the set
+of resources determined by the server from the wildcard subscription.
 
 Requesting Multiple Resources on a Single Stream
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

docs/root/version_history/current.rst

@@ -8,6 +8,7 @@ Incompatible Behavior Changes
 * grpc_bridge_filter: the filter no longer collects grpc stats in favor of the existing grpc stats filter.
   The behavior can be reverted by changing runtime key ``envoy.reloadable_features.grpc_bridge_stats_disabled``.
 * tracing: update Apache SkyWalking tracer version to be compatible with 8.4.0 data collect protocol. This change will introduce incompatibility with SkyWalking 8.3.0.
+* xds: added the explicit wildcard mode, which allows subscribing to specific resource names on top of a wildcard subscription. This means that the resource name ``*`` is reserved. This may mean that in some cases the initial wildcard subscription request on the stream will not be empty, but will have a non-empty list of resources with a special name among them. See :ref:`wildcard mode description <xds_protocol_resource_hints>` and :ref:`unsubscribing from resources <xds_protocol_unsubscribing_from_resources>` for details.
 
 Minor Behavior Changes
 ----------------------

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,76 @@ class ClusterUpdateCallbacksHandle {
 
 using ClusterUpdateCallbacksHandlePtr = std::unique_ptr<ClusterUpdateCallbacksHandle>;
 
+/**
+ * Status enum for the result of an attempted cluster discovery.
+ */
+enum class ClusterDiscoveryStatus {
+  /**
+   * The discovery process timed out. This means that we haven't yet received any reply from
+   * on-demand CDS about it.
+   */
+  Timeout,
+  /**
+   * The discovery process has concluded and on-demand CDS has no such cluster.
+   */
+  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 ClusterDiscoveryCallbackPtr = std::unique_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>;
+
+/**
+ * A handle to an on-demand CDS.
+ */
+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. When the requested cluster is added and warmed up, the passed callback will be invoked
+   * in the same thread that invoked this function.
+   *
+   * The returned handle can be destroyed to prevent the callback to be invoked. Note that the
+   * handle can only be destroyed in the same thread that invoked the function. Destroying the
+   * handle might not stop the discovery process, though. As soon as the callback is invoked,
+   * destroying the handle does nothing. It is a responsibility of the caller to make sure that the
+   * objects captured in the callback outlive 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.
+   * @param timeout describes how long the operation may take before failing.
+   * @return ClusterDiscoveryCallbackHandlePtr the discovery process handle.
+   */
+  virtual ClusterDiscoveryCallbackHandlePtr
+  requestOnDemandClusterDiscovery(absl::string_view name, ClusterDiscoveryCallbackPtr callback,
+                                  std::chrono::milliseconds timeout) PURE;
+};
+
+using OdCdsApiHandlePtr = std::unique_ptr<OdCdsApiHandle>;
+
 class ClusterManagerFactory;
 
 // These are per-cluster per-thread, so not "global" stats.
@@ -309,6 +379,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 nullopt.
+   * @param odcds_resources_locator is a locator for ODCDS. Used over odcds_config if not a nullopt.
+   * @param validation_visitor
+   * @return OdCdsApiHandlePtr the ODCDS handle.
+   */
+  virtual OdCdsApiHandlePtr
+  allocateOdCdsApi(const envoy::config::core::v3::ConfigSource& odcds_config,
+                   OptRef<xds::core::v3::ResourceLocator> odcds_resources_locator,
+                   ProtobufMessage::ValidationVisitor& validation_visitor) PURE;
 };
 
 using ClusterManagerPtr = std::unique_ptr<ClusterManager>;

source/common/config/delta_subscription_state.cc

@@ -22,21 +22,23 @@ DeltaSubscriptionState::DeltaSubscriptionState(std::string type_url,
           [this](const auto& expired) {
             Protobuf::RepeatedPtrField<std::string> removed_resources;
             for (const auto& resource : expired) {
-              setResourceWaitingForServer(resource);
-              removed_resources.Add(std::string(resource));
+              if (setResourceWaitingForServer(resource)) {
+                removed_resources.Add(std::string(resource));
+              }
             }
 
             watch_map_.onConfigUpdate({}, removed_resources, "");
           },
           dispatcher, dispatcher.timeSource()),
-      type_url_(std::move(type_url)), wildcard_(wildcard), watch_map_(watch_map),
+      type_url_(std::move(type_url)),
+      mode_(wildcard ? WildcardMode::Implicit : WildcardMode::Disabled), watch_map_(watch_map),
       local_info_(local_info), dispatcher_(dispatcher) {}
 
 void DeltaSubscriptionState::updateSubscriptionInterest(
     const absl::flat_hash_set<std::string>& cur_added,
     const absl::flat_hash_set<std::string>& cur_removed) {
   for (const auto& a : cur_added) {
-    setResourceWaitingForServer(a);
+    addResourceWaitingForServer(a, ResourceType::ExplicitlyRequested);
     // If interest in a resource is removed-then-added (all before a discovery request
     // can be sent), we must treat it as a "new" addition: our user may have forgotten its
     // copy of the resource after instructing us to remove it, and need to be reminded of it.
@@ -53,6 +55,31 @@ void DeltaSubscriptionState::updateSubscriptionInterest(
     names_added_.erase(r);
     names_removed_.insert(r);
   }
+  switch (mode_) {
+  case WildcardMode::Implicit:
+    if (names_removed_.find("*") != names_removed_.end()) {
+      // we explicitly cancel the wildcard subscription
+      mode_ = WildcardMode::Disabled;
+    } else if (!names_added_.empty()) {
+      // switch to explicit mode if we requested some extra names
+      mode_ = WildcardMode::Explicit;
+    }
+    break;
+
+  case WildcardMode::Explicit:
+    if (names_removed_.find("*") != names_removed_.end()) {
+      // we explicitly cancel the wildcard subscription
+      mode_ = WildcardMode::Disabled;
+    }
+    break;
+
+  case WildcardMode::Disabled:
+    if (names_added_.find("*") != names_added_.end()) {
+      // we switch into an explicit wildcard subscription
+      mode_ = WildcardMode::Explicit;
+    }
+    break;
+  }
 }
 
 // Not having sent any requests yet counts as an "update pending" since you're supposed to resend
@@ -124,7 +151,7 @@ void DeltaSubscriptionState::handleGoodResponse(
   {
     const auto scoped_update = ttl_.scopedTtlUpdate();
     for (const auto& resource : message.resources()) {
-      addResourceState(resource);
+      addResourceState(resource, ResourceType::ReceivedFromServer);
     }
   }
 
@@ -140,9 +167,7 @@ void DeltaSubscriptionState::handleGoodResponse(
   // initial_resource_versions messages, but will remind us to explicitly tell the server "I'm
   // cancelling my subscription" when we lose interest.
   for (const auto& resource_name : message.removed_resources()) {
-    if (resource_names_.find(resource_name) != resource_names_.end()) {
-      setResourceWaitingForServer(resource_name);
-    }
+    setResourceWaitingForServer(resource_name);
   }
   ENVOY_LOG(debug, "Delta config for {} accepted with {} resources added, {} removed", type_url_,
             message.resources().size(), message.removed_resources().size());
@@ -177,16 +202,20 @@ DeltaSubscriptionState::getNextRequestAckless() {
       if (!resource_state.waitingForServer()) {
         (*request.mutable_initial_resource_versions())[resource_name] = resource_state.version();
       }
-      // As mentioned above, fill resource_names_subscribe with everything, including names we
-      // have yet to receive any resource for unless this is a wildcard subscription, for which
-      // the first request on a stream must be without any resource names.
-      if (!wildcard_) {
+      // Add resource names to resource_names_subscribe only if this is not a wildcard subscription
+      // request or if we requested this resource explicitly (so we are actually in explicit
+      // wildcard mode).
+      if (mode_ == WildcardMode::Disabled ||
+          resource_state.type() == ResourceType::ExplicitlyRequested) {
         names_added_.insert(resource_name);
       }
     }
-    // Wildcard subscription initial requests must have no resource_names_subscribe.
-    if (wildcard_) {
-      names_added_.clear();
+    // We are not clearing the names_added_ set. If we are in implicit wildcard subscription mode,
+    // then the set should already be empty. If we are in explicit wildcard mode then the set will
+    // contain the names we explicitly requested, but we need to add * to the list to make sure it's
+    // sent too.
+    if (mode_ == WildcardMode::Explicit) {
+      names_added_.insert("*");
     }
     names_removed_.clear();
   }
@@ -214,26 +243,44 @@ DeltaSubscriptionState::getNextRequestWithAck(const UpdateAck& ack) {
 }
 
 void DeltaSubscriptionState::addResourceState(
-    const envoy::service::discovery::v3::Resource& resource) {
+    const envoy::service::discovery::v3::Resource& resource, ResourceType type) {
   if (resource.has_ttl()) {
     ttl_.add(std::chrono::milliseconds(DurationUtil::durationToMilliseconds(resource.ttl())),
              resource.name());
   } else {
     ttl_.clear(resource.name());
   }
 
-  resource_state_[resource.name()] = ResourceState(resource);
-  resource_names_.insert(resource.name());
+  if (auto it = resource_state_.find(resource.name()); it != resource_state_.end()) {
+    auto old_type = it->second.type();
+    it->second = ResourceState(resource, effectiveResourceType(old_type, type));
+  } else {
+    resource_state_.insert({resource.name(), ResourceState(resource, type)});
+  }
+}
+
+bool DeltaSubscriptionState::setResourceWaitingForServer(const std::string& resource_name) {
+  auto itr = resource_state_.find(resource_name);
+  if (itr == resource_state_.end()) {
+    return false;
+  }
+  auto old_type = itr->second.type();
+  itr->second = ResourceState(old_type);
+  return true;
 }
 
-void DeltaSubscriptionState::setResourceWaitingForServer(const std::string& resource_name) {
-  resource_state_[resource_name] = ResourceState();
-  resource_names_.insert(resource_name);
+void DeltaSubscriptionState::addResourceWaitingForServer(const std::string& resource_name,
+                                                         ResourceType type) {
+  if (auto it = resource_state_.find(resource_name); it != resource_state_.end()) {
+    auto old_type = it->second.type();
+    it->second = ResourceState(effectiveResourceType(old_type, type));
+  } else {
+    resource_state_.insert({resource_name, ResourceState(type)});
+  }
 }
 
 void DeltaSubscriptionState::removeResourceState(const std::string& resource_name) {
   resource_state_.erase(resource_name);
-  resource_names_.erase(resource_name);
 }
 
 } // namespace Config