From dfc552724f1883a2c56c3269d46969e78bbcded2 Mon Sep 17 00:00:00 2001 From: He Jie Xu Date: Thu, 3 Feb 2022 17:39:00 +0000 Subject: [PATCH 1/6] listener: check the listener isn't stopped when balance the request Signed-off-by: He Jie Xu --- source/server/connection_handler_impl.cc | 6 +- test/server/connection_handler_test.cc | 149 ++++++++++++++++++++++- 2 files changed, 150 insertions(+), 5 deletions(-) diff --git a/source/server/connection_handler_impl.cc b/source/server/connection_handler_impl.cc index 89533d944818c..0da427d917034 100644 --- a/source/server/connection_handler_impl.cc +++ b/source/server/connection_handler_impl.cc @@ -243,7 +243,8 @@ ConnectionHandlerImpl::getBalancedHandlerByAddress(const Network::Address::Insta // We do not return stopped listeners. // If there is exact address match, return the corresponding listener. if (auto listener_it = tcp_listener_map_by_address_.find(address.asStringView()); - listener_it != tcp_listener_map_by_address_.end()) { + listener_it != tcp_listener_map_by_address_.end() && + listener_it->second->listener_->listener() != nullptr) { return Network::BalancedConnectionHandlerOptRef( listener_it->second->tcpListener().value().get()); } @@ -260,7 +261,8 @@ ConnectionHandlerImpl::getBalancedHandlerByAddress(const Network::Address::Insta : Network::Utility::getIpv6AnyAddress(address.ip()->port())->asString(); auto iter = tcp_listener_map_by_address_.find(addr_str); - if (iter != tcp_listener_map_by_address_.end()) { + if (iter != tcp_listener_map_by_address_.end() && + iter->second->listener_->listener() != nullptr) { details = *iter->second; } } else { diff --git a/test/server/connection_handler_test.cc b/test/server/connection_handler_test.cc index 2ad3a517cd469..626d3f0ddbb58 100644 --- a/test/server/connection_handler_test.cc +++ b/test/server/connection_handler_test.cc @@ -779,8 +779,13 @@ TEST_F(ConnectionHandlerTest, MatchLatestListener) { handler_->addListener(absl::nullopt, *test_listener1); // Listener2 will be replaced by Listener3. + auto listener2_overridden_filter_chain_manager = + std::make_shared>(); auto listener2 = new NiceMock(); - TestListener* test_listener2 = addListener(2, false, false, "test_listener2", listener2); + TestListener* test_listener2 = + addListener(2, false, false, "test_listener2", listener2, nullptr, nullptr, nullptr, + Network::Socket::Type::Stream, std::chrono::milliseconds(15000), false, + listener2_overridden_filter_chain_manager); Network::Address::InstanceConstSharedPtr listener2_address( new Network::Address::Ipv4Instance("127.0.0.1", 10002)); EXPECT_CALL(test_listener2->socket_factory_, localAddress()) @@ -788,8 +793,13 @@ TEST_F(ConnectionHandlerTest, MatchLatestListener) { handler_->addListener(absl::nullopt, *test_listener2); // Listener3 will replace the listener2. + auto listener3_overridden_filter_chain_manager = + std::make_shared>(); auto listener3 = new NiceMock(); - TestListener* test_listener3 = addListener(3, false, false, "test_listener3", listener3); + TestListener* test_listener3 = + addListener(3, false, false, "test_listener3", listener3, nullptr, nullptr, nullptr, + Network::Socket::Type::Stream, std::chrono::milliseconds(15000), false, + listener3_overridden_filter_chain_manager); Network::Address::InstanceConstSharedPtr listener3_address( new Network::Address::Ipv4Instance("127.0.0.1", 10002)); EXPECT_CALL(test_listener3->socket_factory_, localAddress()) @@ -823,7 +833,12 @@ TEST_F(ConnectionHandlerTest, MatchLatestListener) { cb.socket().connectionInfoProvider().restoreLocalAddress(alt_address); return Network::FilterStatus::Continue; })); - EXPECT_CALL(manager_, findFilterChain(_)).WillOnce(Return(filter_chain_.get())); + // Ensure the request is going to the listener3. + EXPECT_CALL(*listener3_overridden_filter_chain_manager, findFilterChain(_)) + .WillOnce(Return(filter_chain_.get())); + // Ensure the request isn't going to the listener1 and listener2. + EXPECT_CALL(manager_, findFilterChain(_)).Times(0); + EXPECT_CALL(*listener2_overridden_filter_chain_manager, findFilterChain(_)).Times(0); auto* connection = new NiceMock(); EXPECT_CALL(dispatcher_, createServerConnection_()).WillOnce(Return(connection)); @@ -836,6 +851,134 @@ TEST_F(ConnectionHandlerTest, MatchLatestListener) { EXPECT_CALL(*access_log_, log(_, _, _, _)); } +TEST_F(ConnectionHandlerTest, EnsureNotMatchStoppedListener) { + Network::TcpListenerCallbacks* listener_callbacks; + // The Listener1 will accept the new connection first then balanced to other listener. + auto listener1 = new NiceMock(); + TestListener* test_listener1 = + addListener(1, true, true, "test_listener1", listener1, &listener_callbacks); + EXPECT_CALL(test_listener1->socket_factory_, localAddress()) + .WillRepeatedly(ReturnRef(local_address_)); + handler_->addListener(absl::nullopt, *test_listener1); + + auto listener2_overridden_filter_chain_manager = + std::make_shared>(); + auto listener2 = new NiceMock(); + TestListener* test_listener2 = + addListener(2, false, false, "test_listener2", listener2, nullptr, nullptr, nullptr, + Network::Socket::Type::Stream, std::chrono::milliseconds(15000), false, + listener2_overridden_filter_chain_manager); + Network::Address::InstanceConstSharedPtr listener2_address( + new Network::Address::Ipv4Instance("127.0.0.1", 10002)); + EXPECT_CALL(test_listener2->socket_factory_, localAddress()) + .WillRepeatedly(ReturnRef(listener2_address)); + handler_->addListener(absl::nullopt, *test_listener2); + + // Stop the listener2. + EXPECT_CALL(*listener2, onDestroy()); + handler_->stopListeners(2); + + Network::MockListenerFilter* test_filter = new Network::MockListenerFilter(); + EXPECT_CALL(*test_filter, destroy_()); + Network::MockConnectionSocket* accepted_socket = new NiceMock(); + bool redirected = false; + EXPECT_CALL(factory_, createListenerFilterChain(_)) + .WillRepeatedly(Invoke([&](Network::ListenerFilterManager& manager) -> bool { + // Insert the Mock filter. + if (!redirected) { + manager.addAcceptFilter(listener_filter_matcher_, + Network::ListenerFilterPtr{test_filter}); + redirected = true; + } + return true; + })); + // This is the address of listener2. + Network::Address::InstanceConstSharedPtr alt_address( + new Network::Address::Ipv4Instance("127.0.0.1", 10002, nullptr)); + EXPECT_CALL(*test_filter, onAccept(_)) + .WillOnce(Invoke([&](Network::ListenerFilterCallbacks& cb) -> Network::FilterStatus { + cb.socket().connectionInfoProvider().restoreLocalAddress(alt_address); + return Network::FilterStatus::Continue; + })); + // Ensure the request isn't going to the listener2. + EXPECT_CALL(*listener2_overridden_filter_chain_manager, findFilterChain(_)).Times(0); + // Ensure the request is going to the listener1. + EXPECT_CALL(manager_, findFilterChain(_)).WillOnce(Return(filter_chain_.get())); + + auto* connection = new NiceMock(); + EXPECT_CALL(dispatcher_, createServerConnection_()).WillOnce(Return(connection)); + EXPECT_CALL(factory_, createNetworkFilterChain(_, _)).WillOnce(Return(true)); + listener_callbacks->onAccept(Network::ConnectionSocketPtr{accepted_socket}); + EXPECT_EQ(1UL, handler_->numConnections()); + + EXPECT_CALL(*listener1, onDestroy()); + EXPECT_CALL(*access_log_, log(_, _, _, _)); +} + +TEST_F(ConnectionHandlerTest, EnsureNotMatchStoppedAnyAddressListener) { + Network::TcpListenerCallbacks* listener_callbacks; + // The Listener1 will accept the new connection first then balanced to other listener. + auto listener1 = new NiceMock(); + TestListener* test_listener1 = + addListener(1, true, true, "test_listener1", listener1, &listener_callbacks); + EXPECT_CALL(test_listener1->socket_factory_, localAddress()) + .WillRepeatedly(ReturnRef(local_address_)); + handler_->addListener(absl::nullopt, *test_listener1); + + auto listener2_overridden_filter_chain_manager = + std::make_shared>(); + auto listener2 = new NiceMock(); + TestListener* test_listener2 = + addListener(2, false, false, "test_listener2", listener2, nullptr, nullptr, nullptr, + Network::Socket::Type::Stream, std::chrono::milliseconds(15000), false, + listener2_overridden_filter_chain_manager); + Network::Address::InstanceConstSharedPtr listener2_address( + new Network::Address::Ipv4Instance("0.0.0.0", 10002)); + EXPECT_CALL(test_listener2->socket_factory_, localAddress()) + .WillRepeatedly(ReturnRef(listener2_address)); + handler_->addListener(absl::nullopt, *test_listener2); + + // Stop the listener2. + EXPECT_CALL(*listener2, onDestroy()); + handler_->stopListeners(2); + + Network::MockListenerFilter* test_filter = new Network::MockListenerFilter(); + EXPECT_CALL(*test_filter, destroy_()); + Network::MockConnectionSocket* accepted_socket = new NiceMock(); + bool redirected = false; + EXPECT_CALL(factory_, createListenerFilterChain(_)) + .WillRepeatedly(Invoke([&](Network::ListenerFilterManager& manager) -> bool { + // Insert the Mock filter. + if (!redirected) { + manager.addAcceptFilter(listener_filter_matcher_, + Network::ListenerFilterPtr{test_filter}); + redirected = true; + } + return true; + })); + // This is the address of listener2. + Network::Address::InstanceConstSharedPtr alt_address( + new Network::Address::Ipv4Instance("127.0.0.1", 10002, nullptr)); + EXPECT_CALL(*test_filter, onAccept(_)) + .WillOnce(Invoke([&](Network::ListenerFilterCallbacks& cb) -> Network::FilterStatus { + cb.socket().connectionInfoProvider().restoreLocalAddress(alt_address); + return Network::FilterStatus::Continue; + })); + // Ensure the request isn't going to the listener2. + EXPECT_CALL(*listener2_overridden_filter_chain_manager, findFilterChain(_)).Times(0); + // Ensure the request is going to the listener1. + EXPECT_CALL(manager_, findFilterChain(_)).WillOnce(Return(filter_chain_.get())); + + auto* connection = new NiceMock(); + EXPECT_CALL(dispatcher_, createServerConnection_()).WillOnce(Return(connection)); + EXPECT_CALL(factory_, createNetworkFilterChain(_, _)).WillOnce(Return(true)); + listener_callbacks->onAccept(Network::ConnectionSocketPtr{accepted_socket}); + EXPECT_EQ(1UL, handler_->numConnections()); + + EXPECT_CALL(*listener1, onDestroy()); + EXPECT_CALL(*access_log_, log(_, _, _, _)); +} + TEST_F(ConnectionHandlerTest, FallbackToWildcardListener) { Network::TcpListenerCallbacks* listener_callbacks1; auto listener1 = new NiceMock(); From 7dcd41c7a932cb1dec9e52baa4e8f403e28a6b33 Mon Sep 17 00:00:00 2001 From: He Jie Xu Date: Thu, 3 Feb 2022 18:42:05 +0000 Subject: [PATCH 2/6] fix comment Signed-off-by: He Jie Xu --- test/server/connection_handler_test.cc | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/test/server/connection_handler_test.cc b/test/server/connection_handler_test.cc index 626d3f0ddbb58..c0e16a1f1e40c 100644 --- a/test/server/connection_handler_test.cc +++ b/test/server/connection_handler_test.cc @@ -766,11 +766,11 @@ TEST_F(ConnectionHandlerTest, NormalRedirect) { // When update a listener, the old listener will be stopped and the new listener will // be added into ConnectionHandler before remove the old listener from ConnectionHandler. -// This test ensure ConnectionHandler can query the correct Listener when balanced the connection +// This test ensure ConnectionHandler can query the correct Listener when redirect the connection // through `getBalancedHandlerByAddress` TEST_F(ConnectionHandlerTest, MatchLatestListener) { Network::TcpListenerCallbacks* listener_callbacks; - // The Listener1 will accept the new connection first then balanced to other listener. + // The Listener1 will accept the new connection first then redirect to other listener. auto listener1 = new NiceMock(); TestListener* test_listener1 = addListener(1, true, true, "test_listener1", listener1, &listener_callbacks); @@ -853,7 +853,7 @@ TEST_F(ConnectionHandlerTest, MatchLatestListener) { TEST_F(ConnectionHandlerTest, EnsureNotMatchStoppedListener) { Network::TcpListenerCallbacks* listener_callbacks; - // The Listener1 will accept the new connection first then balanced to other listener. + // The Listener1 will accept the new connection first then redirect to other listener. auto listener1 = new NiceMock(); TestListener* test_listener1 = addListener(1, true, true, "test_listener1", listener1, &listener_callbacks); @@ -917,7 +917,7 @@ TEST_F(ConnectionHandlerTest, EnsureNotMatchStoppedListener) { TEST_F(ConnectionHandlerTest, EnsureNotMatchStoppedAnyAddressListener) { Network::TcpListenerCallbacks* listener_callbacks; - // The Listener1 will accept the new connection first then balanced to other listener. + // The Listener1 will accept the new connection first then redirect to other listener. auto listener1 = new NiceMock(); TestListener* test_listener1 = addListener(1, true, true, "test_listener1", listener1, &listener_callbacks); From bef9023f3413402e20b67335cea147612244e984 Mon Sep 17 00:00:00 2001 From: He Jie Xu Date: Thu, 3 Feb 2022 18:46:17 +0000 Subject: [PATCH 3/6] add test Signed-off-by: He Jie Xu --- test/server/connection_handler_test.cc | 69 ++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/test/server/connection_handler_test.cc b/test/server/connection_handler_test.cc index c0e16a1f1e40c..2f809e280c744 100644 --- a/test/server/connection_handler_test.cc +++ b/test/server/connection_handler_test.cc @@ -979,6 +979,75 @@ TEST_F(ConnectionHandlerTest, EnsureNotMatchStoppedAnyAddressListener) { EXPECT_CALL(*access_log_, log(_, _, _, _)); } +TEST_F(ConnectionHandlerTest, EnsureNotMatchStoppedAnyAddressListenerOnOldBehavior) { + auto scoped_runtime = std::make_unique(); + + Runtime::LoaderSingleton::getExisting()->mergeValues( + {{"envoy.reloadable_features.listener_wildcard_match_ip_family", "false"}}); + + Network::TcpListenerCallbacks* listener_callbacks; + // The Listener1 will accept the new connection first then redirect to other listener. + auto listener1 = new NiceMock(); + TestListener* test_listener1 = + addListener(1, true, true, "test_listener1", listener1, &listener_callbacks); + EXPECT_CALL(test_listener1->socket_factory_, localAddress()) + .WillRepeatedly(ReturnRef(local_address_)); + handler_->addListener(absl::nullopt, *test_listener1); + + auto listener2_overridden_filter_chain_manager = + std::make_shared>(); + auto listener2 = new NiceMock(); + TestListener* test_listener2 = + addListener(2, false, false, "test_listener2", listener2, nullptr, nullptr, nullptr, + Network::Socket::Type::Stream, std::chrono::milliseconds(15000), false, + listener2_overridden_filter_chain_manager); + Network::Address::InstanceConstSharedPtr listener2_address( + new Network::Address::Ipv4Instance("0.0.0.0", 10002)); + EXPECT_CALL(test_listener2->socket_factory_, localAddress()) + .WillRepeatedly(ReturnRef(listener2_address)); + handler_->addListener(absl::nullopt, *test_listener2); + + // Stop the listener2. + EXPECT_CALL(*listener2, onDestroy()); + handler_->stopListeners(2); + + Network::MockListenerFilter* test_filter = new Network::MockListenerFilter(); + EXPECT_CALL(*test_filter, destroy_()); + Network::MockConnectionSocket* accepted_socket = new NiceMock(); + bool redirected = false; + EXPECT_CALL(factory_, createListenerFilterChain(_)) + .WillRepeatedly(Invoke([&](Network::ListenerFilterManager& manager) -> bool { + // Insert the Mock filter. + if (!redirected) { + manager.addAcceptFilter(listener_filter_matcher_, + Network::ListenerFilterPtr{test_filter}); + redirected = true; + } + return true; + })); + // This is the address of listener2. + Network::Address::InstanceConstSharedPtr alt_address( + new Network::Address::Ipv4Instance("127.0.0.1", 10002, nullptr)); + EXPECT_CALL(*test_filter, onAccept(_)) + .WillOnce(Invoke([&](Network::ListenerFilterCallbacks& cb) -> Network::FilterStatus { + cb.socket().connectionInfoProvider().restoreLocalAddress(alt_address); + return Network::FilterStatus::Continue; + })); + // Ensure the request isn't going to the listener2. + EXPECT_CALL(*listener2_overridden_filter_chain_manager, findFilterChain(_)).Times(0); + // Ensure the request is going to the listener1. + EXPECT_CALL(manager_, findFilterChain(_)).WillOnce(Return(filter_chain_.get())); + + auto* connection = new NiceMock(); + EXPECT_CALL(dispatcher_, createServerConnection_()).WillOnce(Return(connection)); + EXPECT_CALL(factory_, createNetworkFilterChain(_, _)).WillOnce(Return(true)); + listener_callbacks->onAccept(Network::ConnectionSocketPtr{accepted_socket}); + EXPECT_EQ(1UL, handler_->numConnections()); + + EXPECT_CALL(*listener1, onDestroy()); + EXPECT_CALL(*access_log_, log(_, _, _, _)); +} + TEST_F(ConnectionHandlerTest, FallbackToWildcardListener) { Network::TcpListenerCallbacks* listener_callbacks1; auto listener1 = new NiceMock(); From 48b574d11c211cf074f79ff8e20a6eb35f84b403 Mon Sep 17 00:00:00 2001 From: He Jie Xu Date: Sat, 5 Feb 2022 08:31:47 +0000 Subject: [PATCH 4/6] more fix Signed-off-by: He Jie Xu --- source/server/connection_handler_impl.cc | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/source/server/connection_handler_impl.cc b/source/server/connection_handler_impl.cc index 0da427d917034..f22eabac25864 100644 --- a/source/server/connection_handler_impl.cc +++ b/source/server/connection_handler_impl.cc @@ -231,7 +231,7 @@ ConnectionHandlerImpl::getBalancedHandlerByTag(uint64_t listener_tag) { auto active_listener = findActiveListenerByTag(listener_tag); if (active_listener.has_value()) { ASSERT(absl::holds_alternative>( - active_listener->get().typed_listener_)); + active_listener->get().typed_listener_) && active_listener->get().listener_->listener() != nullptr); return Network::BalancedConnectionHandlerOptRef( active_listener->get().tcpListener().value().get()); } @@ -240,6 +240,9 @@ ConnectionHandlerImpl::getBalancedHandlerByTag(uint64_t listener_tag) { Network::BalancedConnectionHandlerOptRef ConnectionHandlerImpl::getBalancedHandlerByAddress(const Network::Address::Instance& address) { + if (address.type() != Network::Address::Type::Ip) { + return absl::nullopt; + } // We do not return stopped listeners. // If there is exact address match, return the corresponding listener. if (auto listener_it = tcp_listener_map_by_address_.find(address.asStringView()); From 72d978ab34cf229e091db33032d45e22e598ef90 Mon Sep 17 00:00:00 2001 From: He Jie Xu Date: Mon, 7 Feb 2022 11:55:20 +0000 Subject: [PATCH 5/6] Assert the ip address Signed-off-by: He Jie Xu --- source/common/network/utility.cc | 16 +++++++++++----- source/common/network/utility.h | 6 ++---- source/server/connection_handler_impl.cc | 17 +++++++++-------- 3 files changed, 22 insertions(+), 17 deletions(-) diff --git a/source/common/network/utility.cc b/source/common/network/utility.cc index b168252d17591..e8f0fd50ef0de 100644 --- a/source/common/network/utility.cc +++ b/source/common/network/utility.cc @@ -346,12 +346,14 @@ Address::InstanceConstSharedPtr Utility::getIpv6LoopbackAddress() { new Address::Ipv6Instance("::1", 0, nullptr)); } -Address::InstanceConstSharedPtr Utility::getIpv4AnyAddress(uint32_t port) { - CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr, new Address::Ipv4Instance(port)); +Address::InstanceConstSharedPtr Utility::getIpv4AnyAddress() { + CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr, + new Address::Ipv4Instance(static_cast(0))); } -Address::InstanceConstSharedPtr Utility::getIpv6AnyAddress(uint32_t port) { - CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr, new Address::Ipv6Instance(port)); +Address::InstanceConstSharedPtr Utility::getIpv6AnyAddress() { + CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr, + new Address::Ipv6Instance(static_cast(0))); } const std::string& Utility::getIpv4CidrCatchAllAddress() { @@ -713,6 +715,8 @@ Api::IoErrorPtr Utility::readPacketsFromSocket(IoHandle& handle, : num_packets_to_read); // Make sure to read at least once. num_reads = std::max(1, num_reads); + bool honor_read_limit = + Runtime::runtimeFeatureEnabled("envoy.reloadable_features.udp_per_event_loop_read_limit"); do { const uint32_t old_packets_dropped = packets_dropped; const MonotonicTime receive_time = time_source.monotonicTime(); @@ -740,7 +744,9 @@ Api::IoErrorPtr Utility::readPacketsFromSocket(IoHandle& handle, delta); udp_packet_processor.onDatagramsDropped(delta); } - --num_reads; + if (honor_read_limit) { + --num_reads; + } if (num_reads == 0) { return std::move(result.err_); } diff --git a/source/common/network/utility.h b/source/common/network/utility.h index 721140b530def..d042d130071f6 100644 --- a/source/common/network/utility.h +++ b/source/common/network/utility.h @@ -246,20 +246,18 @@ class Utility { static Address::InstanceConstSharedPtr getIpv6LoopbackAddress(); /** - * @param port to be included in address, the default is 0. * @return Address::InstanceConstSharedPtr an address that represents the IPv4 wildcard address * (i.e. "0.0.0.0"). Used during binding to indicate that incoming connections to any * local IPv4 address are to be accepted. */ - static Address::InstanceConstSharedPtr getIpv4AnyAddress(uint32_t port = 0); + static Address::InstanceConstSharedPtr getIpv4AnyAddress(); /** - * @param port to be included in address, the default is 0. * @return Address::InstanceConstSharedPtr an address that represents the IPv6 wildcard address * (i.e. "::"). Used during binding to indicate that incoming connections to any local * IPv6 address are to be accepted. */ - static Address::InstanceConstSharedPtr getIpv6AnyAddress(uint32_t port = 0); + static Address::InstanceConstSharedPtr getIpv6AnyAddress(); /** * @return the IPv4 CIDR catch-all address (0.0.0.0/0). diff --git a/source/server/connection_handler_impl.cc b/source/server/connection_handler_impl.cc index f22eabac25864..920c083a89c9b 100644 --- a/source/server/connection_handler_impl.cc +++ b/source/server/connection_handler_impl.cc @@ -7,6 +7,7 @@ #include "source/common/common/logger.h" #include "source/common/event/deferred_task.h" +#include "source/common/network/address_impl.h" #include "source/common/network/utility.h" #include "source/common/runtime/runtime_features.h" #include "source/server/active_internal_listener.h" @@ -231,7 +232,8 @@ ConnectionHandlerImpl::getBalancedHandlerByTag(uint64_t listener_tag) { auto active_listener = findActiveListenerByTag(listener_tag); if (active_listener.has_value()) { ASSERT(absl::holds_alternative>( - active_listener->get().typed_listener_) && active_listener->get().listener_->listener() != nullptr); + active_listener->get().typed_listener_) && + active_listener->get().listener_->listener() != nullptr); return Network::BalancedConnectionHandlerOptRef( active_listener->get().tcpListener().value().get()); } @@ -240,9 +242,9 @@ ConnectionHandlerImpl::getBalancedHandlerByTag(uint64_t listener_tag) { Network::BalancedConnectionHandlerOptRef ConnectionHandlerImpl::getBalancedHandlerByAddress(const Network::Address::Instance& address) { - if (address.type() != Network::Address::Type::Ip) { - return absl::nullopt; - } + // Only Ip address can be restored to original address and redirect. + ASSERT(address.type() == Network::Address::Type::Ip); + // We do not return stopped listeners. // If there is exact address match, return the corresponding listener. if (auto listener_it = tcp_listener_map_by_address_.find(address.asStringView()); @@ -258,10 +260,9 @@ ConnectionHandlerImpl::getBalancedHandlerByAddress(const Network::Address::Insta // TODO(wattli): consolidate with previous search for more efficiency. if (Runtime::runtimeFeatureEnabled( "envoy.reloadable_features.listener_wildcard_match_ip_family")) { - std::string addr_str = - address.ip()->version() == Network::Address::IpVersion::v4 - ? Network::Utility::getIpv4AnyAddress(address.ip()->port())->asString() - : Network::Utility::getIpv6AnyAddress(address.ip()->port())->asString(); + std::string addr_str = address.ip()->version() == Network::Address::IpVersion::v4 + ? Network::Address::Ipv4Instance(address.ip()->port()).asString() + : Network::Address::Ipv6Instance(address.ip()->port()).asString(); auto iter = tcp_listener_map_by_address_.find(addr_str); if (iter != tcp_listener_map_by_address_.end() && From d32a126d22d0c1148d77d7da7f59a0e2b96391c8 Mon Sep 17 00:00:00 2001 From: He Jie Xu Date: Mon, 7 Feb 2022 14:31:59 +0000 Subject: [PATCH 6/6] revert unrelated change Signed-off-by: He Jie Xu --- source/common/network/utility.cc | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/source/common/network/utility.cc b/source/common/network/utility.cc index e8f0fd50ef0de..e9be775aca015 100644 --- a/source/common/network/utility.cc +++ b/source/common/network/utility.cc @@ -715,8 +715,6 @@ Api::IoErrorPtr Utility::readPacketsFromSocket(IoHandle& handle, : num_packets_to_read); // Make sure to read at least once. num_reads = std::max(1, num_reads); - bool honor_read_limit = - Runtime::runtimeFeatureEnabled("envoy.reloadable_features.udp_per_event_loop_read_limit"); do { const uint32_t old_packets_dropped = packets_dropped; const MonotonicTime receive_time = time_source.monotonicTime(); @@ -744,9 +742,7 @@ Api::IoErrorPtr Utility::readPacketsFromSocket(IoHandle& handle, delta); udp_packet_processor.onDatagramsDropped(delta); } - if (honor_read_limit) { - --num_reads; - } + --num_reads; if (num_reads == 0) { return std::move(result.err_); }