From 8b2da558e690c9ec05024969a90cc355110f9104 Mon Sep 17 00:00:00 2001 From: Kateryna Nezdolii Date: Thu, 6 Aug 2026 13:49:29 +0000 Subject: [PATCH] envoy: keep ADS policy maps alive NPDS and NPHDS share the ADS stream with LDS. Removing the final listener used to destroy their maps while the stream kept the subscriptions, so later policy responses were treated as unwatched. Pin both maps for the ADS stream lifetime so policy updates remain subscribed and processed. Signed-off-by: Kateryna Nezdolii --- cilium/bpf_metadata.cc | 11 ++++- tests/bpf_metadata_integration_test.cc | 58 ++++++++++++++++++++++++-- 2 files changed, 64 insertions(+), 5 deletions(-) diff --git a/cilium/bpf_metadata.cc b/cilium/bpf_metadata.cc index 365aec4fa..cb0c8817c 100644 --- a/cilium/bpf_metadata.cc +++ b/cilium/bpf_metadata.cc @@ -244,13 +244,19 @@ Config::Config(const ::cilium::BpfMetadata& config, fmt::format("cilium.bpf_metadata: ipv6_source_address is not an IPv6 address: {}", config.ipv6_source_address())); } + // Keep policy maps alive with the shared ADS stream to avoid leaving stale + // wildcard subscriptions when the last filter instance is destroyed. + const bool pin_for_ads = + config_source_.config_source_specifier_case() == envoy::config::core::v3::ConfigSource::kAds; + if (config.use_nphds()) { hosts_ = context.serverFactoryContext().singletonManager().getTyped( SINGLETON_MANAGER_REGISTERED_NAME(cilium_host_map), [&context, config_source = config_source_] { return std::make_shared(context.serverFactoryContext(), config_source); - }); + }, + pin_for_ads); // update desired config source on the map hosts_->configure(config_source_); } @@ -293,7 +299,8 @@ Config::Config(const ::cilium::BpfMetadata& config, SINGLETON_MANAGER_REGISTERED_NAME(cilium_network_policy), [&context, config_source = config_source_] { return std::make_shared(context, config_source); - }); + }, + pin_for_ads); // update desired config source on the map npmap_->configure(config_source_); } diff --git a/tests/bpf_metadata_integration_test.cc b/tests/bpf_metadata_integration_test.cc index 6ed243426..fd1d7189c 100644 --- a/tests/bpf_metadata_integration_test.cc +++ b/tests/bpf_metadata_integration_test.cc @@ -416,14 +416,19 @@ class BpfMetadataIntegrationTest : public BaseIntegrationTest, stream.sendGrpcMessage(response); } - void sendNphdsResponse(FakeStream& stream, const std::string& version) { + void sendNphdsResponse(FakeStream& stream, const std::string& version, + const std::vector& policy_host_configs = {policy_host1, + policy_host2}) { envoy::service::discovery::v3::DiscoveryResponse response; response.set_version_info(version); response.set_nonce(version); response.set_type_url(NetworkPolicyHostsTypeUrl); std::vector proto_configs; - proto_configs.emplace_back(TestUtility::parseYaml(policy_host1)); - proto_configs.emplace_back(TestUtility::parseYaml(policy_host2)); + proto_configs.reserve(policy_host_configs.size()); + for (const auto& policy_host_config : policy_host_configs) { + proto_configs.emplace_back( + TestUtility::parseYaml(policy_host_config)); + } for (const auto& policy_host_config : proto_configs) { response.add_resources()->PackFrom(policy_host_config); } @@ -597,6 +602,53 @@ TEST_P(BpfMetadataIntegrationTest, BpfMetadataWithNpdsAndNpdhsViaAds) { test_server_->waitForCounterGe("cilium.hostmap.update_success", 1); } +TEST_P(BpfMetadataIntegrationTest, AdsPolicyMapsSurviveLastListenerRemoval) { + on_server_init_function_ = [&]() { + createAdsStream(); + addBpfMetadataListenerFilter(listener_config_, /*use_ads=*/true); + EXPECT_TRUE(compareDiscoveryRequest( + Config::TestTypeUrl::get().Cluster, "", {}, {}, {}, + /*expect_node=*/true, Envoy::Grpc::Status::WellKnownGrpcStatus::Ok, "", ads_stream_.get())); + sendCdsResponse(*ads_stream_, "1"); + EXPECT_TRUE(compareDiscoveryRequest( + Config::TestTypeUrl::get().Listener, "", {}, {}, {}, /*expect_node=*/false, + Grpc::Status::WellKnownGrpcStatus::Ok, "", ads_stream_.get())); + sendLdsResponse(*ads_stream_, {MessageUtil::getYamlStringFromMessage(listener_config_)}, "1"); + }; + initializeAds(); + + test_server_->waitForCounterGe("listener_manager.lds.update_success", 1); + sendNpdsResponse(*ads_stream_, "1"); + test_server_->waitForCounterGe("cilium.policy.update_success", 1); + sendNphdsResponse(*ads_stream_, "1"); + test_server_->waitForCounterGe("cilium.hostmap.update_success", 1); + + { + const auto policy_map = networkPolicyMap(); + EXPECT_TRUE(policy_map->exists("10.1.1.1")); + EXPECT_TRUE(policy_map->exists("10.2.2.2")); + } + EXPECT_EQ(resolveHostPolicyId("10.1.1.1"), 111); + EXPECT_EQ(resolveHostPolicyId("10.2.2.2"), 222); + + sendLdsResponse(*ads_stream_, std::vector{}, "2"); + test_server_->waitForCounterGe("listener_manager.lds.update_success", 2); + test_server_->waitForCounterEq("listener_manager.listener_removed", 1); + test_server_->waitForGaugeEq("listener_manager.total_listeners_draining", 0); + EXPECT_TRUE(test_server_->server().listenerManager().listeners().empty()); + + sendNpdsResponse(*ads_stream_, "2", {policy2}); + test_server_->waitForCounterGe("cilium.policy.update_success", 2); + sendNphdsResponse(*ads_stream_, "2", {policy_host2}); + test_server_->waitForCounterGe("cilium.hostmap.update_success", 2); + + const auto policy_map = networkPolicyMap(); + EXPECT_FALSE(policy_map->exists("10.1.1.1")); + EXPECT_TRUE(policy_map->exists("10.2.2.2")); + EXPECT_EQ(resolveHostPolicyId("10.1.1.1"), Cilium::ID::UNKNOWN); + EXPECT_EQ(resolveHostPolicyId("10.2.2.2"), 222); +} + TEST_P(BpfMetadataIntegrationTest, PolicyStreamGenerationTracksAcceptedAdsGrpcStreams) { on_server_init_function_ = [&]() { createAdsStream();