config: fix a bug where the on demand subscription not works

changelogs/changelogs.yaml

@@ -52,6 +52,8 @@ areas:
     title: circuit_breaker
   composite:
     title: composite
+  config:
+    title: config
   dns_resolver:
     title: dns_resolver
   dynamic_modules:

changelogs/current/bug_fixes/config__fixed-on-demand-xds-resource-routing.rst

@@ -0,0 +1,6 @@
+Fixed a bug in on-demand xDS where a resource requested after the initial subscription could be
+silently dropped instead of delivered. An on-demand update only updated the subscription sent to
+the management server, not the internal watch routing, so the server's response for the requested
+resource matched no watch and was discarded. This affected on-demand cluster discovery (ODCDS) when
+requesting a second, different cluster, and VHDS virtual hosts under a route configuration.
+On-demand requests now also register watch routing, so these responses are delivered.

envoy/config/grpc_mux.h

@@ -48,6 +48,39 @@ class GrpcMuxWatch {
    * @param resources set of resource names to watch for
    */
   virtual void update(const absl::flat_hash_set<std::string>& resources) PURE;
+
+  /**
+   * Adds resources to the set the watch is interested in, without disturbing the resources it
+   * already watches (the additive counterpart to update()). This updates BOTH the watch-level
+   * routing and the subscription sent to the management server, so it is how on-demand requests add
+   * a new resource: the watch will route the resource when it arrives, and the server is told to
+   * send it.
+   * @param resources set of resource names to additionally watch for and subscribe to.
+   */
+  virtual void append(const absl::flat_hash_set<std::string>& resources) PURE;
+
+  /**
+   * Declares the resource-name patterns this watch ACCEPTS for routing. This affects routing ONLY
+   * and never changes the subscription sent to the management server (that is driven solely by
+   * update()/append()).
+   *
+   * By default (accept() never called), the watch is routed exactly the resources it watches via
+   * update()/append(); if that set is empty or contains the wildcard "*", the watch is a catch-all
+   * and is routed every resource.
+   *
+   * When accept(patterns) is called, the patterns are evaluated FIRST and SUPPRESS the default
+   * wildcard (catch-all) semantics: the watch is then routed only resources matching 'patterns',
+   * plus any concrete names it explicitly watches. Each pattern must be the wildcard "*" (accept
+   * every resource) or a suffix glob -- a prefix followed by a slash and an asterisk -- which
+   * accepts every resource named with that prefix. This is how a watch can subscribe to the
+   * wildcard on the wire (update({"*"})) while routing only a subset -- e.g. VHDS, which accepts
+   * only virtual hosts under its route configuration. To keep wildcard routing while using
+   * accept(), include "*" in the patterns.
+   *
+   * Calling accept({}) clears any previously registered patterns and restores the default behavior.
+   * @param patterns set of wildcard / suffix-glob patterns to accept.
+   */
+  virtual void accept(const absl::flat_hash_set<std::string>& patterns) PURE;
 };
 
 using GrpcMuxWatchPtr = std::unique_ptr<GrpcMuxWatch>;
@@ -108,9 +141,6 @@ class GrpcMux {
                                    OpaqueResourceDecoderSharedPtr resource_decoder,
                                    const SubscriptionOptions& options) PURE;
 
