listener: fix inplace update with conn balancer

docs/root/version_history/current.rst

@@ -28,8 +28,9 @@ Minor Behavior Changes
 * http: fixed the 100-continue response path to properly handle upstream failure by sending 5xx responses. This behavior can be temporarily reverted by setting `envoy.reloadable_features.allow_500_after_100` to false.
 * http: the per-stream FilterState maintained by the HTTP connection manager will now provide read/write access to the downstream connection FilterState. As such, code that relies on interacting with this might
   see a change in behavior.
-* logging: add fine-grain logging for file level log control with logger management at administration interface. It can be enabled by option `--enable-fine-grain-logging`.
-* logging: change default log format to `"[%Y-%m-%d %T.%e][%t][%l][%n] [%g:%#] %v"` and default value of :option:`--log-format-prefix-with-location` to `0`.
+* listener: fixed crash at listener inplace update when connetion load balancer is set.
+* logging: added fine-grain logging for file level log control with logger management at administration interface. It can be enabled by option `--enable-fine-grain-logging`.
+* logging: changed default log format to `"[%Y-%m-%d %T.%e][%t][%l][%n] [%g:%#] %v"` and default value of :option:`--log-format-prefix-with-location` to `0`.
 * logging: nghttp2 log messages no longer appear at trace level unless `ENVOY_NGHTTP2_TRACE` is set
   in the environment.
 * router: added transport failure reason to response body when upstream reset happens. After this change, the response body will be of the form `upstream connect error or disconnect/reset before headers. reset reason:{}, transport failure reason:{}`.This behavior may be reverted by setting runtime feature `envoy.reloadable_features.http_transport_failure_reason_in_body` to false.

include/envoy/network/connection_balancer.h

@@ -65,7 +65,7 @@ class ConnectionBalancer {
   pickTargetHandler(BalancedConnectionHandler& current_handler) PURE;
 };
 
-using ConnectionBalancerPtr = std::unique_ptr<ConnectionBalancer>;
+using ConnectionBalancerSharedPtr = std::shared_ptr<ConnectionBalancer>;
 
 } // namespace Network
 } // namespace Envoy

source/server/connection_handler_impl.cc

