xds: remove warming listeners in xDS SOTW updates

include/envoy/server/listener_manager.h

@@ -131,6 +131,16 @@ class ListenerManager {
     All,
   };
 
+  // The types of listeners to be returned from listeners(ListenerState).
+  // An enum instead of enum class so the underlying type is an int and bitwise operations can be
+  // used without casting.
+  enum ListenerState : uint8_t {
+    ACTIVE = 1 << 0,
+    WARMING = 1 << 1,
+    DRAINING = 1 << 2,
+    ALL = ACTIVE | WARMING | DRAINING
+  };
+
   virtual ~ListenerManager() = default;
 
   /**
@@ -161,11 +171,15 @@ class ListenerManager {
   virtual void createLdsApi(const envoy::config::core::v3::ConfigSource& lds_config) PURE;
 
   /**
-   * @return std::vector<std::reference_wrapper<Network::ListenerConfig>> a list of the currently
-   * loaded listeners. Note that this routine returns references to the existing listeners. The
-   * references are only valid in the context of the current call stack and should not be stored.
+   * @param state the type of listener to be returned (defaults to ACTIVE), states can be OR'd
+   * together to return multiple different types
+   * @return std::vector<std::reference_wrapper<Network::ListenerConfig>> a list of currently known
+   * listeners in the requested state. Note that this routine returns references to the existing
+   * listeners. The references are only valid in the context of the current call stack and should
+   * not be stored.
    */