-  virtual void requestOnDemandUpdate(const std::string& type_url,
-                                     const absl::flat_hash_set<std::string>& for_update) PURE;
-
   /**
    * Returns an EdsResourcesCache for this GrpcMux if there is one.
    * @return EdsResourcesCacheOptRef optional eds resources cache for the gRPC-mux.

envoy/config/subscription.h

@@ -137,12 +137,6 @@ class SubscriptionCallbacks {
  * Options associated with a Subscription.
  */
 struct SubscriptionOptions {
-  /**
-   * For legacy VHDS, should an xDS resource name be treated as <namespace>/<resource name>? This is
-   * incompatible with the use of xdstp:// naming.
-   */
-  bool use_namespace_matching_{};
-
   /**
    * For xdstp:// resource names, should node context parameters be added at the transport layer?
    */
@@ -186,10 +180,7 @@ class UntypedConfigUpdateCallbacks {
    * @param removed_resources names of resources that this fetch instructed to be removed.
    * @param system_version_info aggregate response data "version", for debugging.
    * @throw EnvoyException with reason if the config changes are rejected. Otherwise the changes
-   * @param use_namespace_matching if the resources should me matched on their namespaces, rather
-   * than unique names. This is used when a collection of resources (e.g. virtual hosts in VHDS) is
-   * being updated. Accepted changes have their version_info reflected in subsequent
-   * requests.
+   * are accepted and have their version_info reflected in subsequent requests.
    */
   virtual void
   onConfigUpdate(absl::Span<const envoy::service::discovery::v3::Resource* const> added_resources,
@@ -232,6 +223,25 @@ class Subscription {
    * @param add_these_names resource ids for inclusion in the discovery request.
    */
   virtual void requestOnDemandUpdate(const absl::flat_hash_set<std::string>& add_these_names) PURE;
+
+  /**
+   * Declares the resource-name patterns this subscription ACCEPTS for routing. This affects routing
+   * ONLY and never changes the discovery request sent to the management server.
+   *
+   * By default (accept() never called), the subscription routes exactly the resources it requested;
+   * if that set is empty or contains the wildcard "*", it is a catch-all and routes every resource.
+   * When accept(patterns) is called, the patterns are evaluated first and suppress the default
+   * wildcard (catch-all) semantics: only resources matching 'patterns' (plus explicitly requested
+   * names) are routed. Each pattern must be the wildcard (the "*" string) or a suffix glob (a
+   * prefix followed by a slash and an asterisk), matching every resource under that prefix. To keep
+   * wildcard routing while using accept(), include "*" in the patterns.
+   *
+   * This lets a subscription accept resources it cannot enumerate up front while still subscribing
+   * to the wildcard on the wire (e.g. VHDS virtual hosts under a route configuration). Only
+   * meaningful for gRPC subscriptions; other subscription types treat this as a no-op.
+   * @param patterns wildcard / suffix-glob patterns to accept.
+   */
+  virtual void accept(const absl::flat_hash_set<std::string>& patterns) PURE;
 };
 
 using SubscriptionPtr = std::unique_ptr<Subscription>;

source/common/config/null_grpc_mux_impl.h

@@ -23,10 +23,6 @@ class NullGrpcMuxImpl : public GrpcMux,
     ExceptionUtil::throwEnvoyException("ADS must be configured to support an ADS config source");
   }
 
-  void requestOnDemandUpdate(const std::string&, const absl::flat_hash_set<std::string>&) override {
-    ENVOY_BUG(false, "unexpected request for on demand update");
-  }
-
   absl::Status updateMuxSource(
       Grpc::RawAsyncClientSharedPtr&&, Grpc::RawAsyncClientSharedPtr&&, Stats::Scope&,
       BackOffStrategyPtr&&, const envoy::config::core::v3::ApiConfigSource&,

source/common/router/vhds.cc

@@ -73,15 +73,20 @@ VhdsSubscription::VhdsSubscription(RouteConfigUpdatePtr& config_update_info,
       init_target_(fmt::format("VhdsConfigSubscription {}",
                                config_update_info_->protobufConfigurationCast().name()),
                    [this]() {
-                     subscription_->start(
-                         {config_update_info_->protobufConfigurationCast().name()});
+                     // Start with no concrete subscription; virtual hosts are fetched on demand.
+                     // Accept every resource under this route configuration's namespace, so that
+                     // on-demand virtual hosts -- whose resource names are
+                     // "<route_config_name>/<vhost>" -- are routed back to this subscription's
+                     // watch without being subscribed on the wire.
+                     subscription_->accept(
+                         {config_update_info_->protobufConfigurationCast().name() + "/*"});
+                     subscription_->start({});
                    }),
       resource_type_helper_(factory_context.messageValidationContext().dynamicValidationVisitor(),
                             "name"),
       route_config_provider_(route_config_provider) {
   const auto resource_name = resource_type_helper_.getResourceName();
   Envoy::Config::SubscriptionOptions options;
-  options.use_namespace_matching_ = true;
   absl::StatusOr<Envoy::Config::SubscriptionPtr> status_or =
       factory_context.clusterManager().subscriptionFactory().subscriptionFromConfigSource(
           config_update_info_->protobufConfigurationCast().vhds().config_source(),

source/extensions/config_subscription/filesystem/filesystem_subscription_impl.h

@@ -39,6 +39,7 @@ class FilesystemSubscriptionImpl : public Config::Subscription,
   void requestOnDemandUpdate(const absl::flat_hash_set<std::string>&) override {
     ENVOY_BUG(false, "unexpected request for on demand update");
   }
+  void accept(const absl::flat_hash_set<std::string>&) override {}
 
 protected:
   virtual std::string refreshInternal(ProtobufTypes::MessagePtr* config_update);

source/extensions/config_subscription/grpc/grpc_mux_impl.h

@@ -66,9 +66,6 @@ class GrpcMuxImpl : public GrpcMux,
                            OpaqueResourceDecoderSharedPtr resource_decoder,
                            const SubscriptionOptions& options) override;
 
-  void requestOnDemandUpdate(const std::string&, const absl::flat_hash_set<std::string>&) override {
-  }
-
   EdsResourcesCacheOptRef edsResourcesCache() override {
     return makeOptRefFromPtr(eds_resources_cache_.get());
   }
@@ -164,6 +161,11 @@ class GrpcMuxImpl : public GrpcMux,
       parent_.queueDiscoveryRequest(type_url_);
     }
 
+    // On-demand and glob watch interest are only supported by the delta muxes; this legacy
+    // state-of-the-world mux does not implement them.
+    void append(const absl::flat_hash_set<std::string>&) override {}
+    void accept(const absl::flat_hash_set<std::string>&) override {}
+
     // Maintain deterministic wire ordering via ordered std::set.
     std::set<std::string> resources_;
     SubscriptionCallbacks& callbacks_;

source/extensions/config_subscription/grpc/grpc_subscription_impl.cc

@@ -39,6 +39,14 @@ void GrpcSubscriptionImpl::start(const absl::flat_hash_set<std::string>& resourc
 
   watch_ = grpc_mux_->addWatch(type_url_, resources, *this, resource_decoder_, options_);
 
+  // Apply any accept() patterns declared before start(). This happens before grpc_mux_->start()
+  // below (and delivery to a watch only ever happens via an async discovery response), so the
+  // routing filter is guaranteed to be in effect before any resource can be routed to this watch.
+  if (!deferred_accept_patterns_.empty()) {
+    watch_->accept(deferred_accept_patterns_);
+    deferred_accept_patterns_.clear();
+  }
+
   // The attempt stat here is maintained for the purposes of having consistency between ADS and
   // gRPC/filesystem/REST Subscriptions. Since ADS is push based and muxed, the notion of an
   // "attempt" for a given xDS API combined by ADS is not really that meaningful.
@@ -59,10 +67,24 @@ void GrpcSubscriptionImpl::updateResourceInterest(
 
 void GrpcSubscriptionImpl::requestOnDemandUpdate(
     const absl::flat_hash_set<std::string>& for_update) {
-  grpc_mux_->requestOnDemandUpdate(type_url_, for_update);
+  // Additionally add the on-demand resources to this subscription's watch. This updates both the
+  // watch-map routing (so the response is delivered to our callbacks) and the subscription sent to
+  // the management server (so it is fetched). Without the watch-map update, a resource that was not
+  // named at start() time would match no watch and its update would be silently dropped.
+  watch_->append(for_update);
   stats_.update_attempt_.inc();
 }
 
+void GrpcSubscriptionImpl::accept(const absl::flat_hash_set<std::string>& patterns) {
+  if (watch_ == nullptr) {
+    // Called before start(): the watch does not exist yet. Remember the patterns; start() applies
+    // them before starting the stream, so the routing filter is in place before any delivery.
+    deferred_accept_patterns_ = patterns;
+    return;
+  }
+  watch_->accept(patterns);
+}
+
 // Config::SubscriptionCallbacks
 absl::Status
 GrpcSubscriptionImpl::onConfigUpdate(const std::vector<Config::DecodedResourceRef>& resources,

source/extensions/config_subscription/grpc/grpc_subscription_impl.h

@@ -32,6 +32,7 @@ class GrpcSubscriptionImpl : public Subscription,
   void
   updateResourceInterest(const absl::flat_hash_set<std::string>& update_to_these_names) override;
   void requestOnDemandUpdate(const absl::flat_hash_set<std::string>& add_these_names) override;
+  void accept(const absl::flat_hash_set<std::string>& patterns) override;
   // Config::SubscriptionCallbacks (all pass through to callbacks_!)
   absl::Status onConfigUpdate(const std::vector<Config::DecodedResourceRef>& resources,
                               const std::string& version_info) override;
@@ -53,6 +54,10 @@ class GrpcSubscriptionImpl : public Subscription,
   SubscriptionStats stats_;
   const std::string type_url_;
   GrpcMuxWatchPtr watch_;
+  // accept() patterns declared before start() (i.e. before watch_ exists). They are applied to the
+  // watch in start(), before the gRPC stream is started, so the routing filter is in effect before
+  // any resource can be delivered. Empty once applied (or if accept() was never called early).
+  absl::flat_hash_set<std::string> deferred_accept_patterns_;
   Event::Dispatcher& dispatcher_;
   // NOTE: if another subscription of the same type_url has already been started, this value will be
   // ignored in favor of the other subscription's.

source/extensions/config_subscription/grpc/new_grpc_mux_impl.cc

@@ -246,7 +246,7 @@ 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!
-    entry = addSubscription(type_url, options.use_namespace_matching_);
+    entry = addSubscription(type_url);
   }
 
   Watch* watch = entry->second->watch_map_.addWatch(callbacks, *resource_decoder);
@@ -293,13 +293,9 @@ NewGrpcMuxImpl::updateMuxSource(Grpc::RawAsyncClientSharedPtr&& primary_async_cl
 // Updates the list of resource names watched by the given watch. If an added name is new across
 // the whole subscription, or if a removed name has no other watch interested in it, then the
 // subscription will enqueue and attempt to send an appropriate discovery request.
-void NewGrpcMuxImpl::updateWatch(const std::string& type_url, Watch* watch,
-                                 const absl::flat_hash_set<std::string>& resources,
-                                 const SubscriptionOptions& options) {
-  ASSERT(watch != nullptr);
-  auto sub = subscriptions_.find(type_url);
-  RELEASE_ASSERT(sub != subscriptions_.end(),
-                 fmt::format("Watch of {} has no subscription to update.", type_url));
+absl::flat_hash_set<std::string>
+NewGrpcMuxImpl::effectiveResources(const absl::flat_hash_set<std::string>& resources,
+                                   const SubscriptionOptions& options) {
   // We need to prepare xdstp:// resources for the transport, by normalizing and adding any extra
   // context parameters.
   absl::flat_hash_set<std::string> effective_resources;
@@ -320,32 +316,43 @@ void NewGrpcMuxImpl::updateWatch(const std::string& type_url, Watch* watch,
       effective_resources.insert(resource);
     }
   }
+  return effective_resources;
+}
+
+void NewGrpcMuxImpl::updateWatch(const std::string& type_url, Watch* watch,
+                                 const absl::flat_hash_set<std::string>& resources,
+                                 const SubscriptionOptions& options) {
+  ASSERT(watch != nullptr);
+  auto sub = subscriptions_.find(type_url);
+  RELEASE_ASSERT(sub != subscriptions_.end(),
+                 fmt::format("Watch of {} has no subscription to update.", type_url));
+  const absl::flat_hash_set<std::string> effective_resources =
+      effectiveResources(resources, options);
   auto added_removed = sub->second->watch_map_.updateWatchInterest(watch, effective_resources);
   if (xds_config_tracker_.has_value() && !added_removed.removed_.empty()) {
     for (absl::string_view resource : added_removed.removed_) {
       xds_config_tracker_->onResourceUnsubscribed(type_url, resource);
     }
   }
-  if (options.use_namespace_matching_) {
-    // This is to prevent sending out of requests that contain prefixes instead of resource names
-    sub->second->sub_state_.updateSubscriptionInterest({}, {});
-  } else {
-    sub->second->sub_state_.updateSubscriptionInterest(added_removed.added_,
-                                                       added_removed.removed_);
-  }
+  sub->second->sub_state_.updateSubscriptionInterest(added_removed.added_, added_removed.removed_);
   // Tell the server about our change in interest, if any.
   if (sub->second->sub_state_.subscriptionUpdatePending()) {
     trySendDiscoveryRequests();
   }
 }
 
-void NewGrpcMuxImpl::requestOnDemandUpdate(const std::string& type_url,
-                                           const absl::flat_hash_set<std::string>& for_update) {
+void NewGrpcMuxImpl::appendWatch(const std::string& type_url, Watch* watch,
+                                 const absl::flat_hash_set<std::string>& resources,
+                                 const SubscriptionOptions& options) {
+  ASSERT(watch != nullptr);
   auto sub = subscriptions_.find(type_url);
   RELEASE_ASSERT(sub != subscriptions_.end(),
                  fmt::format("Watch of {} has no subscription to update.", type_url));
-  sub->second->sub_state_.updateSubscriptionInterest(for_update, {});
-  // Tell the server about our change in interest, if any.
+  // Additionally update the watch-map routing, then subscribe to whatever became newly interesting
+  // across the whole subscription. This keeps watch_interest_ and the subscription consistent.
+  auto added_removed =
+      sub->second->watch_map_.appendWatchInterest(watch, effectiveResources(resources, options));
+  sub->second->sub_state_.updateSubscriptionInterest(added_removed.added_, {});
   if (sub->second->sub_state_.subscriptionUpdatePending()) {
     trySendDiscoveryRequests();
   }
@@ -359,18 +366,27 @@ void NewGrpcMuxImpl::removeWatch(const std::string& type_url, Watch* watch) {
   entry->second->watch_map_.removeWatch(watch);
 }
 
+void NewGrpcMuxImpl::accept(const std::string& type_url, Watch* watch,
+                            const absl::flat_hash_set<std::string>& patterns) {
+  ASSERT(watch != nullptr);
+  auto sub = subscriptions_.find(type_url);
+  RELEASE_ASSERT(sub != subscriptions_.end(),
+                 fmt::format("Watch of {} has no subscription to update.", type_url));
+  // Glob interest affects routing only; the subscription sent to the server is left untouched.
+  sub->second->watch_map_.accept(watch, patterns);
+}
+
 NewGrpcMuxImpl::SubscriptionsMap::iterator
-NewGrpcMuxImpl::addSubscription(const std::string& type_url, const bool use_namespace_matching) {
+NewGrpcMuxImpl::addSubscription(const std::string& type_url) {
   // Resource cache is only used for EDS resources.
   EdsResourcesCacheOptRef resources_cache{std::nullopt};
   if (eds_resources_cache_ &&
       (type_url == Config::getTypeUrl<envoy::config::endpoint::v3::ClusterLoadAssignment>())) {
     resources_cache = makeOptRefFromPtr(eds_resources_cache_.get());
   }
   auto [it, success] = subscriptions_.emplace(
-      type_url, std::make_unique<SubscriptionStuff>(type_url, use_namespace_matching, dispatcher_,
-                                                    config_validators_.get(), xds_config_tracker_,
-                                                    resources_cache));
+      type_url, std::make_unique<SubscriptionStuff>(type_url, dispatcher_, config_validators_.get(),
+                                                    xds_config_tracker_, resources_cache));
   // Insertion must succeed, as the addSubscription method is only called if
   // the map doesn't have the type_url.
   ASSERT(success);