@@ -165,6 +165,7 @@ ConnectionHandlerImpl::ActiveTcpListener::ActiveTcpListener(ConnectionHandlerImp
 void ConnectionHandlerImpl::ActiveTcpListener::updateListenerConfig(
     Network::ListenerConfig& config) {
   ENVOY_LOG(trace, "replacing listener ", config_->listenerTag(), " by ", config.listenerTag());
+  ASSERT(&config_->connectionBalancer() == &config.connectionBalancer());
   config_ = &config;
 }
 

source/server/listener_impl.cc

@@ -325,6 +325,7 @@ ListenerImpl::ListenerImpl(ListenerImpl& origin,
       listener_filters_timeout_(
           PROTOBUF_GET_MS_OR_DEFAULT(config, listener_filters_timeout, 15000)),
       continue_on_listener_filters_timeout_(config.continue_on_listener_filters_timeout()),
+      connection_balancer_(origin.connection_balancer_),
       listener_factory_context_(std::make_shared<PerListenerFactoryContextImpl>(
           origin.listener_factory_context_->listener_factory_context_base_, this, *this)),
       filter_chain_manager_(address_, origin.listener_factory_context_->parentFactoryContext(),
@@ -488,12 +489,15 @@ void ListenerImpl::buildFilterChains() {
 
 void ListenerImpl::buildSocketOptions() {
   // TCP specific setup.
-  if (config_.has_connection_balance_config()) {
-    // Currently exact balance is the only supported type and there are no options.
-    ASSERT(config_.connection_balance_config().has_exact_balance());
-    connection_balancer_ = std::make_unique<Network::ExactConnectionBalancerImpl>();
-  } else {
-    connection_balancer_ = std::make_unique<Network::NopConnectionBalancerImpl>();
+  if (connection_balancer_ == nullptr) {
+    // Not in place listener update.
+    if (config_.has_connection_balance_config()) {
+      // Currently exact balance is the only supported type and there are no options.
+      ASSERT(config_.connection_balance_config().has_exact_balance());
+      connection_balancer_ = std::make_shared<Network::ExactConnectionBalancerImpl>();
+    } else {
+      connection_balancer_ = std::make_shared<Network::NopConnectionBalancerImpl>();
+    }
   }
 
   if (config_.has_tcp_fast_open_queue_length()) {

source/server/listener_impl.h

@@ -393,7 +393,7 @@ class ListenerImpl final : public Network::ListenerConfig,
   const bool continue_on_listener_filters_timeout_;
   Network::ActiveUdpListenerFactoryPtr udp_listener_factory_;
   Network::UdpPacketWriterFactoryPtr udp_writer_factory_;
-  Network::ConnectionBalancerPtr connection_balancer_;
+  Network::ConnectionBalancerSharedPtr connection_balancer_;
   std::shared_ptr<PerListenerFactoryContextImpl> listener_factory_context_;
   FilterChainManagerImpl filter_chain_manager_;
 

test/integration/xds_integration_test.cc

@@ -290,7 +290,13 @@ class LdsInplaceUpdateHttpIntegrationTest
     std::string tls_inspector_config = ConfigHelper::tlsInspectorFilter();
     config_helper_.addListenerFilter(tls_inspector_config);
     config_helper_.addSslConfig();
-    config_helper_.addConfigModifier([](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+    config_helper_.addConfigModifier([this](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+      if (!use_default_balancer_) {
+        bootstrap.mutable_static_resources()
+            ->mutable_listeners(0)
+            ->mutable_connection_balance_config()
+            ->mutable_exact_balance();
+      }
       auto* filter_chain_0 =
           bootstrap.mutable_static_resources()->mutable_listeners(0)->mutable_filter_chains(0);
       *filter_chain_0->mutable_filter_chain_match()->mutable_application_protocols()->Add() =
@@ -356,11 +362,18 @@ class LdsInplaceUpdateHttpIntegrationTest
     }
   }
 
+  void expectConnenctionServed(std::string alpn = "alpn0") {
+    auto codec_client_after_config_update = createHttpCodec(alpn);
+    expectResponseHeaderConnectionClose(*codec_client_after_config_update, false);
+    codec_client_after_config_update->close();
+  }
+
   std::unique_ptr<Ssl::ContextManager> context_manager_;
   Network::TransportSocketFactoryPtr context_;
   testing::NiceMock<Secret::MockSecretManager> secret_manager_;
   Network::Address::InstanceConstSharedPtr address_;
-};
+  bool use_default_balancer_{false};
+}; // namespace
 
 // Verify that http response on filter chain 0 has "Connection: close" header when filter chain 0
 // is deleted during the listener update.
@@ -388,6 +401,7 @@ TEST_P(LdsInplaceUpdateHttpIntegrationTest, ReloadConfigDeletingFilterChain) {
   expectResponseHeaderConnectionClose(*codec_client_1, true);
   test_server_->waitForGaugeGe("listener_manager.total_filter_chains_draining", 0);
   expectResponseHeaderConnectionClose(*codec_client_0, false);
+  expectConnenctionServed();
 }
 
 // Verify that http clients of filter chain 0 survives if new listener config adds new filter
@@ -416,6 +430,29 @@ TEST_P(LdsInplaceUpdateHttpIntegrationTest, ReloadConfigAddingFilterChain) {
   Cleanup cleanup2([c2 = codec_client_2.get()]() { c2->close(); });
   expectResponseHeaderConnectionClose(*codec_client_2, false);
   expectResponseHeaderConnectionClose(*codec_client_0, false);
+  expectConnenctionServed();
+}
+
+// Verify that balancer is inherited. Test only default balancer because ExactConnectionBalancer
+// is verified in filter chain add and delete test case.
+TEST_P(LdsInplaceUpdateHttpIntegrationTest, OverlappingFilterChainServesNewConnection) {
+  use_default_balancer_ = true;
+  initialize();
+
+  auto codec_client_0 = createHttpCodec("alpn0");
+  Cleanup cleanup([c0 = codec_client_0.get()]() { c0->close(); });
+  ConfigHelper new_config_helper(version_, *api_,
+                                 MessageUtil::getJsonStringFromMessage(config_helper_.bootstrap()));
+  new_config_helper.addConfigModifier(
+      [&](envoy::config::bootstrap::v3::Bootstrap& bootstrap) -> void {
+        auto* listener = bootstrap.mutable_static_resources()->mutable_listeners(0);
+        listener->mutable_filter_chains()->RemoveLast();
+      });
+
+  new_config_helper.setLds("1");
+  test_server_->waitForCounterGe("listener_manager.listener_in_place_updated", 1);
+  expectResponseHeaderConnectionClose(*codec_client_0, false);
+  expectConnenctionServed();
 }
 
 INSTANTIATE_TEST_SUITE_P(IpVersions, LdsInplaceUpdateHttpIntegrationTest,

test/server/connection_handler_test.cc

@@ -60,14 +60,18 @@ class ConnectionHandlerTest : public testing::Test, protected Logger::Loggable<L
         Network::ListenSocketFactorySharedPtr socket_factory,
         std::shared_ptr<AccessLog::MockInstance> access_log,
         std::shared_ptr<NiceMock<Network::MockFilterChainManager>> filter_chain_manager = nullptr,
-        uint32_t tcp_backlog_size = ENVOY_TCP_BACKLOG_SIZE)
+        uint32_t tcp_backlog_size = ENVOY_TCP_BACKLOG_SIZE,
+        Network::ConnectionBalancerSharedPtr connection_balancer = nullptr)
         : parent_(parent), socket_(std::make_shared<NiceMock<Network::MockListenSocket>>()),
           socket_factory_(std::move(socket_factory)), tag_(tag), bind_to_port_(bind_to_port),
           tcp_backlog_size_(tcp_backlog_size),
           hand_off_restored_destination_connections_(hand_off_restored_destination_connections),
           name_(name), listener_filters_timeout_(listener_filters_timeout),
           continue_on_listener_filters_timeout_(continue_on_listener_filters_timeout),
-          connection_balancer_(std::make_unique<Network::NopConnectionBalancerImpl>()),
+
+          connection_balancer_(connection_balancer == nullptr
+                                   ? std::make_shared<Network::NopConnectionBalancerImpl>()
+                                   : connection_balancer),
           access_logs_({access_log}), inline_filter_chain_manager_(filter_chain_manager),
           init_manager_(nullptr) {
       envoy::config::listener::v3::UdpListenerConfig dummy;
@@ -135,7 +139,7 @@ class ConnectionHandlerTest : public testing::Test, protected Logger::Loggable<L
     const bool continue_on_listener_filters_timeout_;
     std::unique_ptr<Network::ActiveUdpListenerFactory> udp_listener_factory_;
     std::unique_ptr<Network::UdpPacketWriterFactory> udp_writer_factory_;
-    Network::ConnectionBalancerPtr connection_balancer_;
+    Network::ConnectionBalancerSharedPtr connection_balancer_;
     BasicResourceLimitImpl open_connections_;
     const std::vector<AccessLog::InstanceSharedPtr> access_logs_;
     std::shared_ptr<NiceMock<Network::MockFilterChainManager>> inline_filter_chain_manager_;
@@ -182,7 +186,7 @@ class ConnectionHandlerTest : public testing::Test, protected Logger::Loggable<L
       uint64_t tag, bool bind_to_port, bool hand_off_restored_destination_connections,
       const std::string& name, Network::Listener* listener,
       Network::TcpListenerCallbacks** listener_callbacks = nullptr,
-      Network::MockConnectionBalancer* connection_balancer = nullptr,
+      std::shared_ptr<Network::MockConnectionBalancer> connection_balancer = nullptr,
       Network::BalancedConnectionHandler** balanced_connection_handler = nullptr,
       Network::Socket::Type socket_type = Network::Socket::Type::Stream,
       std::chrono::milliseconds listener_filters_timeout = std::chrono::milliseconds(15000),
@@ -193,8 +197,9 @@ class ConnectionHandlerTest : public testing::Test, protected Logger::Loggable<L
     listeners_.emplace_back(std::make_unique<TestListener>(
         *this, tag, bind_to_port, hand_off_restored_destination_connections, name, socket_type,
         listener_filters_timeout, continue_on_listener_filters_timeout, socket_factory_,
-        access_log_, overridden_filter_chain_manager, tcp_backlog_size));
+        access_log_, overridden_filter_chain_manager, tcp_backlog_size, connection_balancer));
     EXPECT_CALL(*socket_factory_, socketType()).WillOnce(Return(socket_type));
+
     if (listener == nullptr) {
       // Expecting listener config in place update.
       // If so, dispatcher would not create new network listener.
@@ -219,12 +224,11 @@ class ConnectionHandlerTest : public testing::Test, protected Logger::Loggable<L
           }));
     }
 
-    if (connection_balancer != nullptr) {
-      listeners_.back()->connection_balancer_.reset(connection_balancer);
-      ASSERT(balanced_connection_handler != nullptr);
+    if (balanced_connection_handler != nullptr) {
       EXPECT_CALL(*connection_balancer, registerHandler(_))
           .WillOnce(SaveArgAddress(balanced_connection_handler));
     }
+
     return listeners_.back().get();
   }
 
@@ -257,7 +261,7 @@ TEST_F(ConnectionHandlerTest, RemoveListenerDuringRebalance) {
 
   Network::TcpListenerCallbacks* listener_callbacks;
   auto listener = new NiceMock<Network::MockListener>();
-  Network::MockConnectionBalancer* connection_balancer = new Network::MockConnectionBalancer();
+  auto connection_balancer = std::make_shared<Network::MockConnectionBalancer>();
   Network::BalancedConnectionHandler* current_handler;
   TestListener* test_listener =
       addListener(1, true, false, "test_listener", listener, &listener_callbacks,
@@ -988,9 +992,14 @@ TEST_F(ConnectionHandlerTest, TcpListenerInplaceUpdate) {
   uint64_t old_listener_tag = 1;
   uint64_t new_listener_tag = 2;
   Network::TcpListenerCallbacks* old_listener_callbacks;
+  Network::BalancedConnectionHandler* current_handler;
+
   auto old_listener = new NiceMock<Network::MockListener>();
-  TestListener* old_test_listener = addListener(old_listener_tag, true, false, "test_listener",
-                                                old_listener, &old_listener_callbacks);
+  auto mock_connection_balancer = std::make_shared<Network::MockConnectionBalancer>();
+
+  TestListener* old_test_listener =
+      addListener(old_listener_tag, true, false, "test_listener", old_listener,
+                  &old_listener_callbacks, mock_connection_balancer, &current_handler);
   EXPECT_CALL(*socket_factory_, localAddress()).WillOnce(ReturnRef(local_address_));
   handler_->addListener(absl::nullopt, *old_test_listener);
   ASSERT_NE(old_test_listener, nullptr);
@@ -999,19 +1008,25 @@ TEST_F(ConnectionHandlerTest, TcpListenerInplaceUpdate) {
 
   auto overridden_filter_chain_manager =
       std::make_shared<NiceMock<Network::MockFilterChainManager>>();
-  TestListener* new_test_listener =
-      addListener(new_listener_tag, true, false, "test_listener", /* Network::Listener */ nullptr,
-                  &new_listener_callbacks, nullptr, nullptr, Network::Socket::Type::Stream,
-                  std::chrono::milliseconds(15000), false, overridden_filter_chain_manager);
+  TestListener* new_test_listener = addListener(
+      new_listener_tag, true, false, "test_listener", /* Network::Listener */ nullptr,
+      &new_listener_callbacks, mock_connection_balancer, nullptr, Network::Socket::Type::Stream,
+      std::chrono::milliseconds(15000), false, overridden_filter_chain_manager);
   handler_->addListener(old_listener_tag, *new_test_listener);
   ASSERT_EQ(new_listener_callbacks, nullptr)
       << "new listener should be inplace added and callback should not change";
 
   Network::MockConnectionSocket* connection = new NiceMock<Network::MockConnectionSocket>();
+  current_handler->incNumConnections();
+
+  EXPECT_CALL(*mock_connection_balancer, pickTargetHandler(_))
+      .WillOnce(ReturnRef(*current_handler));
   EXPECT_CALL(manager_, findFilterChain(_)).Times(0);
   EXPECT_CALL(*overridden_filter_chain_manager, findFilterChain(_)).WillOnce(Return(nullptr));
   EXPECT_CALL(*access_log_, log(_, _, _, _)).Times(1);
+  EXPECT_CALL(*mock_connection_balancer, unregisterHandler(_));
   old_listener_callbacks->onAccept(Network::ConnectionSocketPtr{connection});
+  EXPECT_EQ(0UL, handler_->numConnections());
   EXPECT_CALL(*old_listener, onDestroy());
 }