-  virtual std::vector<std::reference_wrapper<Network::ListenerConfig>> listeners() PURE;
+  virtual std::vector<std::reference_wrapper<Network::ListenerConfig>>
+  listeners(ListenerState state = ListenerState::ACTIVE) PURE;
 
   /**
    * @return uint64_t the total number of connections owned by all listeners across all workers.
@@ -223,5 +237,13 @@ class ListenerManager {
   virtual ApiListenerOptRef apiListener() PURE;
 };
 
+// overload operator| to allow ListenerManager::listeners(ListenerState) to be called using a
+// combination of flags, such as listeners(ListenerState::WARMING|ListenerState::ACTIVE)
+constexpr ListenerManager::ListenerState operator|(const ListenerManager::ListenerState lhs,
+                                                   const ListenerManager::ListenerState rhs) {
+  return static_cast<ListenerManager::ListenerState>(static_cast<uint8_t>(lhs) |
+                                                     static_cast<uint8_t>(rhs));
+}
+
 } // namespace Server
 } // namespace Envoy

source/server/lds_api.cc

@@ -97,7 +97,8 @@ void LdsApiImpl::onConfigUpdate(const std::vector<Config::DecodedResourceRef>& r
   // We need to keep track of which listeners need to remove.
   // Specifically, it's [listeners we currently have] - [listeners found in the response].
   absl::node_hash_set<std::string> listeners_to_remove;
-  for (const auto& listener : listener_manager_.listeners()) {
+  for (const auto& listener :
+       listener_manager_.listeners(ListenerManager::WARMING | ListenerManager::ACTIVE)) {
     listeners_to_remove.insert(listener.get().name());
   }
   for (const auto& resource : resources) {

source/server/listener_manager_impl.cc

@@ -639,11 +639,30 @@ ListenerManagerImpl::getListenerByName(ListenerList& listeners, const std::strin
   return ret;
 }
 
-std::vector<std::reference_wrapper<Network::ListenerConfig>> ListenerManagerImpl::listeners() {
+std::vector<std::reference_wrapper<Network::ListenerConfig>>
+ListenerManagerImpl::listeners(ListenerState state) {
   std::vector<std::reference_wrapper<Network::ListenerConfig>> ret;
-  ret.reserve(active_listeners_.size());
-  for (const auto& listener : active_listeners_) {
-    ret.push_back(*listener);
+
+  size_t size = 0;
+  size += state & WARMING ? warming_listeners_.size() : 0;
+  size += state & ACTIVE ? active_listeners_.size() : 0;
+  size += state & DRAINING ? draining_listeners_.size() : 0;
+  ret.reserve(size);
+
+  if (state & WARMING) {
+    for (const auto& listener : warming_listeners_) {
+      ret.push_back(*listener);
+    }
+  }
+  if (state & ACTIVE) {
+    for (const auto& listener : active_listeners_) {
+      ret.push_back(*listener);
+    }
+  }
+  if (state & DRAINING) {
+    for (const auto& draining_listener : draining_listeners_) {
+      ret.push_back(*(draining_listener.listener_));
+    }
   }
   return ret;
 }

source/server/listener_manager_impl.h

@@ -186,7 +186,8 @@ class ListenerManagerImpl : public ListenerManager, Logger::Loggable<Logger::Id:
     ASSERT(lds_api_ == nullptr);
     lds_api_ = factory_.createLdsApi(lds_config);
   }
-  std::vector<std::reference_wrapper<Network::ListenerConfig>> listeners() override;
+  std::vector<std::reference_wrapper<Network::ListenerConfig>>
+  listeners(ListenerState state = ListenerState::ACTIVE) override;
   uint64_t numConnections() const override;
   bool removeListener(const std::string& listener_name) override;
   void startWorkers(GuardDog& guard_dog) override;

test/integration/ads_integration_test.cc

@@ -420,6 +420,68 @@ TEST_P(AdsIntegrationTest, CdsPausedDuringWarming) {
                                       {"warming_cluster_2", "warming_cluster_1"}, {}, {}));
 }
 
+// Validate that warming listeners are removed when left out of SOTW update.
+TEST_P(AdsIntegrationTest, RemoveWarmingListener) {
+  initialize();
+
+  // Send initial configuration to start workers, validate we can process a request.
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Cluster, "", {}, {}, {}, true));
+  sendDiscoveryResponse<envoy::config::cluster::v3::Cluster>(Config::TypeUrl::get().Cluster,
+                                                             {buildCluster("cluster_0")},
+                                                             {buildCluster("cluster_0")}, {}, "1");
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().ClusterLoadAssignment, "",
+                                      {"cluster_0"}, {"cluster_0"}, {}));
+
+  sendDiscoveryResponse<envoy::config::endpoint::v3::ClusterLoadAssignment>(
+      Config::TypeUrl::get().ClusterLoadAssignment, {buildClusterLoadAssignment("cluster_0")},
+      {buildClusterLoadAssignment("cluster_0")}, {}, "1");
+
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Cluster, "1", {}, {}, {}));
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Listener, "", {}, {}, {}));
+  sendDiscoveryResponse<envoy::config::listener::v3::Listener>(
+      Config::TypeUrl::get().Listener, {buildListener("listener_0", "route_config_0")},
+      {buildListener("listener_0", "route_config_0")}, {}, "1");
+
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().ClusterLoadAssignment, "1",
+                                      {"cluster_0"}, {}, {}));
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().RouteConfiguration, "",
+                                      {"route_config_0"}, {"route_config_0"}, {}));
+  sendDiscoveryResponse<envoy::config::route::v3::RouteConfiguration>(
+      Config::TypeUrl::get().RouteConfiguration, {buildRouteConfig("route_config_0", "cluster_0")},
+      {buildRouteConfig("route_config_0", "cluster_0")}, {}, "1");
+
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Listener, "1", {}, {}, {}));
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().RouteConfiguration, "1",
+                                      {"route_config_0"}, {}, {}));
+
+  test_server_->waitForCounterGe("listener_manager.listener_create_success", 1);
+  makeSingleRequest();
+
+  // Send a listener without its route, so it will be added as warming.
+  sendDiscoveryResponse<envoy::config::listener::v3::Listener>(
+      Config::TypeUrl::get().Listener,
+      {buildListener("listener_0", "route_config_0"),
+       buildListener("warming_listener_1", "nonexistent_route")},
+      {buildListener("warming_listener_1", "nonexistent_route")}, {}, "2");
+  test_server_->waitForGaugeEq("listener_manager.total_listeners_warming", 1);
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().RouteConfiguration, "1",
+                                      {"nonexistent_route", "route_config_0"},
+                                      {"nonexistent_route"}, {}));
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Listener, "2", {}, {}, {}));
+
+  // Send a request removing the warming listener.
+  sendDiscoveryResponse<envoy::config::listener::v3::Listener>(
+      Config::TypeUrl::get().Listener, {buildListener("listener_0", "route_config_0")},
+      {buildListener("listener_0", "route_config_0")}, {"warming_listener_1"}, "3");
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().RouteConfiguration, "1",
+                                      {"route_config_0"}, {}, {"nonexistent_route"}));
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Listener, "3", {}, {}, {}));
+
+  // The warming listener should be successfully removed.
+  test_server_->waitForCounterEq("listener_manager.listener_removed", 1);
+  test_server_->waitForGaugeEq("listener_manager.total_listeners_warming", 0);
+}
+
 // Verify cluster warming is finished only on named EDS response.
 TEST_P(AdsIntegrationTest, ClusterWarmingOnNamedResponse) {
   initialize();

test/mocks/server/listener_manager.h

@@ -15,7 +15,8 @@ class MockListenerManager : public ListenerManager {
               (const envoy::config::listener::v3::Listener& config, const std::string& version_info,
                bool modifiable));
   MOCK_METHOD(void, createLdsApi, (const envoy::config::core::v3::ConfigSource& lds_config));
-  MOCK_METHOD(std::vector<std::reference_wrapper<Network::ListenerConfig>>, listeners, ());
+  MOCK_METHOD(std::vector<std::reference_wrapper<Network::ListenerConfig>>, listeners,
+              (ListenerState state));
   MOCK_METHOD(uint64_t, numConnections, (), (const));
   MOCK_METHOD(bool, removeListener, (const std::string& listener_name));
   MOCK_METHOD(void, startWorkers, (GuardDog & guard_dog));

test/server/hot_restarting_parent_test.cc

@@ -36,7 +36,8 @@ TEST_F(HotRestartingParentTest, GetListenSocketsForChildNotFound) {
   MockListenerManager listener_manager;
   std::vector<std::reference_wrapper<Network::ListenerConfig>> listeners;
   EXPECT_CALL(server_, listenerManager()).WillOnce(ReturnRef(listener_manager));
-  EXPECT_CALL(listener_manager, listeners()).WillOnce(Return(listeners));
+  EXPECT_CALL(listener_manager, listeners(ListenerManager::ListenerState::ACTIVE))
+      .WillOnce(Return(listeners));
 
   HotRestartMessage::Request request;
   request.mutable_pass_listen_socket()->set_address("tcp://127.0.0.1:80");
@@ -51,7 +52,8 @@ TEST_F(HotRestartingParentTest, GetListenSocketsForChildNotBindPort) {
   InSequence s;
   listeners.push_back(std::ref(*static_cast<Network::ListenerConfig*>(&listener_config)));
   EXPECT_CALL(server_, listenerManager()).WillOnce(ReturnRef(listener_manager));
-  EXPECT_CALL(listener_manager, listeners()).WillOnce(Return(listeners));
+  EXPECT_CALL(listener_manager, listeners(ListenerManager::ListenerState::ACTIVE))
+      .WillOnce(Return(listeners));
   EXPECT_CALL(listener_config, listenSocketFactory());
   EXPECT_CALL(listener_config.socket_factory_, localAddress());
   EXPECT_CALL(listener_config, bindToPort()).WillOnce(Return(false));

test/server/lds_api_test.cc

@@ -76,7 +76,8 @@ class LdsApiTest : public testing::Test {
       listeners_.back().name_ = name;
       refs.emplace_back(listeners_.back());
     }
-    EXPECT_CALL(listener_manager_, listeners()).WillOnce(Return(refs));
+    EXPECT_CALL(listener_manager_, listeners(ListenerManager::WARMING | ListenerManager::ACTIVE))
+        .WillOnce(Return(refs));
     EXPECT_CALL(listener_manager_, beginListenerUpdate());
   }
 
@@ -120,7 +121,8 @@ TEST_F(LdsApiTest, MisconfiguredListenerNameIsPresentInException) {
   socket_address->set_port_value(1);
   listener.add_filter_chains();
 
-  EXPECT_CALL(listener_manager_, listeners()).WillOnce(Return(existing_listeners));
+  EXPECT_CALL(listener_manager_, listeners(ListenerManager::WARMING | ListenerManager::ACTIVE))
+      .WillOnce(Return(existing_listeners));
 
   EXPECT_CALL(listener_manager_, beginListenerUpdate());
   EXPECT_CALL(listener_manager_, addOrUpdateListener(_, _, true))
@@ -141,7 +143,8 @@ TEST_F(LdsApiTest, EmptyListenersUpdate) {
 
   std::vector<std::reference_wrapper<Network::ListenerConfig>> existing_listeners;
 
-  EXPECT_CALL(listener_manager_, listeners()).WillOnce(Return(existing_listeners));
+  EXPECT_CALL(listener_manager_, listeners(ListenerManager::WARMING | ListenerManager::ACTIVE))
+      .WillOnce(Return(existing_listeners));
   EXPECT_CALL(listener_manager_, beginListenerUpdate());
   EXPECT_CALL(listener_manager_, endListenerUpdate(_))
       .WillOnce(Invoke([](ListenerManager::FailureStates&& state) { EXPECT_EQ(0, state.size()); }));
@@ -164,7 +167,8 @@ TEST_F(LdsApiTest, ListenerCreationContinuesEvenAfterException) {
   const auto listener_2 = buildListener("valid-listener-2");
   const auto listener_3 = buildListener("invalid-listener-2");
 
-  EXPECT_CALL(listener_manager_, listeners()).WillOnce(Return(existing_listeners));
+  EXPECT_CALL(listener_manager_, listeners(ListenerManager::WARMING | ListenerManager::ACTIVE))
+      .WillOnce(Return(existing_listeners));
 
   EXPECT_CALL(listener_manager_, beginListenerUpdate());
   EXPECT_CALL(listener_manager_, addOrUpdateListener(_, _, true))
@@ -195,7 +199,8 @@ TEST_F(LdsApiTest, ValidateDuplicateListeners) {
   const auto listener = buildListener("duplicate_listener");
 
   std::vector<std::reference_wrapper<Network::ListenerConfig>> existing_listeners;
-  EXPECT_CALL(listener_manager_, listeners()).WillOnce(Return(existing_listeners));
+  EXPECT_CALL(listener_manager_, listeners(ListenerManager::WARMING | ListenerManager::ACTIVE))
+      .WillOnce(Return(existing_listeners));
   EXPECT_CALL(listener_manager_, beginListenerUpdate());
   EXPECT_CALL(listener_manager_, addOrUpdateListener(_, _, true)).WillOnce(Return(true));
   EXPECT_CALL(listener_manager_, endListenerUpdate(_));
@@ -347,7 +352,7 @@ version_info: '1'
       address: tcp://0.0.0.1
       port_value: 61000
   filter_chains:
-  - filters: 
+  - filters:
   )EOF";
   auto response1 =
       TestUtility::parseYaml<envoy::service::discovery::v3::DiscoveryResponse>(response1_yaml);