listener: support multiple addresses

api/envoy/admin/v2alpha/listeners.proto

@@ -26,7 +26,7 @@ message ListenerStatus {
   // Name of the listener
   string name = 1;
 
-  // The actual local address that the listener is listening on. If a listener was configured
+  // The actual local addresses that the listener is listening on. If a listener was configured
   // to listen on port 0, then this address has the port that was allocated by the OS.
   api.v2.core.Address local_address = 2;
 }

api/envoy/admin/v3/listeners.proto

@@ -31,7 +31,7 @@ message ListenerStatus {
   // Name of the listener
   string name = 1;
 
-  // The actual local address that the listener is listening on. If a listener was configured
+  // The actual local addresses that the listener is listening on. If a listener was configured
   // to listen on port 0, then this address has the port that was allocated by the OS.
-  config.core.v3.Address local_address = 2;
+  repeated config.core.v3.Address local_addresses = 2;
 }

api/envoy/config/listener/v3/listener.proto

@@ -38,7 +38,7 @@ message ListenerCollection {
   repeated xds.core.v3.CollectionEntry entries = 1;
 }
 
-// [#next-free-field: 33]
+// [#next-free-field: 34]
 message Listener {
   option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.Listener";
 
@@ -107,8 +107,15 @@ message Listener {
   // that is governed by the bind rules of the OS. E.g., multiple listeners can listen on port 0 on
   // Linux as the actual port will be allocated by the OS.
   // Required unless *api_listener* or *listener_specifier* is populated.
+  // TODO (soulxu): deprecated the `address` field after `addresses` implemented.
   core.v3.Address address = 2;
 
+  // A list of addresses that the listener should listen on. In general, the address must be unique, though
+  // that is governed by the bind rules of the OS. E.g., multiple listeners can listen on port 0 on
+  // Linux as the actual port will be allocated by the OS. For multiple addresses in single listener,
+  // all addresses are the same protocol, and multiple internal addresses don't support now.
+  repeated core.v3.Address addresses = 33;
+
   // Optional prefix to use on listener stats. If empty, the stats will be rooted at
   // `listener.<address as string>.`. If non-empty, stats will be rooted at
   // `listener.<stat_prefix>.`.

envoy/network/connection_handler.h

@@ -13,6 +13,8 @@
 
 #include "source/common/common/interval_value.h"
 
+#include "address.h"
+
 namespace Envoy {
 namespace Network {
 
@@ -178,10 +180,12 @@ class TcpConnectionHandler : public virtual ConnectionHandler {
   /**
    * Obtain the rebalancer of the tcp listener.
    * @param listener_tag supplies the tag of the tcp listener that was passed to addListener().
+   * @param address is used to query the address specific handler.
    * @return BalancedConnectionHandlerOptRef the balancer attached to the listener. `nullopt` if
    * listener doesn't exist or rebalancer doesn't exist.
    */
-  virtual BalancedConnectionHandlerOptRef getBalancedHandlerByTag(uint64_t listener_tag) PURE;
+  virtual BalancedConnectionHandlerOptRef
+  getBalancedHandlerByTag(uint64_t listener_tag, const Network::Address::Instance& address) PURE;
 
   /**
    * Obtain the rebalancer of the tcp listener.
@@ -199,11 +203,12 @@ class TcpConnectionHandler : public virtual ConnectionHandler {
 class UdpConnectionHandler : public virtual ConnectionHandler {
 public:
   /**
-   * Get the ``UdpListenerCallbacks`` associated with ``listener_tag``. This will be
+   * Get the ``UdpListenerCallbacks`` associated with ``listener_tag`` and ``address``. This will be
    * absl::nullopt for non-UDP listeners and for ``listener_tag`` values that have already been
    * removed.
    */
-  virtual UdpListenerCallbacksOptRef getUdpListenerCallbacks(uint64_t listener_tag) PURE;
+  virtual UdpListenerCallbacksOptRef
+  getUdpListenerCallbacks(uint64_t listener_tag, const Network::Address::Instance& address) PURE;
 };
 
 /**
@@ -219,15 +224,17 @@ class ActiveUdpListenerFactory {
    * @param runtime the runtime for this server.
    * @param worker_index The index of the worker this listener is being created on.
    * @param parent is the owner of the created ActiveListener objects.
+   * @param listen_socket_ptr is the UDP socket.
    * @param dispatcher is used to create actual UDP listener.
    * @param config provides information needed to create ActiveUdpListener and
    * UdpListener objects.
    * @return the ActiveUdpListener created.
    */
   virtual ConnectionHandler::ActiveUdpListenerPtr
   createActiveUdpListener(Runtime::Loader& runtime, uint32_t worker_index,
-                          UdpConnectionHandler& parent, Event::Dispatcher& dispatcher,
-                          Network::ListenerConfig& config) PURE;
+                          UdpConnectionHandler& parent,
+                          Network::SocketSharedPtr&& listen_socket_ptr,
+                          Event::Dispatcher& dispatcher, Network::ListenerConfig& config) PURE;
 
   /**
    * @return true if the UDP passing through listener doesn't form stateful connections.

envoy/network/listener.h

@@ -20,6 +20,8 @@
 
 #include "source/common/common/interval_value.h"
 
+#include "address.h"
+
 namespace Envoy {
 namespace Network {
 
@@ -95,9 +97,11 @@ class UdpListenerConfig {
   virtual UdpPacketWriterFactory& packetWriterFactory() PURE;
 
   /**
+   * @param address is used to query the address specific router.
    * @return the UdpListenerWorkerRouter for this listener.
    */
-  virtual UdpListenerWorkerRouter& listenerWorkerRouter() PURE;
+  virtual UdpListenerWorkerRouter&
+  listenerWorkerRouter(const Network::Address::Instance& address) PURE;
 
   /**
    * @return the configuration for the listener.
@@ -148,14 +152,14 @@ class ListenerConfig {
   /**
    * @return ListenSocketFactory& the factory to create listen socket.
    */
-  virtual ListenSocketFactory& listenSocketFactory() PURE;
+  virtual const std::vector<ListenSocketFactoryPtr>& listenSocketFactories() PURE;
 
   /**
    * @return bool specifies whether the listener should actually listen on the port.
    *         A listener that doesn't listen on a port can only receive connections
    *         redirected from other listeners.
    */
-  virtual bool bindToPort() PURE;
+  virtual bool bindToPort() const PURE;
 
   /**
    * @return bool if a connection should be handed off to another Listener after the original
@@ -215,10 +219,11 @@ class ListenerConfig {
   virtual envoy::config::core::v3::TrafficDirection direction() const PURE;
 
   /**
+   * @param address is used for query the address specific connection balancer.
    * @return the connection balancer for this listener. All listeners have a connection balancer,
    *         though the implementation may be a NOP balancer.
    */
-  virtual ConnectionBalancer& connectionBalancer() PURE;
+  virtual ConnectionBalancer& connectionBalancer(const Network::Address::Instance& address) PURE;
 
   /**
    * Open connection resources for this listener.

source/common/network/utility.h

@@ -418,5 +418,14 @@ class Utility {
   static absl::uint128 flipOrder(const absl::uint128& input);
 };
 
+/**
+ * Log formatter for a list of addresses.
+ */
+struct AddressStrFormatter {
+  void operator()(std::string* out, const Network::Address::InstanceConstSharedPtr& instance) {
+    out->append(instance->asString());
+  }
+};
+
 } // namespace Network
 } // namespace Envoy

source/common/quic/active_quic_listener.cc

@@ -26,21 +26,7 @@ bool ActiveQuicListenerFactory::disable_kernel_bpf_packet_routing_for_test_ = fa
 ActiveQuicListener::ActiveQuicListener(
     Runtime::Loader& runtime, uint32_t worker_index, uint32_t concurrency,
     Event::Dispatcher& dispatcher, Network::UdpConnectionHandler& parent,
-    Network::ListenerConfig& listener_config, const quic::QuicConfig& quic_config,
-    bool kernel_worker_routing, const envoy::config::core::v3::RuntimeFeatureFlag& enabled,
-    QuicStatNames& quic_stat_names, uint32_t packets_received_to_connection_count_ratio,
-    EnvoyQuicCryptoServerStreamFactoryInterface& crypto_server_stream_factory,
-    EnvoyQuicProofSourceFactoryInterface& proof_source_factory)
-    : ActiveQuicListener(runtime, worker_index, concurrency, dispatcher, parent,
-                         listener_config.listenSocketFactory().getListenSocket(worker_index),
-                         listener_config, quic_config, kernel_worker_routing, enabled,
-                         quic_stat_names, packets_received_to_connection_count_ratio,
-                         crypto_server_stream_factory, proof_source_factory) {}
-
-ActiveQuicListener::ActiveQuicListener(
-    Runtime::Loader& runtime, uint32_t worker_index, uint32_t concurrency,
-    Event::Dispatcher& dispatcher, Network::UdpConnectionHandler& parent,
-    Network::SocketSharedPtr listen_socket, Network::ListenerConfig& listener_config,
+    Network::SocketSharedPtr&& listen_socket, Network::ListenerConfig& listener_config,
     const quic::QuicConfig& quic_config, bool kernel_worker_routing,
     const envoy::config::core::v3::RuntimeFeatureFlag& enabled, QuicStatNames& quic_stat_names,
     uint32_t packets_to_read_to_connection_count_ratio,
@@ -338,11 +324,12 @@ ActiveQuicListenerFactory::ActiveQuicListenerFactory(
 
 Network::ConnectionHandler::ActiveUdpListenerPtr ActiveQuicListenerFactory::createActiveUdpListener(
     Runtime::Loader& runtime, uint32_t worker_index, Network::UdpConnectionHandler& parent,
-    Event::Dispatcher& disptacher, Network::ListenerConfig& config) {
+    Network::SocketSharedPtr&& listen_socket_ptr, Event::Dispatcher& disptacher,
+    Network::ListenerConfig& config) {
   ASSERT(crypto_server_stream_factory_.has_value());
   return std::make_unique<ActiveQuicListener>(
-      runtime, worker_index, concurrency_, disptacher, parent, config, quic_config_,
-      kernel_worker_routing_, enabled_, quic_stat_names_,
+      runtime, worker_index, concurrency_, disptacher, parent, std::move(listen_socket_ptr), config,
+      quic_config_, kernel_worker_routing_, enabled_, quic_stat_names_,
       packets_to_read_to_connection_count_ratio_, crypto_server_stream_factory_.value(),
       proof_source_factory_.value());
 }

source/common/quic/active_quic_listener.h

@@ -30,17 +30,7 @@ class ActiveQuicListener : public Envoy::Server::ActiveUdpListenerBase,
 
   ActiveQuicListener(Runtime::Loader& runtime, uint32_t worker_index, uint32_t concurrency,
                      Event::Dispatcher& dispatcher, Network::UdpConnectionHandler& parent,
-                     Network::ListenerConfig& listener_config, const quic::QuicConfig& quic_config,
-                     bool kernel_worker_routing,
-                     const envoy::config::core::v3::RuntimeFeatureFlag& enabled,
-                     QuicStatNames& quic_stat_names,
-                     uint32_t packets_to_read_to_connection_count_ratio,
-                     EnvoyQuicCryptoServerStreamFactoryInterface& crypto_server_stream_factory,
-                     EnvoyQuicProofSourceFactoryInterface& proof_source_factory);
-
-  ActiveQuicListener(Runtime::Loader& runtime, uint32_t worker_index, uint32_t concurrency,
-                     Event::Dispatcher& dispatcher, Network::UdpConnectionHandler& parent,
-                     Network::SocketSharedPtr listen_socket,
+                     Network::SocketSharedPtr&& listen_socket,
                      Network::ListenerConfig& listener_config, const quic::QuicConfig& quic_config,
                      bool kernel_worker_routing,
                      const envoy::config::core::v3::RuntimeFeatureFlag& enabled,
@@ -108,8 +98,9 @@ class ActiveQuicListenerFactory : public Network::ActiveUdpListenerFactory,
   // Network::ActiveUdpListenerFactory.
   Network::ConnectionHandler::ActiveUdpListenerPtr
   createActiveUdpListener(Runtime::Loader& runtime, uint32_t worker_index,
-                          Network::UdpConnectionHandler& parent, Event::Dispatcher& disptacher,
-                          Network::ListenerConfig& config) override;
+                          Network::UdpConnectionHandler& parent,
+                          Network::SocketSharedPtr&& listen_socket_ptr,
+                          Event::Dispatcher& disptacher, Network::ListenerConfig& config) override;
   bool isTransportConnectionless() const override { return false; }
   const Network::Socket::OptionsSharedPtr& socketOptions() const override { return options_; }
 

source/server/active_raw_udp_listener_config.cc

@@ -15,10 +15,11 @@ ActiveRawUdpListenerFactory::ActiveRawUdpListenerFactory(uint32_t concurrency)
 Network::ConnectionHandler::ActiveUdpListenerPtr
 ActiveRawUdpListenerFactory::createActiveUdpListener(Runtime::Loader&, uint32_t worker_index,
                                                      Network::UdpConnectionHandler& parent,
+                                                     Network::SocketSharedPtr&& listen_socket_ptr,
                                                      Event::Dispatcher& dispatcher,
                                                      Network::ListenerConfig& config) {
-  return std::make_unique<ActiveRawUdpListener>(worker_index, concurrency_, parent, dispatcher,
-                                                config);
+  return std::make_unique<ActiveRawUdpListener>(worker_index, concurrency_, parent,
+                                                std::move(listen_socket_ptr), dispatcher, config);
 }
 
 } // namespace Server

source/server/active_raw_udp_listener_config.h

@@ -11,8 +11,9 @@ class ActiveRawUdpListenerFactory : public Network::ActiveUdpListenerFactory {
 
   Network::ConnectionHandler::ActiveUdpListenerPtr
   createActiveUdpListener(Runtime::Loader&, uint32_t worker_index,
-                          Network::UdpConnectionHandler& parent, Event::Dispatcher& disptacher,
-                          Network::ListenerConfig& config) override;
+                          Network::UdpConnectionHandler& parent,
+                          Network::SocketSharedPtr&& listen_socket_ptr,
+                          Event::Dispatcher& disptacher, Network::ListenerConfig& config) override;
   bool isTransportConnectionless() const override { return true; }
   const Network::Socket::OptionsSharedPtr& socketOptions() const override { return options_; }
 

source/server/active_tcp_listener.cc

@@ -14,28 +14,32 @@ namespace Server {
 
 ActiveTcpListener::ActiveTcpListener(Network::TcpConnectionHandler& parent,
                                      Network::ListenerConfig& config, Runtime::Loader& runtime,
-                                     uint32_t worker_index)
-    : OwnedActiveStreamListenerBase(parent, parent.dispatcher(),
-                                    parent.dispatcher().createListener(
-                                        config.listenSocketFactory().getListenSocket(worker_index),
-                                        *this, runtime, config.bindToPort(),
-                                        config.ignoreGlobalConnLimit()),
-                                    config),
-      tcp_conn_handler_(parent) {
-  config.connectionBalancer().registerHandler(*this);
+                                     Network::SocketSharedPtr&& socket,
+                                     Network::Address::InstanceConstSharedPtr& address,
+                                     Network::ConnectionBalancer& connection_balancer)
+    : OwnedActiveStreamListenerBase(
+          parent, parent.dispatcher(),
+          parent.dispatcher().createListener(std::move(socket), *this, runtime, config.bindToPort(),
+                                             config.ignoreGlobalConnLimit()),
+          config),
+      tcp_conn_handler_(parent), connection_balancer_(connection_balancer), address_(address) {
+  connection_balancer_.registerHandler(*this);
 }
 
 ActiveTcpListener::ActiveTcpListener(Network::TcpConnectionHandler& parent,
                                      Network::ListenerPtr&& listener,
-                                     Network::ListenerConfig& config, Runtime::Loader&)
+                                     Network::Address::InstanceConstSharedPtr& address,
+                                     Network::ListenerConfig& config,
+                                     Network::ConnectionBalancer& connection_balancer,
+                                     Runtime::Loader&)
     : OwnedActiveStreamListenerBase(parent, parent.dispatcher(), std::move(listener), config),
-      tcp_conn_handler_(parent) {
-  config.connectionBalancer().registerHandler(*this);
+      tcp_conn_handler_(parent), connection_balancer_(connection_balancer), address_(address) {
+  connection_balancer_.registerHandler(*this);
 }
 
 ActiveTcpListener::~ActiveTcpListener() {
   is_deleting_ = true;
-  config_->connectionBalancer().unregisterHandler(*this);
+  connection_balancer_.unregisterHandler(*this);
 
   // Purge sockets that have not progressed to connections. This should only happen when
   // a listener filter stops iteration and never resumes.
@@ -65,7 +69,6 @@ ActiveTcpListener::~ActiveTcpListener() {
 
 void ActiveTcpListener::updateListenerConfig(Network::ListenerConfig& config) {
   ENVOY_LOG(trace, "replacing listener ", config_->listenerTag(), " by ", config.listenerTag());
-  ASSERT(&config_->connectionBalancer() == &config.connectionBalancer());
   config_ = &config;
 }
 
@@ -98,7 +101,7 @@ void ActiveTcpListener::onAcceptWorker(Network::ConnectionSocketPtr&& socket,
                                        bool rebalanced) {
   if (!rebalanced) {
     Network::BalancedConnectionHandler& target_handler =
-        config_->connectionBalancer().pickTargetHandler(*this);
+        connection_balancer_.pickTargetHandler(*this);
     if (&target_handler != this) {
       target_handler.post(std::move(socket));
       return;
@@ -153,10 +156,11 @@ void ActiveTcpListener::post(Network::ConnectionSocketPtr&& socket) {
   RebalancedSocketSharedPtr socket_to_rebalance = std::make_shared<RebalancedSocket>();
   socket_to_rebalance->socket = std::move(socket);
 
-  dispatcher().post([socket_to_rebalance, tag = config_->listenerTag(),
+  dispatcher().post([socket_to_rebalance, address = address_, tag = config_->listenerTag(),
                      &tcp_conn_handler = tcp_conn_handler_,
                      handoff = config_->handOffRestoredDestinationConnections()]() {
-    auto balanced_handler = tcp_conn_handler.getBalancedHandlerByTag(tag);
+    // `getBalancedHandlerByAddress` only support the IP address but there can be unix socket also.
+    auto balanced_handler = tcp_conn_handler.getBalancedHandlerByTag(tag, *address);
     if (balanced_handler.has_value()) {
       balanced_handler->get().onAcceptWorker(std::move(socket_to_rebalance->socket), handoff, true);
       return;

source/server/active_tcp_listener.h

@@ -27,9 +27,13 @@ class ActiveTcpListener final : public Network::TcpListenerCallbacks,
                                 public Network::BalancedConnectionHandler {
 public:
   ActiveTcpListener(Network::TcpConnectionHandler& parent, Network::ListenerConfig& config,
-                    Runtime::Loader& runtime, uint32_t worker_index);
+                    Runtime::Loader& runtime, Network::SocketSharedPtr&& socket,
+                    Network::Address::InstanceConstSharedPtr& address,
+                    Network::ConnectionBalancer& connection_balancer);
   ActiveTcpListener(Network::TcpConnectionHandler& parent, Network::ListenerPtr&& listener,
-                    Network::ListenerConfig& config, Runtime::Loader& runtime);
+                    Network::Address::InstanceConstSharedPtr& address,
+                    Network::ListenerConfig& config,
+                    Network::ConnectionBalancer& connection_balancer, Runtime::Loader& runtime);
   ~ActiveTcpListener() override;
 
   bool listenerConnectionLimitReached() const {
@@ -81,6 +85,9 @@ class ActiveTcpListener final : public Network::TcpListenerCallbacks,
   // The number of connections currently active on this listener. This is typically used for
   // connection balancing across per-handler listeners.
   std::atomic<uint64_t> num_listener_connections_{};
+
+  Network::ConnectionBalancer& connection_balancer_;
+  Network::Address::InstanceConstSharedPtr address_;
 };
 
 using ActiveTcpListenerOptRef = absl::optional<std::reference_wrapper<ActiveTcpListener>>;