From 900b62adefcd7c4b5ba995f3b6352d91b1e62564 Mon Sep 17 00:00:00 2001 From: wbpcode Date: Fri, 17 Jul 2026 08:26:49 +0000 Subject: [PATCH 1/3] xds: add integration tests for multiple clusters Signed-off-by: wbpcode --- .../http/on_demand/odcds_integration_test.cc | 80 +++++++++++++++++++ 1 file changed, 80 insertions(+) diff --git a/test/extensions/filters/http/on_demand/odcds_integration_test.cc b/test/extensions/filters/http/on_demand/odcds_integration_test.cc index 64d45be0df764..ae7b6adfc3e97 100644 --- a/test/extensions/filters/http/on_demand/odcds_integration_test.cc +++ b/test/extensions/filters/http/on_demand/odcds_integration_test.cc @@ -600,6 +600,86 @@ TEST_P(OdCdsIntegrationTest, OnDemandClusterDiscoveryRemembersDiscoveredCluster) cleanupUpstreamAndDownstream(); } +// Regression test for on-demand CDS with two DIFFERENT clusters on the same subscription: +// - a request to unknown cluster "new_cluster" triggers on-demand discovery (the initial +// subscription, via start()), which is resolved and the request succeeds; +// - a request to a different unknown cluster "new_cluster_2" triggers a SECOND on-demand +// discovery, which goes through requestOnDemandUpdate()/append() on the already-started +// subscription. +TEST_P(OdCdsIntegrationTest, OnDemandClusterDiscoverySecondClusterRoutesResponse) { + addPerRouteConfig(OdCdsIntegrationHelper::createPerRouteConfig( + OdCdsIntegrationHelper::createOdCdsConfigSource("odcds_cluster"), 2500), + "integration", {}); + initialize(); + + // A second cluster for the second on-demand request. It points at the same upstream as + // "new_cluster" -- the endpoint is irrelevant here; what matters is that the second on-demand + // response is routed so the cluster is added. + auto second_cluster = ConfigHelper::buildStaticCluster( + "new_cluster_2", fake_upstreams_[new_cluster_upstream_idx_]->localAddress()->ip()->port(), + Network::Test::getLoopbackAddressString(ipVersion())); + + codec_client_ = makeHttpConnection(makeClientConnection(lookupPort("http"))); + + // First request -> first on-demand cluster ("new_cluster"). This is the initial (start()-time) + // subscription, which was always routed correctly. + Http::TestRequestHeaderMapImpl request_headers_1{{":method", "GET"}, + {":path", "/"}, + {":scheme", "http"}, + {":authority", "vhost.first"}, + {"Pick-This-Cluster", "new_cluster"}}; + IntegrationStreamDecoderPtr response_1 = codec_client_->makeHeaderOnlyRequest(request_headers_1); + + createXdsConnection(); + auto result = xds_connection_->waitForNewStream(*dispatcher_, odcds_stream_); + RELEASE_ASSERT(result, result.message()); + odcds_stream_->startGrpcStream(); + + EXPECT_TRUE(compareDeltaDiscoveryRequest(Config::TestTypeUrl::get().Cluster, {"new_cluster"}, {}, + odcds_stream_.get())); + sendDeltaDiscoveryResponse( + Config::TestTypeUrl::get().Cluster, {new_cluster_}, {}, "1", odcds_stream_.get()); + EXPECT_TRUE(compareDeltaDiscoveryRequest(Config::TestTypeUrl::get().Cluster, {}, {}, + odcds_stream_.get())); + + waitForNextUpstreamRequest(new_cluster_upstream_idx_); + upstream_request_->encodeHeaders(default_response_headers_, true); + ASSERT_TRUE(response_1->waitForEndStream()); + verifyResponse(std::move(response_1), "200", {}, {}); + + // Reset the downstream/upstream between the two independent requests (the second request routes + // to a different cluster, i.e. a new upstream connection). The ODCDS (xds) stream persists. + cleanupUpstreamAndDownstream(); + ASSERT_TRUE(codec_client_->waitForDisconnect()); + + // Second request -> a DIFFERENT on-demand cluster ("new_cluster_2"). This is the second on-demand + // request on the same subscription, so it flows through requestOnDemandUpdate()/append(). The + // response must be routed to the ODCDS callbacks so the cluster is added and the request succeeds + // (before the fix this timed out with a 503). + codec_client_ = makeHttpConnection(makeClientConnection(lookupPort("http"))); + Http::TestRequestHeaderMapImpl request_headers_2{{":method", "GET"}, + {":path", "/"}, + {":scheme", "http"}, + {":authority", "vhost.first"}, + {"Pick-This-Cluster", "new_cluster_2"}}; + IntegrationStreamDecoderPtr response_2 = codec_client_->makeHeaderOnlyRequest(request_headers_2); + + EXPECT_TRUE(compareDeltaDiscoveryRequest(Config::TestTypeUrl::get().Cluster, {"new_cluster_2"}, + {}, odcds_stream_.get())); + sendDeltaDiscoveryResponse( + Config::TestTypeUrl::get().Cluster, {second_cluster}, {}, "2", odcds_stream_.get()); + EXPECT_TRUE(compareDeltaDiscoveryRequest(Config::TestTypeUrl::get().Cluster, {}, {}, + odcds_stream_.get())); + + waitForNextUpstreamRequest(new_cluster_upstream_idx_); + upstream_request_->encodeHeaders(default_response_headers_, true); + ASSERT_TRUE(response_2->waitForEndStream()); + verifyResponse(std::move(response_2), "200", {}, {}); + + cleanUpXdsConnection(); + cleanupUpstreamAndDownstream(); +} + // tests a scenario when: // - making a request to an unknown cluster // - odcds initiates a connection with a request for the cluster From 36242f5763cc4ebd7d201c4313952fc436d65b41 Mon Sep 17 00:00:00 2001 From: wbpcode Date: Fri, 17 Jul 2026 08:56:51 +0000 Subject: [PATCH 2/3] config: fix a bug where the on demand subscription not works Signed-off-by: wbpcode --- changelogs/changelogs.yaml | 2 + ...__fixed-on-demand-xds-resource-routing.rst | 6 + envoy/config/grpc_mux.h | 36 ++++- envoy/config/subscription.h | 30 ++-- source/common/config/null_grpc_mux_impl.h | 4 - source/common/router/vhds.cc | 11 +- .../filesystem/filesystem_subscription_impl.h | 1 + .../config_subscription/grpc/grpc_mux_impl.h | 8 +- .../grpc/grpc_subscription_impl.cc | 24 ++- .../grpc/grpc_subscription_impl.h | 5 + .../grpc/new_grpc_mux_impl.cc | 62 +++++--- .../grpc/new_grpc_mux_impl.h | 35 ++++- .../config_subscription/grpc/watch_map.cc | 134 ++++++++++++++--- .../config_subscription/grpc/watch_map.h | 58 +++++++- .../grpc/xds_mux/grpc_mux_impl.cc | 72 +++++---- .../grpc/xds_mux/grpc_mux_impl.h | 35 +++-- .../rest/http_subscription_impl.h | 1 + .../grpc/delta_subscription_impl_test.cc | 137 +++++++++++++++++ .../grpc/grpc_mux_impl_test.cc | 11 -- .../grpc/new_grpc_mux_impl_test.cc | 16 +- .../grpc/watch_map_test.cc | 139 ++++++++++++++++-- .../grpc/xds_grpc_mux_impl_test.cc | 5 - .../on_demand/on_demand_integration_test.cc | 64 ++++++++ test/mocks/config/mocks.h | 5 +- 24 files changed, 741 insertions(+), 160 deletions(-) create mode 100644 changelogs/current/bug_fixes/config__fixed-on-demand-xds-resource-routing.rst diff --git a/changelogs/changelogs.yaml b/changelogs/changelogs.yaml index 64baef06f11d5..915dd38b0ba16 100644 --- a/changelogs/changelogs.yaml +++ b/changelogs/changelogs.yaml @@ -52,6 +52,8 @@ areas: title: circuit_breaker composite: title: composite + config: + title: config dns_resolver: title: dns_resolver dynamic_modules: diff --git a/changelogs/current/bug_fixes/config__fixed-on-demand-xds-resource-routing.rst b/changelogs/current/bug_fixes/config__fixed-on-demand-xds-resource-routing.rst new file mode 100644 index 0000000000000..6146e8672e8bf --- /dev/null +++ b/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. diff --git a/envoy/config/grpc_mux.h b/envoy/config/grpc_mux.h index f8aa1b3e4bf29..839e70ec4fce3 100644 --- a/envoy/config/grpc_mux.h +++ b/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& 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& 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& patterns) PURE; }; using GrpcMuxWatchPtr = std::unique_ptr; @@ -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& for_update) PURE; - /** * Returns an EdsResourcesCache for this GrpcMux if there is one. * @return EdsResourcesCacheOptRef optional eds resources cache for the gRPC-mux. diff --git a/envoy/config/subscription.h b/envoy/config/subscription.h index 75e99acaacb82..3a51a26eba7d4 100644 --- a/envoy/config/subscription.h +++ b/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 /? 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 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& 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& patterns) PURE; }; using SubscriptionPtr = std::unique_ptr; diff --git a/source/common/config/null_grpc_mux_impl.h b/source/common/config/null_grpc_mux_impl.h index 590d32f30bf8c..03f99df00ab7a 100644 --- a/source/common/config/null_grpc_mux_impl.h +++ b/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&) 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&, diff --git a/source/common/router/vhds.cc b/source/common/router/vhds.cc index 2f905193c61ef..b7d6db225d644 100644 --- a/source/common/router/vhds.cc +++ b/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 + // "/" -- 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 status_or = factory_context.clusterManager().subscriptionFactory().subscriptionFromConfigSource( config_update_info_->protobufConfigurationCast().vhds().config_source(), diff --git a/source/extensions/config_subscription/filesystem/filesystem_subscription_impl.h b/source/extensions/config_subscription/filesystem/filesystem_subscription_impl.h index 258d8e7b028bc..10ab219430ed0 100644 --- a/source/extensions/config_subscription/filesystem/filesystem_subscription_impl.h +++ b/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&) override { ENVOY_BUG(false, "unexpected request for on demand update"); } + void accept(const absl::flat_hash_set&) override {} protected: virtual std::string refreshInternal(ProtobufTypes::MessagePtr* config_update); diff --git a/source/extensions/config_subscription/grpc/grpc_mux_impl.h b/source/extensions/config_subscription/grpc/grpc_mux_impl.h index eb47c049ffd03..c38943297ec16 100644 --- a/source/extensions/config_subscription/grpc/grpc_mux_impl.h +++ b/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&) 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&) override {} + void accept(const absl::flat_hash_set&) override {} + // Maintain deterministic wire ordering via ordered std::set. std::set resources_; SubscriptionCallbacks& callbacks_; diff --git a/source/extensions/config_subscription/grpc/grpc_subscription_impl.cc b/source/extensions/config_subscription/grpc/grpc_subscription_impl.cc index 54b4da8b31d39..70c79b3281954 100644 --- a/source/extensions/config_subscription/grpc/grpc_subscription_impl.cc +++ b/source/extensions/config_subscription/grpc/grpc_subscription_impl.cc @@ -39,6 +39,14 @@ void GrpcSubscriptionImpl::start(const absl::flat_hash_set& 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& 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& 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& resources, diff --git a/source/extensions/config_subscription/grpc/grpc_subscription_impl.h b/source/extensions/config_subscription/grpc/grpc_subscription_impl.h index 5039449ba9b7b..3144d2e889d0b 100644 --- a/source/extensions/config_subscription/grpc/grpc_subscription_impl.h +++ b/source/extensions/config_subscription/grpc/grpc_subscription_impl.h @@ -32,6 +32,7 @@ class GrpcSubscriptionImpl : public Subscription, void updateResourceInterest(const absl::flat_hash_set& update_to_these_names) override; void requestOnDemandUpdate(const absl::flat_hash_set& add_these_names) override; + void accept(const absl::flat_hash_set& patterns) override; // Config::SubscriptionCallbacks (all pass through to callbacks_!) absl::Status onConfigUpdate(const std::vector& 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 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. diff --git a/source/extensions/config_subscription/grpc/new_grpc_mux_impl.cc b/source/extensions/config_subscription/grpc/new_grpc_mux_impl.cc index 0a59a4abcf656..4514addb67305 100644 --- a/source/extensions/config_subscription/grpc/new_grpc_mux_impl.cc +++ b/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& 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 +NewGrpcMuxImpl::effectiveResources(const absl::flat_hash_set& 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 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& 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 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& for_update) { +void NewGrpcMuxImpl::appendWatch(const std::string& type_url, Watch* watch, + const absl::flat_hash_set& 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,8 +366,18 @@ 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& 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_ && @@ -368,9 +385,8 @@ NewGrpcMuxImpl::addSubscription(const std::string& type_url, const bool use_name resources_cache = makeOptRefFromPtr(eds_resources_cache_.get()); } auto [it, success] = subscriptions_.emplace( - type_url, std::make_unique(type_url, use_namespace_matching, dispatcher_, - config_validators_.get(), xds_config_tracker_, - resources_cache)); + type_url, std::make_unique(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); diff --git a/source/extensions/config_subscription/grpc/new_grpc_mux_impl.h b/source/extensions/config_subscription/grpc/new_grpc_mux_impl.h index 45b8a3fc59115..10de2e2dc5665 100644 --- a/source/extensions/config_subscription/grpc/new_grpc_mux_impl.h +++ b/source/extensions/config_subscription/grpc/new_grpc_mux_impl.h @@ -53,9 +53,6 @@ class NewGrpcMuxImpl OpaqueResourceDecoderSharedPtr resource_decoder, const SubscriptionOptions& options) override; - void requestOnDemandUpdate(const std::string& type_url, - const absl::flat_hash_set& for_update) override; - EdsResourcesCacheOptRef edsResourcesCache() override { return makeOptRefFromPtr(eds_resources_cache_.get()); } @@ -104,11 +101,11 @@ class NewGrpcMuxImpl } struct SubscriptionStuff { - SubscriptionStuff(const std::string& type_url, const bool use_namespace_matching, - Event::Dispatcher& dispatcher, CustomConfigValidators* config_validators, + SubscriptionStuff(const std::string& type_url, Event::Dispatcher& dispatcher, + CustomConfigValidators* config_validators, XdsConfigTrackerOptRef xds_config_tracker, EdsResourcesCacheOptRef eds_resources_cache) - : watch_map_(use_namespace_matching, type_url, config_validators, eds_resources_cache), + : watch_map_(type_url, config_validators, eds_resources_cache), sub_state_(type_url, watch_map_, dispatcher, xds_config_tracker) { // If eds resources cache is provided, then the type must be ClusterLoadAssignment. ASSERT( @@ -151,6 +148,14 @@ class NewGrpcMuxImpl parent_.updateWatch(type_url_, watch_, resources, options_); } + void append(const absl::flat_hash_set& resources) override { + parent_.appendWatch(type_url_, watch_, resources, options_); + } + + void accept(const absl::flat_hash_set& patterns) override { + parent_.accept(type_url_, watch_, patterns); + } + private: const std::string type_url_; Watch* watch_; @@ -178,9 +183,23 @@ class NewGrpcMuxImpl const absl::flat_hash_set& resources, const SubscriptionOptions& options); + // Additionally adds resources to the given watch's interest, updating both the watch-map routing + // and the subscription (see GrpcMuxWatch::append). + void appendWatch(const std::string& type_url, Watch* watch, + const absl::flat_hash_set& resources, + const SubscriptionOptions& options); + // Registers glob interest for the given watch; routing only, never affects the subscription + // (see GrpcMuxWatch::accept). + void accept(const std::string& type_url, Watch* watch, + const absl::flat_hash_set& patterns); + + // Normalizes xdstp:// resource names for the transport; non-xdstp names are returned unchanged. + absl::flat_hash_set + effectiveResources(const absl::flat_hash_set& resources, + const SubscriptionOptions& options); + // Adds a subscription for the type_url to the subscriptions map and order list. - SubscriptionsMap::iterator addSubscription(const std::string& type_url, - bool use_namespace_matching); + SubscriptionsMap::iterator addSubscription(const std::string& type_url); void trySendDiscoveryRequests(); diff --git a/source/extensions/config_subscription/grpc/watch_map.cc b/source/extensions/config_subscription/grpc/watch_map.cc index 2cb93629305ba..b19889dbd3f6b 100644 --- a/source/extensions/config_subscription/grpc/watch_map.cc +++ b/source/extensions/config_subscription/grpc/watch_map.cc @@ -30,19 +30,80 @@ Watch* WatchMap::addWatch(SubscriptionCallbacks& callbacks, return watch_ptr; } +void WatchMap::forgetWatch(Watch* watch) { + wildcard_watches_.erase(watch); // may or may not be in there, but we want it gone. + forgetAccept(watch); + watches_.erase(watch); +} + +bool WatchMap::watchTrackedInInterest(Watch* watch) { + for (const auto& [name, watches] : watch_interest_) { + if (watches.contains(watch)) { + return true; + } + } + return false; +} + +void WatchMap::forgetAccept(Watch* watch) { + // Remove this watch from the glob reverse-index, using its own record of accepted patterns. + for (absl::string_view pattern : watch->accept_patterns_) { + if (pattern == Wildcard) { + accept_wildcard_watches_.erase(watch); + continue; + } + // "/*" -- keyed by "". + auto it = accept_prefix_watches_.find(pattern.substr(0, pattern.size() - 2)); + if (it != accept_prefix_watches_.end()) { + it->second.erase(watch); + if (it->second.empty()) { + accept_prefix_watches_.erase(it); + } + } + } +} + +void WatchMap::updateWildcardWatches(Watch* watch) { + // accept() patterns are evaluated FIRST and take precedence: a watch that has declared any + // accept() pattern is routed by those patterns (plus its concrete names) only, so it is never a + // catch-all wildcard watch here -- even if its resource_names_ is empty or contains "*". This is + // what lets a watch subscribe to the wildcard on the wire while routing only a subset (e.g. + // VHDS). To keep wildcard routing while using accept(), the caller includes "*" in the accept() + // patterns, which is tracked separately in accept_wildcard_watches_. + if (!watch->accept_patterns_.empty()) { + wildcard_watches_.erase(watch); + return; + } + // No accept() filter: the watched set defines routing. An explicit "*" or an empty watched set + // makes the watch a catch-all wildcard; otherwise it is routed only its specific resource names. + if (watch->resource_names_.contains(Wildcard) || watch->resource_names_.empty()) { + wildcard_watches_.insert(watch); + } else { + wildcard_watches_.erase(watch); + } +} + void WatchMap::removeWatch(Watch* watch) { + // Contract: the caller must first empty this watch's resource-name interest via + // updateWatchInterest(watch, {}) so that the per-name reference counts in watch_interest_ are + // decremented and any resulting server-side unsubscribe is emitted (see findRemovals()). Removal + // here only cleans the membership structures (wildcard_watches_, the accept_* + // indices, watches_); it deliberately does not touch watch_interest_. If the watch is still + // referenced by any watch_interest_ entry, a resource name was leaked (and the server never told + // to unsubscribe it), so fail loudly in debug builds. + ASSERT(!watchTrackedInInterest(watch), + "removeWatch() called before the watch's resource names were drained via " + "updateWatchInterest(watch, {})"); if (deferred_removed_during_update_ != nullptr) { deferred_removed_during_update_->insert(watch); } else { - wildcard_watches_.erase(watch); // may or may not be in there, but we want it gone. - watches_.erase(watch); + forgetWatch(watch); } } void WatchMap::removeDeferredWatches() { - for (auto& watch : *deferred_removed_during_update_) { - wildcard_watches_.erase(watch); // may or may not be in there, but we want it gone. - watches_.erase(watch); + for (auto* watch : *deferred_removed_during_update_) { + forgetWatch(watch); } deferred_removed_during_update_ = nullptr; } @@ -50,12 +111,6 @@ void WatchMap::removeDeferredWatches() { AddedRemoved WatchMap::updateWatchInterest(Watch* watch, const absl::flat_hash_set& update_to_these_names) { - if (update_to_these_names.empty() || update_to_these_names.contains(Wildcard)) { - wildcard_watches_.insert(watch); - } else { - wildcard_watches_.erase(watch); - } - absl::flat_hash_set newly_added_to_watch; SetUtil::setDifference(update_to_these_names, watch->resource_names_, newly_added_to_watch); @@ -63,6 +118,7 @@ WatchMap::updateWatchInterest(Watch* watch, SetUtil::setDifference(watch->resource_names_, update_to_these_names, newly_removed_from_watch); watch->resource_names_ = update_to_these_names; + updateWildcardWatches(watch); // First resources are added and only then removed, so a watch won't be removed // if its interest has been replaced (rather than completely removed). @@ -80,10 +136,56 @@ WatchMap::updateWatchInterest(Watch* watch, return {std::move(added_resources), std::move(removed_resources)}; } +AddedRemoved +WatchMap::appendWatchInterest(Watch* watch, + const absl::flat_hash_set& resources_to_add) { + absl::flat_hash_set newly_added_to_watch; + for (const auto& name : resources_to_add) { + if (watch->resource_names_.insert(name).second) { + newly_added_to_watch.insert(name); + } + } + + updateWildcardWatches(watch); + + absl::flat_hash_set added_resources = findAdditions(newly_added_to_watch, watch); + return {std::move(added_resources), {}}; +} + +void WatchMap::accept(Watch* watch, const absl::flat_hash_set& patterns) { + // Forget any previous other-resource interest, so we can precisely update the reverse-index. + forgetAccept(watch); + watch->accept_patterns_.clear(); + + for (absl::string_view pattern : patterns) { + if (pattern == Wildcard) { + accept_wildcard_watches_.insert(watch); + } else if (absl::EndsWith(pattern, "/*")) { + // A "/*" glob: accept every resource named "/...". Keyed by "". + accept_prefix_watches_[pattern.substr(0, pattern.size() - 2)].insert(watch); + } else { + IS_ENVOY_BUG("accept() requires the '*' wildcard or a '/*' glob"); + continue; + } + // Record the pattern on the watch itself (forward index), so it can be cleaned up precisely. + watch->accept_patterns_.insert(std::string(pattern)); + } + + updateWildcardWatches(watch); +} + absl::flat_hash_set WatchMap::watchesInterestedIn(const std::string& resource_name) { - absl::flat_hash_set ret; - if (!use_namespace_matching_) { - ret = wildcard_watches_; + absl::flat_hash_set ret = wildcard_watches_; + // Watches that accepted every other resource via accept("*"). + ret.insert(accept_wildcard_watches_.begin(), accept_wildcard_watches_.end()); + // Watches that accepted this resource's prefix via accept("/*"). + if (!accept_prefix_watches_.empty()) { + const std::string prefix = namespaceFromName(resource_name); + if (!prefix.empty()) { + if (auto it = accept_prefix_watches_.find(prefix); it != accept_prefix_watches_.end()) { + ret.insert(it->second.begin(), it->second.end()); + } + } } const bool is_xdstp = XdsResourceIdentifier::hasXdsTpScheme(resource_name); xds::core::v3::ResourceName xdstp_resource; @@ -105,9 +207,7 @@ absl::flat_hash_set WatchMap::watchesInterestedIn(const std::string& res // only happen for glob collections. TODO(htuch): It should be possible to have much more // efficient matchers here. if (watches_interested == watch_interest_.end()) { - if (use_namespace_matching_) { - watches_interested = watch_interest_.find(namespaceFromName(resource_name)); - } else if (is_xdstp) { + if (is_xdstp) { // Replace resource name component with glob for purpose of matching. const auto pos = xdstp_resource.id().find_last_of('/'); xdstp_resource.set_id(pos == std::string::npos ? "*" diff --git a/source/extensions/config_subscription/grpc/watch_map.h b/source/extensions/config_subscription/grpc/watch_map.h index b9481f59a6424..e628d43eb4939 100644 --- a/source/extensions/config_subscription/grpc/watch_map.h +++ b/source/extensions/config_subscription/grpc/watch_map.h @@ -36,6 +36,12 @@ struct Watch { // Whether the most recent update contained any resources this watch cares about. // If true, a new update that also contains no resources can skip this watch. bool state_of_the_world_empty_{true}; + // Glob patterns this watch accepts via accept() (the "*" wildcard and/or + // "/*" suffix globs). These affect routing only and never the subscription; a watch that + // holds any of them is never treated as a catch-all (empty-interest) wildcard watch. This is the + // forward index; the WatchMap keeps the reverse index in accept_wildcard_watches_ and + // accept_prefix_watches_ (mirroring resource_names_ <-> watch_interest_). + absl::flat_hash_set accept_patterns_; }; // NOTE: Users are responsible for eventually calling removeWatch() on the Watch* returned @@ -72,10 +78,10 @@ struct Watch { // the resource from the cache. class WatchMap : public UntypedConfigUpdateCallbacks, public Logger::Loggable { public: - WatchMap(const bool use_namespace_matching, const std::string& type_url, - CustomConfigValidators* config_validators, EdsResourcesCacheOptRef eds_resources_cache) - : use_namespace_matching_(use_namespace_matching), type_url_(type_url), - config_validators_(config_validators), eds_resources_cache_(eds_resources_cache) { + WatchMap(const std::string& type_url, CustomConfigValidators* config_validators, + EdsResourcesCacheOptRef eds_resources_cache) + : type_url_(type_url), config_validators_(config_validators), + eds_resources_cache_(eds_resources_cache) { // If eds resources cache is provided, then the type must be ClusterLoadAssignment. ASSERT(!eds_resources_cache_.has_value() || (type_url == Config::getTypeUrl())); @@ -94,6 +100,30 @@ class WatchMap : public UntypedConfigUpdateCallbacks, public Logger::Loggable& update_to_these_names); + // Additive counterpart to updateWatchInterest(): adds 'resources_to_add' to the watch's interest + // without disturbing the names it already watches. Returns, in added_, the names that are new + // across the entire subscription (the caller should subscribe to these); removed_ is always + // empty. A name of the form "/*" registers a suffix-glob interest: the watch will then + // receive every resource named "/". This is how a watch can accept resources that + // were not named at addWatch() time (e.g. on-demand or VHDS virtual hosts). + AddedRemoved appendWatchInterest(Watch* watch, + const absl::flat_hash_set& resources_to_add); + + // Declares the routing-acceptance patterns for the given watch (replacing any previously declared + // patterns). These are evaluated FIRST and take precedence: a watch that has declared any accept + // pattern is routed only the resources matching 'patterns' (plus the concrete names it watches), + // and is never treated as a catch-all wildcard watch -- even if its resource_names_ is empty or + // contains "*". This lets a watch subscribe to the wildcard on the wire while routing only a + // subset (e.g. VHDS virtual hosts under a route configuration). To keep wildcard routing, include + // "*" in the patterns. + // + // Each pattern must be the wildcard "*" (accept every resource) or a "/*" glob (accept + // every resource named "/..."). Unlike updateWatchInterest()/appendWatchInterest(), this + // affects routing ONLY: the patterns are held in dedicated structures, are never added to + // watch_interest_, and never contribute to the subscription sent to the management server. + // Passing an empty set clears the previously declared patterns and restores default routing. + void accept(Watch* watch, const absl::flat_hash_set& patterns); + // Expects that the watch to be removed has already had all of its resource names removed via // updateWatchInterest(). void removeWatch(Watch* watch); @@ -121,6 +151,17 @@ class WatchMap : public UntypedConfigUpdateCallbacks, public Logger::Loggable @@ -149,7 +190,14 @@ class WatchMap : public UntypedConfigUpdateCallbacks, public Logger::Loggable> watch_interest_; - const bool use_namespace_matching_; + // Glob interest registered via accept(). Unlike wildcard_watches_/watch_interest_ + // above, these affect routing only and never contribute to the subscription. Parallel to + // wildcard_watches_: watches that accept every resource (registered with "*"). + absl::flat_hash_set accept_wildcard_watches_; + // Parallel to watch_interest_: maps a name prefix P (from a "P/*" pattern) to the watches that + // accept every resource named "P/...". + absl::flat_hash_map> accept_prefix_watches_; + const std::string type_url_; CustomConfigValidators* config_validators_; EdsResourcesCacheOptRef eds_resources_cache_; diff --git a/source/extensions/config_subscription/grpc/xds_mux/grpc_mux_impl.cc b/source/extensions/config_subscription/grpc/xds_mux/grpc_mux_impl.cc index cb221ac8891e0..243e04a2968af 100644 --- a/source/extensions/config_subscription/grpc/xds_mux/grpc_mux_impl.cc +++ b/source/extensions/config_subscription/grpc/xds_mux/grpc_mux_impl.cc @@ -147,9 +147,8 @@ Config::GrpcMuxWatchPtr GrpcMuxImpl::addWatch( // We don't yet have a subscription for type_url! Make one! watch_map = watch_maps_ - .emplace(type_url, - std::make_unique(options.use_namespace_matching_, type_url, - config_validators_.get(), resources_cache)) + .emplace(type_url, std::make_unique( + type_url, config_validators_.get(), resources_cache)) .first; subscriptions_.emplace(type_url, subscription_state_factory_->makeSubscriptionState( type_url, *watch_maps_[type_url], resource_decoder, @@ -168,14 +167,9 @@ Config::GrpcMuxWatchPtr GrpcMuxImpl::addWatch( // 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. template -void GrpcMuxImpl::updateWatch(const std::string& type_url, Watch* watch, - const absl::flat_hash_set& resources, - const SubscriptionOptions& options) { - ENVOY_LOG(debug, "GrpcMuxImpl::updateWatch for {}", type_url); - ASSERT(watch != nullptr); - auto& sub = subscriptionStateFor(type_url); - WatchMap& watch_map = watchMapFor(type_url); - +absl::flat_hash_set +GrpcMuxImpl::effectiveResources(const absl::flat_hash_set& 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 effective_resources; @@ -196,6 +190,20 @@ void GrpcMuxImpl::updateWatch(const std::string& type_url, Watch* effective_resources.insert(resource); } } + return effective_resources; +} + +template +void GrpcMuxImpl::updateWatch(const std::string& type_url, Watch* watch, + const absl::flat_hash_set& resources, + const SubscriptionOptions& options) { + ENVOY_LOG(debug, "GrpcMuxImpl::updateWatch for {}", type_url); + ASSERT(watch != nullptr); + auto& sub = subscriptionStateFor(type_url); + WatchMap& watch_map = watchMapFor(type_url); + + const absl::flat_hash_set effective_resources = + effectiveResources(resources, options); auto added_removed = watch_map.updateWatchInterest(watch, effective_resources); if (xds_config_tracker_.has_value() && !added_removed.removed_.empty()) { @@ -207,12 +215,7 @@ void GrpcMuxImpl::updateWatch(const std::string& type_url, Watch* 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.updateSubscriptionInterest({}, {}); - } else { - sub.updateSubscriptionInterest(added_removed.added_, added_removed.removed_); - } + sub.updateSubscriptionInterest(added_removed.added_, added_removed.removed_); // Tell the server about our change in interest, if any. if (sub.subscriptionUpdatePending()) { @@ -220,6 +223,30 @@ void GrpcMuxImpl::updateWatch(const std::string& type_url, Watch* } } +template +void GrpcMuxImpl::appendWatch(const std::string& type_url, Watch* watch, + const absl::flat_hash_set& resources, + const SubscriptionOptions& options) { + ASSERT(watch != nullptr); + auto& sub = subscriptionStateFor(type_url); + // Additionally update the watch-map routing, then subscribe to whatever became newly interesting + // across the whole subscription. This keeps the watch map and the subscription consistent. + auto added_removed = + watchMapFor(type_url).appendWatchInterest(watch, effectiveResources(resources, options)); + sub.updateSubscriptionInterest(added_removed.added_, {}); + if (sub.subscriptionUpdatePending()) { + trySendDiscoveryRequests(); + } +} + +template +void GrpcMuxImpl::accept(const std::string& type_url, Watch* watch, + const absl::flat_hash_set& patterns) { + ASSERT(watch != nullptr); + // Glob interest affects routing only; the subscription sent to the server is left untouched. + watchMapFor(type_url).accept(watch, patterns); +} + template void GrpcMuxImpl::removeWatch(const std::string& type_url, Watch* watch) { updateWatch(type_url, watch, {}, {}); @@ -479,17 +506,6 @@ GrpcMuxDelta::GrpcMuxDelta(GrpcMuxContext& grpc_mux_context) : GrpcMuxImpl(std::make_unique(grpc_mux_context.dispatcher_), grpc_mux_context) {} -// GrpcStreamCallbacks for GrpcMuxDelta -void GrpcMuxDelta::requestOnDemandUpdate(const std::string& type_url, - const absl::flat_hash_set& for_update) { - auto& sub = subscriptionStateFor(type_url); - sub.updateSubscriptionInterest(for_update, {}); - // Tell the server about our change in interest, if any. - if (sub.subscriptionUpdatePending()) { - trySendDiscoveryRequests(); - } -} - GrpcMuxSotw::GrpcMuxSotw(GrpcMuxContext& grpc_mux_context) : GrpcMuxImpl(std::make_unique(grpc_mux_context.dispatcher_), grpc_mux_context) {} diff --git a/source/extensions/config_subscription/grpc/xds_mux/grpc_mux_impl.h b/source/extensions/config_subscription/grpc/xds_mux/grpc_mux_impl.h index a2680147016db..d31e2aff36eca 100644 --- a/source/extensions/config_subscription/grpc/xds_mux/grpc_mux_impl.h +++ b/source/extensions/config_subscription/grpc/xds_mux/grpc_mux_impl.h @@ -84,6 +84,15 @@ class GrpcMuxImpl : public GrpcStreamCallbacks, void updateWatch(const std::string& type_url, Watch* watch, const absl::flat_hash_set& resources, const SubscriptionOptions& options); + // Additionally adds resources to the given watch's interest, updating both the watch-map routing + // and the subscription (see GrpcMuxWatch::append). + void appendWatch(const std::string& type_url, Watch* watch, + const absl::flat_hash_set& resources, + const SubscriptionOptions& options); + // Registers glob interest for the given watch; routing only, never affects the subscription + // (see GrpcMuxWatch::accept). + void accept(const std::string& type_url, Watch* watch, + const absl::flat_hash_set& patterns); void removeWatch(const std::string& type_url, Watch* watch); ScopedResume pause(const std::string& type_url) override; @@ -148,6 +157,14 @@ class GrpcMuxImpl : public GrpcStreamCallbacks, parent_.updateWatch(type_url_, watch_, resources, options_); } + void append(const absl::flat_hash_set& resources) override { + parent_.appendWatch(type_url_, watch_, resources, options_); + } + + void accept(const absl::flat_hash_set& patterns) override { + parent_.accept(type_url_, watch_, patterns); + } + private: const std::string type_url_; Watch* watch_; @@ -163,6 +180,11 @@ class GrpcMuxImpl : public GrpcStreamCallbacks, S& subscriptionStateFor(const std::string& type_url); WatchMap& watchMapFor(const std::string& type_url); + // Normalizes xdstp:// resource names for the transport (adds extra context parameters and sorts + // them); non-xdstp names are returned unchanged. + absl::flat_hash_set + effectiveResources(const absl::flat_hash_set& resources, + const SubscriptionOptions& options); void handleEstablishedStream(); void handleStreamEstablishmentFailure(bool next_attempt_may_send_initial_resource_version); // May modify the order of the resources in response_proto to put all the @@ -264,10 +286,6 @@ class GrpcMuxDelta : public GrpcMuxImpl& for_update) override; - private: absl::string_view methodName() const override { return "envoy.service.discovery.v3.AggregatedDiscoveryService.DeltaAggregatedResources"; @@ -280,11 +298,6 @@ class GrpcMuxSotw : public GrpcMuxImpl&) override { - ENVOY_BUG(false, "unexpected request for on demand update"); - } - private: absl::string_view methodName() const override { return "envoy.service.discovery.v3.AggregatedDiscoveryService.StreamAggregatedResources"; @@ -313,10 +326,6 @@ class NullGrpcMuxImpl : public GrpcMux { return absl::UnimplementedError(""); } - void requestOnDemandUpdate(const std::string&, const absl::flat_hash_set&) override { - ENVOY_BUG(false, "unexpected request for on demand update"); - } - EdsResourcesCacheOptRef edsResourcesCache() override { return {}; } Upstream::LoadStatsReporter* loadStatsReporter() const override { return nullptr; } diff --git a/source/extensions/config_subscription/rest/http_subscription_impl.h b/source/extensions/config_subscription/rest/http_subscription_impl.h index 9f8009471c65d..52df2f2e3aa33 100644 --- a/source/extensions/config_subscription/rest/http_subscription_impl.h +++ b/source/extensions/config_subscription/rest/http_subscription_impl.h @@ -41,6 +41,7 @@ class HttpSubscriptionImpl : public Http::RestApiFetcher, void requestOnDemandUpdate(const absl::flat_hash_set&) override { ENVOY_BUG(false, "unexpected request for on demand update"); } + void accept(const absl::flat_hash_set&) override {} // Http::RestApiFetcher void createRequest(Http::RequestMessage& request) override; diff --git a/test/extensions/config_subscription/grpc/delta_subscription_impl_test.cc b/test/extensions/config_subscription/grpc/delta_subscription_impl_test.cc index f0d4549ea4c6b..f5e432c867a8a 100644 --- a/test/extensions/config_subscription/grpc/delta_subscription_impl_test.cc +++ b/test/extensions/config_subscription/grpc/delta_subscription_impl_test.cc @@ -68,6 +68,143 @@ TEST_P(DeltaSubscriptionImplTest, ResponseCausesAck) { deliverConfigUpdate({"name1"}, "someversion", true); } +// Regression test for on-demand updates: a resource requested via requestOnDemandUpdate() that was +// not part of the initial (non-wildcard) subscription must still be routed to the subscription's +// callbacks. Previously requestOnDemandUpdate() only updated the subscription sent to the server, +// not the watch-map routing, so the response for such a resource was silently dropped. +TEST_P(DeltaSubscriptionImplTest, OnDemandUpdateRoutesResponseToWatch) { + // A concrete (non-wildcard) initial subscription, as ODCDS uses. + startSubscription({"name1"}); + deliverConfigUpdate({"name1"}, "version1", true); + + // Request a brand-new resource on demand. This must subscribe on the wire... + expectSendMessage({"name2"}, {}, Grpc::Status::WellKnownGrpcStatus::Ok, "", {}); + subscription_->requestOnDemandUpdate({"name2"}); + Mock::VerifyAndClearExpectations(&async_stream_); + + // ...and register watch-map interest, so the server's response for it reaches the callbacks. + const std::string version = "version2"; + auto response = std::make_unique(); + last_response_nonce_ = std::to_string(HashUtil::xxHash64(version)); + response->set_nonce(last_response_nonce_); + response->set_system_version_info(version); + response->set_type_url(Config::TestTypeUrl::get().ClusterLoadAssignment); + envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment; + load_assignment.set_cluster_name("name2"); + auto* resource = response->add_resources(); + resource->set_name("name2"); + resource->set_version(version); + std::ignore = resource->mutable_resource()->PackFrom(load_assignment); + + bool delivered_name2 = false; + EXPECT_CALL(callbacks_, onConfigUpdate(_, _, version)) + .WillOnce(Invoke([&delivered_name2](const std::vector& added_resources, + const Protobuf::RepeatedPtrField&, + const std::string&) { + EXPECT_EQ(1, added_resources.size()); + if (!added_resources.empty()) { + EXPECT_EQ("name2", added_resources[0].get().name()); + } + delivered_name2 = true; + return absl::OkStatus(); + })); + expectSendMessage({}, version); // ACK + onDiscoveryResponse(std::move(response)); + EXPECT_TRUE(delivered_name2); +} + +// Regression test for the VHDS-style pattern: a subscription that accepts a glob prefix for routing +// (accept("/*"), as VHDS does for its route configuration) then requests, on demand, a +// resource that is NOT covered by that glob (here under a different "/" prefix). The response +// for the on-demand resource must still be routed to the callbacks. Before the fix, +// requestOnDemandUpdate() only updated the wire subscription; a resource matched by neither the +// glob nor a start()-time name matched no watch and was silently dropped. +TEST_P(DeltaSubscriptionImplTest, OnDemandUpdateOutsideAcceptGlobRoutesResponse) { + startSubscription({"name1"}); + deliverConfigUpdate({"name1"}, "version1", true); + + // Accept everything under "rc1/" for routing only (no wire subscription); mirrors VHDS. + subscription_->accept({"rc1/*"}); + + // Request an on-demand resource under a DIFFERENT prefix ("rc2/"), i.e. not covered by the glob. + // This must subscribe on the wire... + expectSendMessage({"rc2/vhost"}, {}, Grpc::Status::WellKnownGrpcStatus::Ok, "", {}); + subscription_->requestOnDemandUpdate({"rc2/vhost"}); + Mock::VerifyAndClearExpectations(&async_stream_); + + // ...and register watch-map interest, so the server's response for it reaches the callbacks even + // though it matches neither the "rc1/*" glob nor the initial "name1" subscription. + const std::string version = "version2"; + auto response = std::make_unique(); + last_response_nonce_ = std::to_string(HashUtil::xxHash64(version)); + response->set_nonce(last_response_nonce_); + response->set_system_version_info(version); + response->set_type_url(Config::TestTypeUrl::get().ClusterLoadAssignment); + envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment; + load_assignment.set_cluster_name("rc2/vhost"); + auto* resource = response->add_resources(); + resource->set_name("rc2/vhost"); + resource->set_version(version); + std::ignore = resource->mutable_resource()->PackFrom(load_assignment); + + bool delivered = false; + EXPECT_CALL(callbacks_, onConfigUpdate(_, _, version)) + .WillOnce( + Invoke([&delivered](const std::vector& added_resources, + const Protobuf::RepeatedPtrField&, const std::string&) { + EXPECT_EQ(1, added_resources.size()); + if (!added_resources.empty()) { + EXPECT_EQ("rc2/vhost", added_resources[0].get().name()); + } + delivered = true; + return absl::OkStatus(); + })); + expectSendMessage({}, version); // ACK + onDiscoveryResponse(std::move(response)); + EXPECT_TRUE(delivered); +} + +// accept() may be called before start() (as VHDS does): the patterns are buffered and applied when +// the watch is created in start(), before the gRPC stream starts, so the routing filter is in +// effect before any resource can be delivered. Verifies the buffered accept() is actually applied. +TEST_P(DeltaSubscriptionImplTest, AcceptBeforeStartIsApplied) { + // Declare the accept glob BEFORE start(); the watch does not exist yet, so this is buffered. + subscription_->accept({"rc1/*"}); + startSubscription({"name1"}); + deliverConfigUpdate({"name1"}, "version1", true); + + // A resource under the accepted "rc1/*" glob is routed even though it was never explicitly + // requested -- proving the buffered accept() was applied during start(). + const std::string version = "version2"; + auto response = std::make_unique(); + last_response_nonce_ = std::to_string(HashUtil::xxHash64(version)); + response->set_nonce(last_response_nonce_); + response->set_system_version_info(version); + response->set_type_url(Config::TestTypeUrl::get().ClusterLoadAssignment); + envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment; + load_assignment.set_cluster_name("rc1/foo"); + auto* resource = response->add_resources(); + resource->set_name("rc1/foo"); + resource->set_version(version); + std::ignore = resource->mutable_resource()->PackFrom(load_assignment); + + bool delivered = false; + EXPECT_CALL(callbacks_, onConfigUpdate(_, _, version)) + .WillOnce( + Invoke([&delivered](const std::vector& added_resources, + const Protobuf::RepeatedPtrField&, const std::string&) { + EXPECT_EQ(1, added_resources.size()); + if (!added_resources.empty()) { + EXPECT_EQ("rc1/foo", added_resources[0].get().name()); + } + delivered = true; + return absl::OkStatus(); + })); + expectSendMessage({}, version); // ACK + onDiscoveryResponse(std::move(response)); + EXPECT_TRUE(delivered); +} + // Checks that after a pause(), no ACK requests are sent until resume(), but that after the // resume, *all* ACKs that arrived during the pause are sent (in order). TEST_P(DeltaSubscriptionImplTest, PauseQueuesAcks) { diff --git a/test/extensions/config_subscription/grpc/grpc_mux_impl_test.cc b/test/extensions/config_subscription/grpc/grpc_mux_impl_test.cc index c17591a3bd9d7..ce6782c367349 100644 --- a/test/extensions/config_subscription/grpc/grpc_mux_impl_test.cc +++ b/test/extensions/config_subscription/grpc/grpc_mux_impl_test.cc @@ -1642,11 +1642,6 @@ TEST_F(NullGrpcMuxImplTest, PauseMultipleArgsImplemented) { EXPECT_NO_THROW(scoped = null_mux_.pause(params)); } -TEST_F(NullGrpcMuxImplTest, RequestOnDemandNotImplemented) { - EXPECT_ENVOY_BUG(null_mux_.requestOnDemandUpdate("type_url", {"for_update"}), - "unexpected request for on demand update"); -} - TEST_F(NullGrpcMuxImplTest, AddWatchRaisesException) { NiceMock callbacks; OpaqueResourceDecoderSharedPtr resource_decoder( @@ -1907,12 +1902,6 @@ TEST_P(GrpcMuxImplTest, ShutdownPreventsSending) { auto bar_sub = grpc_mux_->addWatch("bar", {"z"}, callbacks_, resource_decoder_, {}); } -TEST_P(GrpcMuxImplTest, RequestOnDemandUpdateDoesNothing) { - setup(); - // Should not throw or crash. - grpc_mux_->requestOnDemandUpdate("foo", {"z"}); -} - } // namespace } // namespace Config } // namespace Envoy diff --git a/test/extensions/config_subscription/grpc/new_grpc_mux_impl_test.cc b/test/extensions/config_subscription/grpc/new_grpc_mux_impl_test.cc index b2c0fb2caf467..866750f8d90d4 100644 --- a/test/extensions/config_subscription/grpc/new_grpc_mux_impl_test.cc +++ b/test/extensions/config_subscription/grpc/new_grpc_mux_impl_test.cc @@ -530,8 +530,10 @@ TEST_P(NewGrpcMuxImplTest, ConfigUpdateWithAliases) { const std::string& type_url = Config::TestTypeUrl::get().VirtualHost; SubscriptionOptions options; - options.use_namespace_matching_ = true; - auto watch = grpc_mux_->addWatch(type_url, {"prefix"}, callbacks_, resource_decoder_, options); + auto watch = grpc_mux_->addWatch(type_url, {}, callbacks_, resource_decoder_, options); + // Register a suffix-glob watch so resources named "prefix/" are routed here, without + // subscribing to the glob on the wire (mirrors VHDS). + watch->accept({"prefix/*"}); EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(Return(&async_stream_)); grpc_mux_->start(); @@ -567,8 +569,8 @@ TEST_P(NewGrpcMuxImplTest, ConfigUpdateWithNotFoundResponse) { const std::string& type_url = Config::TestTypeUrl::get().VirtualHost; SubscriptionOptions options; - options.use_namespace_matching_ = true; - auto watch = grpc_mux_->addWatch(type_url, {"prefix"}, callbacks_, resource_decoder_, options); + auto watch = grpc_mux_->addWatch(type_url, {}, callbacks_, resource_decoder_, options); + watch->accept({"prefix/*"}); EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(Return(&async_stream_)); grpc_mux_->start(); @@ -688,10 +690,12 @@ TEST_P(NewGrpcMuxImplTest, RequestOnDemandUpdate) { expectSendMessage({.type_url = "foo", .resource_names_subscribe = {"x", "y"}}); grpc_mux_->start(); + // Additionally add "z" to the watch: it is both subscribed on the wire and routed to this watch. expectSendMessage({.type_url = "foo", .resource_names_subscribe = {"z"}}); - grpc_mux_->requestOnDemandUpdate("foo", {"z"}); + foo_sub->append({"z"}); - expectSendMessage({.type_url = "foo", .resource_names_unsubscribe = {"x", "y"}}); + // On watch removal all of its resources -- including the on-demand "z" -- are unsubscribed. + expectSendMessage({.type_url = "foo", .resource_names_unsubscribe = {"x", "y", "z"}}); } TEST_P(NewGrpcMuxImplTest, Shutdown) { diff --git a/test/extensions/config_subscription/grpc/watch_map_test.cc b/test/extensions/config_subscription/grpc/watch_map_test.cc index 4f094b4c600c1..faf3d93255262 100644 --- a/test/extensions/config_subscription/grpc/watch_map_test.cc +++ b/test/extensions/config_subscription/grpc/watch_map_test.cc @@ -132,7 +132,7 @@ TEST(WatchMapTest, Basic) { TestUtility::TestOpaqueResourceDecoderImpl resource_decoder("cluster_name"); NiceMock config_validators; - WatchMap watch_map(false, "ClusterLoadAssignmentType", &config_validators, {}); + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); Watch* watch = watch_map.addWatch(callbacks, resource_decoder); { @@ -206,7 +206,7 @@ TEST(WatchMapTest, Overlap) { TestUtility::TestOpaqueResourceDecoderImpl resource_decoder("cluster_name"); NiceMock config_validators; - WatchMap watch_map(false, "ClusterLoadAssignmentType", &config_validators, {}); + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); Watch* watch1 = watch_map.addWatch(callbacks1, resource_decoder); Watch* watch2 = watch_map.addWatch(callbacks2, resource_decoder); @@ -275,7 +275,7 @@ TEST(WatchMapTest, CacheResourceAddResource) { NiceMock eds_resources_cache; const std::string eds_type_url = Config::getTypeUrl(); - WatchMap watch_map(false, eds_type_url, &config_validators, + WatchMap watch_map(eds_type_url, &config_validators, makeOptRef(eds_resources_cache)); // The test uses 2 watchers to ensure that interest is kept regardless of // which watcher was the first to add a watch for the assignment. @@ -356,7 +356,7 @@ TEST(WatchMapTest, CacheResourceAddResource) { // WatchMap defers deletes and doesn't crash. class SameWatchRemoval : public testing::Test { public: - SameWatchRemoval() : watch_map_(false, "ClusterLoadAssignmentType", &config_validators, {}) {} + SameWatchRemoval() : watch_map_("ClusterLoadAssignmentType", &config_validators, {}) {} void SetUp() override { envoy::config::endpoint::v3::ClusterLoadAssignment alice; @@ -371,6 +371,11 @@ class SameWatchRemoval : public testing::Test { absl::Status removeAllInterest() { EXPECT_FALSE(watch_cb_invoked_); watch_cb_invoked_ = true; + // removeWatch() requires the watch's resource names to have been drained first (see the + // contract on WatchMap::removeWatch); the real muxes do this via updateWatch(..., {}, {}). + // Draining here also avoids leaving dangling watch pointers in watch_interest_. + watch_map_.updateWatchInterest(watch1_, {}); + watch_map_.updateWatchInterest(watch2_, {}); watch_map_.removeWatch(watch1_); watch_map_.removeWatch(watch2_); return absl::OkStatus(); @@ -436,7 +441,7 @@ TEST(WatchMapTest, AddRemoveAdd) { TestUtility::TestOpaqueResourceDecoderImpl resource_decoder("cluster_name"); NiceMock config_validators; - WatchMap watch_map(false, "ClusterLoadAssignmentType", &config_validators, {}); + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); Watch* watch1 = watch_map.addWatch(callbacks1, resource_decoder); Watch* watch2 = watch_map.addWatch(callbacks2, resource_decoder); @@ -493,7 +498,7 @@ TEST(WatchMapTest, UninterestingUpdate) { TestUtility::TestOpaqueResourceDecoderImpl resource_decoder("cluster_name"); NiceMock config_validators; - WatchMap watch_map(false, "ClusterLoadAssignmentType", &config_validators, {}); + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); Watch* watch = watch_map.addWatch(callbacks, resource_decoder); watch_map.updateWatchInterest(watch, {"alice"}); @@ -538,7 +543,7 @@ TEST(WatchMapTest, WatchingEverything) { TestUtility::TestOpaqueResourceDecoderImpl resource_decoder("cluster_name"); NiceMock config_validators; - WatchMap watch_map(false, "ClusterLoadAssignmentType", &config_validators, {}); + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); /*Watch* watch1 = */ watch_map.addWatch(callbacks1, resource_decoder); Watch* watch2 = watch_map.addWatch(callbacks2, resource_decoder); // watch1 never specifies any names, and so is treated as interested in everything. @@ -575,7 +580,7 @@ TEST(WatchMapTest, DeltaOnConfigUpdate) { TestUtility::TestOpaqueResourceDecoderImpl resource_decoder("cluster_name"); NiceMock config_validators; - WatchMap watch_map(false, "ClusterLoadAssignmentType", &config_validators, {}); + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); Watch* watch1 = watch_map.addWatch(callbacks1, resource_decoder); Watch* watch2 = watch_map.addWatch(callbacks2, resource_decoder); Watch* watch3 = watch_map.addWatch(callbacks3, resource_decoder); @@ -609,7 +614,7 @@ TEST(WatchMapTest, DeltaOnConfigUpdate) { TEST(WatchMapTest, OnConfigUpdateFailed) { NiceMock config_validators; - WatchMap watch_map(false, "ClusterLoadAssignmentType", &config_validators, {}); + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); // calling on empty map doesn't break watch_map.onConfigUpdateFailed(ConfigUpdateFailureReason::UpdateRejected, nullptr); @@ -631,7 +636,7 @@ TEST(WatchMapTest, OnConfigUpdateXdsTpGlobCollections) { TestUtility::TestOpaqueResourceDecoderImpl resource_decoder("cluster_name"); NiceMock config_validators; - WatchMap watch_map(false, "ClusterLoadAssignmentType", &config_validators, {}); + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); Watch* watch = watch_map.addWatch(callbacks, resource_decoder); watch_map.updateWatchInterest(watch, {"xdstp://foo/bar/baz/*?some=thing&thing=some"}); @@ -676,7 +681,7 @@ TEST(WatchMapTest, OnConfigUpdateXdsTpSingletons) { TestUtility::TestOpaqueResourceDecoderImpl resource_decoder("cluster_name"); NiceMock config_validators; - WatchMap watch_map(false, "ClusterLoadAssignmentType", &config_validators, {}); + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); Watch* watch = watch_map.addWatch(callbacks, resource_decoder); watch_map.updateWatchInterest(watch, {"xdstp://foo/bar/baz?some=thing&thing=some"}); @@ -710,20 +715,21 @@ TEST(WatchMapTest, OnConfigUpdateXdsTpSingletons) { } } -TEST(WatchMapTest, OnConfigUpdateUsingNamespaces) { +TEST(WatchMapTest, OnConfigUpdateUsingSuffixGlobs) { MockSubscriptionCallbacks callbacks1; MockSubscriptionCallbacks callbacks2; MockSubscriptionCallbacks callbacks3; TestUtility::TestOpaqueResourceDecoderImpl resource_decoder("cluster_name"); NiceMock config_validators; - WatchMap watch_map(true, "ClusterLoadAssignmentType", &config_validators, {}); + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); Watch* watch1 = watch_map.addWatch(callbacks1, resource_decoder); Watch* watch2 = watch_map.addWatch(callbacks2, resource_decoder); Watch* watch3 = watch_map.addWatch(callbacks3, resource_decoder); - watch_map.updateWatchInterest(watch1, {"ns1"}); - watch_map.updateWatchInterest(watch2, {"ns1", "ns2"}); - watch_map.updateWatchInterest(watch3, {"ns3"}); + // Watches register suffix-glob interest in resources named "/" via accept. + watch_map.accept(watch1, {"ns1/*"}); + watch_map.accept(watch2, {"ns1/*", "ns2/*"}); + watch_map.accept(watch3, {"ns3/*"}); // verify update { @@ -771,6 +777,107 @@ TEST(WatchMapTest, OnConfigUpdateUsingNamespaces) { } } +// appendWatchInterest() adds to a watch's interest without disturbing the names it already +// watches, and only reports names that are new across the whole subscription in added_. +TEST(WatchMapTest, AppendWatchInterest) { + MockSubscriptionCallbacks callbacks; + TestUtility::TestOpaqueResourceDecoderImpl + resource_decoder("cluster_name"); + NiceMock config_validators; + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); + Watch* watch = watch_map.addWatch(callbacks, resource_decoder); + + { + // Start off interested in Alice. + AddedRemoved added_removed = watch_map.updateWatchInterest(watch, {"alice"}); + EXPECT_EQ(absl::flat_hash_set({"alice"}), added_removed.added_); + EXPECT_TRUE(added_removed.removed_.empty()); + } + { + // Appending Bob keeps Alice and reports only Bob as new to the subscription. + AddedRemoved added_removed = watch_map.appendWatchInterest(watch, {"bob"}); + EXPECT_EQ(absl::flat_hash_set({"bob"}), added_removed.added_); + EXPECT_TRUE(added_removed.removed_.empty()); + } + { + // Appending an already-watched name (Alice) plus a new one (Carol) reports only Carol. + AddedRemoved added_removed = watch_map.appendWatchInterest(watch, {"alice", "carol"}); + EXPECT_EQ(absl::flat_hash_set({"carol"}), added_removed.added_); + EXPECT_TRUE(added_removed.removed_.empty()); + } + { + // The watch now receives updates for Alice, Bob, and Carol (all still watched). + Protobuf::RepeatedPtrField updated_resources; + for (const auto& name : {"alice", "bob", "carol"}) { + envoy::config::endpoint::v3::ClusterLoadAssignment cla; + cla.set_cluster_name(name); + std::ignore = updated_resources.Add()->PackFrom(cla); + } + std::vector expected_resources; + for (const auto& name : {"alice", "bob", "carol"}) { + envoy::config::endpoint::v3::ClusterLoadAssignment cla; + cla.set_cluster_name(name); + expected_resources.push_back(cla); + } + expectDeltaUpdate(callbacks, expected_resources, {}, "version1"); + doDeltaUpdate(watch_map, updated_resources, {}, "version1"); + } +} + +// accept() registers glob interest that affects routing only: the watch receives +// resources matching a "/*" glob, is not turned into a catch-all wildcard watch, and the +// glob never enters watch_interest_ (so it never affects the subscription). +TEST(WatchMapTest, AcceptOtherResourcesPrefixGlob) { + MockSubscriptionCallbacks callbacks; + TestUtility::TestOpaqueResourceDecoderImpl + resource_decoder("cluster_name"); + NiceMock config_validators; + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); + Watch* watch = watch_map.addWatch(callbacks, resource_decoder); + + // Accept everything under "prefix/". This must not make the watch a catch-all wildcard. + watch_map.accept(watch, {"prefix/*"}); + + Protobuf::RepeatedPtrField updated_resources; + for (const auto& name : {"prefix/vhost", "other/vhost", "no_prefix"}) { + envoy::config::endpoint::v3::ClusterLoadAssignment cla; + cla.set_cluster_name(name); + std::ignore = updated_resources.Add()->PackFrom(cla); + } + // Only the resource under "prefix/" is delivered; "other/vhost" and "no_prefix" are not (proving + // the watch is glob-scoped, not wildcard). + envoy::config::endpoint::v3::ClusterLoadAssignment matched; + matched.set_cluster_name("prefix/vhost"); + std::vector expected_resources{matched}; + + expectDeltaUpdate(callbacks, expected_resources, {}, "version1"); + doDeltaUpdate(watch_map, updated_resources, {}, "version1"); +} + +// accept("*") makes the watch accept every resource, without an empty +// watch_interest_ entry (i.e. purely for routing). +TEST(WatchMapTest, AcceptOtherResourcesWildcard) { + MockSubscriptionCallbacks callbacks; + TestUtility::TestOpaqueResourceDecoderImpl + resource_decoder("cluster_name"); + NiceMock config_validators; + WatchMap watch_map("ClusterLoadAssignmentType", &config_validators, {}); + Watch* watch = watch_map.addWatch(callbacks, resource_decoder); + + watch_map.accept(watch, {"*"}); + + Protobuf::RepeatedPtrField updated_resources; + std::vector expected_resources; + for (const auto& name : {"anything", "a/b"}) { + envoy::config::endpoint::v3::ClusterLoadAssignment cla; + cla.set_cluster_name(name); + std::ignore = updated_resources.Add()->PackFrom(cla); + expected_resources.push_back(cla); + } + expectDeltaUpdate(callbacks, expected_resources, {}, "version1"); + doDeltaUpdate(watch_map, updated_resources, {}, "version1"); +} + // TODO(adip): Add tests that use the eds cache. // Needs to test the following function onConfigUpdate (sotw&delta) and // updateWatchInterest diff --git a/test/extensions/config_subscription/grpc/xds_grpc_mux_impl_test.cc b/test/extensions/config_subscription/grpc/xds_grpc_mux_impl_test.cc index 637e497714b44..893676b9c6741 100644 --- a/test/extensions/config_subscription/grpc/xds_grpc_mux_impl_test.cc +++ b/test/extensions/config_subscription/grpc/xds_grpc_mux_impl_test.cc @@ -1464,11 +1464,6 @@ TEST_F(NullGrpcMuxImplTest, PauseMultipleArgsImplemented) { EXPECT_NO_THROW(scoped = null_mux_->pause(params)); } -TEST_F(NullGrpcMuxImplTest, RequestOnDemandNotImplemented) { - EXPECT_ENVOY_BUG(null_mux_->requestOnDemandUpdate("type_url", {"for_update"}), - "unexpected request for on demand update"); -} - TEST_F(NullGrpcMuxImplTest, AddWatchRaisesException) { NiceMock callbacks; OpaqueResourceDecoderSharedPtr resource_decoder( diff --git a/test/extensions/filters/http/on_demand/on_demand_integration_test.cc b/test/extensions/filters/http/on_demand/on_demand_integration_test.cc index 67807a5a1104a..a586bec9b2953 100644 --- a/test/extensions/filters/http/on_demand/on_demand_integration_test.cc +++ b/test/extensions/filters/http/on_demand/on_demand_integration_test.cc @@ -824,6 +824,70 @@ TEST_P(OnDemandVhdsIntegrationTest, VhdsWildcardUpgradeOnReconnect) { {"*", "my_route/vhost.first"}, {}, vhds_stream_.get())); } +// End-to-end regression guard for on-demand VHDS after the switch to accept()-based routing: two +// requests to two DIFFERENT unknown virtual hosts each trigger an on-demand VHDS fetch (via +// requestOnDemandUpdate()/append()), and both must be routed and resolved. VHDS declares +// accept("my_route/*") for routing, so on-demand virtual-host names under that route configuration +// are delivered to the subscription and the requests succeed. +TEST_P(OnDemandVhdsIntegrationTest, VhdsTwoOnDemandVirtualHostsRouteResponses) { + testRouterHeaderOnlyRequestAndResponse(nullptr, 1); + cleanupUpstreamAndDownstream(); + ASSERT_TRUE(codec_client_->waitForDisconnect()); + + // First on-demand virtual host: vhost.first. + { + codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http")))); + Http::TestRequestHeaderMapImpl request_headers{{":method", "GET"}, + {":path", "/"}, + {":scheme", "http"}, + {":authority", "vhost.first"}, + {"x-lyft-user-id", "123"}}; + IntegrationStreamDecoderPtr response = codec_client_->makeHeaderOnlyRequest(request_headers); + EXPECT_TRUE(compareDeltaDiscoveryRequest(Config::TestTypeUrl::get().VirtualHost, + {vhdsRequestResourceName("vhost.first")}, {}, + vhds_stream_.get())); + sendDeltaDiscoveryResponse( + Config::TestTypeUrl::get().VirtualHost, {buildVirtualHost2()}, {}, "2", vhds_stream_.get(), + {"my_route/vhost.first"}); + EXPECT_TRUE(compareDeltaDiscoveryRequest(Config::TestTypeUrl::get().VirtualHost, {}, {}, + vhds_stream_.get())); + waitForNextUpstreamRequest(1); + upstream_request_->encodeHeaders(default_response_headers_, true); + response->waitForHeaders(); + EXPECT_EQ("200", response->headers().getStatusValue()); + cleanupUpstreamAndDownstream(); + ASSERT_TRUE(codec_client_->waitForDisconnect()); + } + + // Second on-demand virtual host: vhost.second, a DIFFERENT name -> a second append() on the same + // subscription. Its response must also be routed for the request to succeed. + { + codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http")))); + Http::TestRequestHeaderMapImpl request_headers{{":method", "GET"}, + {":path", "/"}, + {":scheme", "http"}, + {":authority", "vhost.second"}, + {"x-lyft-user-id", "123"}}; + IntegrationStreamDecoderPtr response = codec_client_->makeHeaderOnlyRequest(request_headers); + EXPECT_TRUE(compareDeltaDiscoveryRequest(Config::TestTypeUrl::get().VirtualHost, + {vhdsRequestResourceName("vhost.second")}, {}, + vhds_stream_.get())); + sendDeltaDiscoveryResponse( + Config::TestTypeUrl::get().VirtualHost, + {TestUtility::parseYaml( + virtualHostYaml("my_route/vhost_2", "vhost.second"))}, + {}, "3", vhds_stream_.get(), {"my_route/vhost.second"}); + EXPECT_TRUE(compareDeltaDiscoveryRequest(Config::TestTypeUrl::get().VirtualHost, {}, {}, + vhds_stream_.get())); + waitForNextUpstreamRequest(1); + upstream_request_->encodeHeaders(default_response_headers_, true); + response->waitForHeaders(); + EXPECT_EQ("200", response->headers().getStatusValue()); + cleanupUpstreamAndDownstream(); + ASSERT_TRUE(codec_client_->waitForDisconnect()); + } +} + // Test class for VHDS on-demand updates with request bodies class OnDemandVhdsWithBodyIntegrationTest : public testing::TestWithParam>, diff --git a/test/mocks/config/mocks.h b/test/mocks/config/mocks.h index 72b2fd2133ac3..4277b9671ae31 100644 --- a/test/mocks/config/mocks.h +++ b/test/mocks/config/mocks.h @@ -116,6 +116,7 @@ class MockSubscription : public Subscription { (const absl::flat_hash_set& update_to_these_names)); MOCK_METHOD(void, requestOnDemandUpdate, (const absl::flat_hash_set& add_these_names)); + MOCK_METHOD(void, accept, (const absl::flat_hash_set& patterns)); }; class MockSubscriptionFactory : public SubscriptionFactory { @@ -173,10 +174,6 @@ class MockGrpcMux : public GrpcMux { SubscriptionCallbacks& callbacks, OpaqueResourceDecoderSharedPtr resource_decoder, const SubscriptionOptions& options)); - MOCK_METHOD(void, requestOnDemandUpdate, - (const std::string& type_url, - const absl::flat_hash_set& add_these_names)); - MOCK_METHOD(bool, paused, (const std::string& type_url), (const)); MOCK_METHOD(EdsResourcesCacheOptRef, edsResourcesCache, ()); From 2f37e7b9de98851714d7acb679a27800ccbf92c0 Mon Sep 17 00:00:00 2001 From: wbpcode Date: Sat, 18 Jul 2026 01:55:36 +0000 Subject: [PATCH 3/3] improve coverage Signed-off-by: wbpcode --- .../rest/http_subscription_impl_test.cc | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/test/extensions/config_subscription/rest/http_subscription_impl_test.cc b/test/extensions/config_subscription/rest/http_subscription_impl_test.cc index aacdd8c622bb0..a1a63accabbf6 100644 --- a/test/extensions/config_subscription/rest/http_subscription_impl_test.cc +++ b/test/extensions/config_subscription/rest/http_subscription_impl_test.cc @@ -97,6 +97,17 @@ TEST_F(HttpSubscriptionImplTest, UpdateTimeChangedOnUpdateSuccess) { EXPECT_TRUE(statsAre(3, 2, 0, 0, 0, TEST_TIME_MILLIS + 1, 7148434200721666028, "0")); } +// REST subscription does not support on-demand updates; accept() is a no-op and +// requestOnDemandUpdate() is unexpected. +TEST_F(HttpSubscriptionImplTest, OnDemandUpdateNotSupported) { + startSubscription({"cluster0", "cluster1"}); + // accept() is a no-op for the REST subscription. + subscription_->accept({"cluster0"}); + // requestOnDemandUpdate() is not expected to be called for the REST subscription. + EXPECT_ENVOY_BUG(subscription_->requestOnDemandUpdate({"cluster0"}), + "unexpected request for on demand update"); +} + } // namespace } // namespace Config } // namespace Envoy