Add support for explicit wildcard resource

docs/root/api-docs/xds_protocol.rst

@@ -427,13 +427,14 @@ 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
-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.
+types, there is also a "wildcard" mode, which is triggered by sending a request containing a special
+resource name "*" in a list of resource names to subscribe to. As a special case, sending an initial
+request with no resource names at all also triggers the wildcard mode. 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.
+The client can opt out from the wildcard mode by unsubscribing from the "*" resource name.
+Additional resource names added to the subscription state while being opted-in into the wildcard
+mode should not be ignored by the server.
 
 Client Behavior
 """""""""""""""
@@ -535,6 +536,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 +556,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
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -779,8 +785,8 @@ resources to avoid resending them over the network by sending them in
 :ref:`initial_resource_versions <envoy_v3_api_field_service.discovery.v3.deltadiscoveryrequest.initial_resource_versions>`.
 Because no state is assumed to be preserved from the previous stream, the reconnecting
 client must provide the server with all resource names it is interested in. Note that for wildcard
-requests (CDS/LDS/SRDS), the request must have no resources in both
-:ref:`resource_names_subscribe <envoy_v3_api_field_service.discovery.v3.deltadiscoveryrequest.resource_names_subscribe>` and
+requests (CDS/LDS/SRDS), the request must have only resources the client was explicitly interested in in
+:ref:`resource_names_subscribe <envoy_v3_api_field_service.discovery.v3.deltadiscoveryrequest.resource_names_subscribe>` and no resources in
 :ref:`resource_names_unsubscribe <envoy_v3_api_field_service.discovery.v3.deltadiscoveryrequest.resource_names_unsubscribe>`.
 
 .. figure:: diagrams/incremental-reconnect.svg

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
 ----------------------

source/common/config/delta_subscription_state.cc

@@ -22,21 +22,26 @@ 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 (auto maybe_resource = getResourceState(resource); maybe_resource.has_value()) {
+                maybe_resource->setAsWaitingForServer();
+                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);
+    // This adds a resource state that is waiting for the server for
+    // more information.
+    resource_state_[a] = ResourceState(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 +58,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 +154,7 @@ void DeltaSubscriptionState::handleGoodResponse(
   {
     const auto scoped_update = ttl_.scopedTtlUpdate();
     for (const auto& resource : message.resources()) {
-      addResourceState(resource);
+      addResourceStateFromServer(resource);
     }
   }
 
@@ -140,8 +170,8 @@ 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);
+    if (auto maybe_resource = getResourceState(resource_name); maybe_resource.has_value()) {
+      maybe_resource->setAsWaitingForServer();
     }
   }
   ENVOY_LOG(debug, "Delta config for {} accepted with {} resources added, {} removed", type_url_,
@@ -177,16 +207,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();
   }
@@ -213,7 +247,7 @@ DeltaSubscriptionState::getNextRequestWithAck(const UpdateAck& ack) {
   return request;
 }
 
-void DeltaSubscriptionState::addResourceState(
+void DeltaSubscriptionState::addResourceStateFromServer(
     const envoy::service::discovery::v3::Resource& resource) {
   if (resource.has_ttl()) {
     ttl_.add(std::chrono::milliseconds(DurationUtil::durationToMilliseconds(resource.ttl())),
@@ -222,18 +256,26 @@ void DeltaSubscriptionState::addResourceState(
     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, old_type);
+  } else {
+    resource_state_.insert(
+        {resource.name(), ResourceState(resource, ResourceType::ReceivedFromServer)});
+  }
 }
 
-void DeltaSubscriptionState::setResourceWaitingForServer(const std::string& resource_name) {
-  resource_state_[resource_name] = ResourceState();
-  resource_names_.insert(resource_name);
+OptRef<DeltaSubscriptionState::ResourceState>
+DeltaSubscriptionState::getResourceState(const std::string& resource_name) {
+  auto itr = resource_state_.find(resource_name);
+  if (itr == resource_state_.end()) {
+    return {};
+  }
+  return {itr->second};
 }
 
 void DeltaSubscriptionState::removeResourceState(const std::string& resource_name) {
   resource_state_.erase(resource_name);
-  resource_names_.erase(resource_name);
 }
 
 } // namespace Config

source/common/config/delta_subscription_state.h

@@ -59,18 +59,43 @@ class DeltaSubscriptionState : public Logger::Loggable<Logger::Id::config> {
   void handleGoodResponse(const envoy::service::discovery::v3::DeltaDiscoveryResponse& message);
   void handleBadResponse(const EnvoyException& e, UpdateAck& ack);
 
+  // This enumeration describes the resource type, which is only relevant for wildcard
+  // subscriptions. Depending on its type, the resource will or will not be resent on the initial
+  // wildcard subscription.
+  enum class ResourceType {
+    // Explicitly requested resource type means that we have asked about the resource by updating
+    // the subscription interest. Such resources are resent on the initial wildcard request.
+    ExplicitlyRequested,
+    // Received from server resources are resources that the state knows about only from the server
+    // response. Such resources are not resent on the initial wildcard request.
+    ReceivedFromServer,
+  };
+
+  // Determines the effective resource type. Explicitly requested type overrides the received from
+  // server type.
+  ResourceType effectiveResourceType(ResourceType old_type, ResourceType new_type) {
+    return (old_type == ResourceType::ReceivedFromServer) ? new_type : old_type;
+  }
+
   class ResourceState {
   public:
-    ResourceState(const envoy::service::discovery::v3::Resource& resource)
-        : version_(resource.version()) {}
+    ResourceState(absl::optional<std::string> version, ResourceType type)
+        : version_(std::move(version)), type_(type) {}
+
+    ResourceState(const envoy::service::discovery::v3::Resource& resource, ResourceType type)
+        : ResourceState(resource.version(), type) {}
 
     // Builds a ResourceState in the waitingForServer state.
-    ResourceState() = default;
+    ResourceState(ResourceType type) : ResourceState(absl::nullopt, type) {}
+
+    ResourceType type() const { return type_; }
 
     // If true, we currently have no version of this resource - we are waiting for the server to
     // provide us with one.
     bool waitingForServer() const { return version_ == absl::nullopt; }
 
+    void setAsWaitingForServer() { version_ = absl::nullopt; }
+
     // Must not be called if waitingForServer() == true.
     std::string version() const {
       ASSERT(version_.has_value());
@@ -79,14 +104,24 @@ class DeltaSubscriptionState : public Logger::Loggable<Logger::Id::config> {
 
   private:
     absl::optional<std::string> version_;
+    ResourceType type_;
   };
 
-  // Use these helpers to ensure resource_state_ and resource_names_ get updated together.
-  void addResourceState(const envoy::service::discovery::v3::Resource& resource);
-  void setResourceWaitingForServer(const std::string& resource_name);
-  void removeResourceState(const std::string& resource_name);
+  // Describes the wildcard mode the subscription is in.
+  enum class WildcardMode {
+    // This mode is being expressed by sending a wildcard subscription request with an empty
+    // resource subscription list.
+    Implicit,
+    // This mode is being expressed by sending a wildcard subscription request that contains "*"
+    // special name in the resource subscription list.
+    Explicit,
+    // This mode is means no wildcard subscription.
+    Disabled,
+  };
 
-  void populateDiscoveryRequest(envoy::service::discovery::v3::DeltaDiscoveryResponse& request);
+  void addResourceStateFromServer(const envoy::service::discovery::v3::Resource& resource);
+  OptRef<ResourceState> getResourceState(const std::string& resource_name);
+  void removeResourceState(const std::string& resource_name);
 
   // A map from resource name to per-resource version. The keys of this map are exactly the resource
   // names we are currently interested in. Those in the waitingForServer state currently don't have
@@ -99,13 +134,9 @@ class DeltaSubscriptionState : public Logger::Loggable<Logger::Id::config> {
   // disable heartbeats for these resources (currently only VHDS).
   const bool supports_heartbeats_;
   TtlManager ttl_;
-  // The keys of resource_versions_. Only tracked separately because std::map does not provide an
-  // iterator into just its keys.
-  absl::flat_hash_set<std::string> resource_names_;
 
   const std::string type_url_;
-  // Is the subscription is for a wildcard request.
-  const bool wildcard_;
+  WildcardMode mode_;
   UntypedConfigUpdateCallbacks& watch_map_;
   const LocalInfo::LocalInfo& local_info_;
   Event::Dispatcher& dispatcher_;

source/common/config/new_grpc_mux_impl.cc

@@ -128,7 +128,10 @@ GrpcMuxWatchPtr NewGrpcMuxImpl::addWatch(const std::string& type_url,
   auto entry = subscriptions_.find(type_url);
   if (entry == subscriptions_.end()) {
     // We don't yet have a subscription for type_url! Make one!
-    addSubscription(type_url, options.use_namespace_matching_, resources.empty());
+    // No resources or an existence of the special name implies that
+    // this is a wildcard request subscription.
+    const bool wildcard = resources.empty() || (resources.find("*") != resources.end());
+    addSubscription(type_url, options.use_namespace_matching_, wildcard);
     return addWatch(type_url, resources, callbacks, resource_decoder, options);
   }
 

source/common/config/watch_map.cc

@@ -49,7 +49,8 @@ void WatchMap::removeDeferredWatches() {
 AddedRemoved
 WatchMap::updateWatchInterest(Watch* watch,
                               const absl::flat_hash_set<std::string>& update_to_these_names) {
-  if (update_to_these_names.empty()) {
+  if (update_to_these_names.empty() ||
+      update_to_these_names.find("*") != update_to_these_names.end()) {
     wildcard_watches_.insert(watch);
   } else {
     wildcard_watches_.erase(watch);