udp_listener: refactor ActiveUdpListener creation

api/envoy/api/v2/BUILD

@@ -109,6 +109,7 @@ api_proto_library_internal(
         "//envoy/api/v2/core:address",
         "//envoy/api/v2/core:base",
         "//envoy/api/v2/listener",
+        "//envoy/api/v2/listener:udp_listener_config",
     ],
 )
 
@@ -120,6 +121,7 @@ api_go_grpc_library(
         "//envoy/api/v2/core:address_go_proto",
         "//envoy/api/v2/core:base_go_proto",
         "//envoy/api/v2/listener:listener_go_proto",
+        "//envoy/api/v2/listener:udp_listener_config_go_proto",
     ],
 )
 

api/envoy/api/v2/lds.proto

@@ -12,6 +12,7 @@ import "envoy/api/v2/core/address.proto";
 import "envoy/api/v2/core/base.proto";
 import "envoy/api/v2/discovery.proto";
 import "envoy/api/v2/listener/listener.proto";
+import "envoy/api/v2/listener/udp_listener_config.proto";
 
 import "google/api/annotations.proto";
 import "google/protobuf/duration.proto";
@@ -44,7 +45,7 @@ service ListenerDiscoveryService {
   }
 }
 
-// [#comment:next free field: 18]
+// [#comment:next free field: 19]
 message Listener {
   // The unique name by which this listener is known. If no name is provided,
   // Envoy will allocate an internal UUID for the listener. If the listener is to be dynamically
@@ -194,4 +195,13 @@ message Listener {
 
   // Specifies the intended direction of the traffic relative to the local Envoy.
   core.TrafficDirection traffic_direction = 16;
+
+  // If the protocol in the listener socket address in :ref:`protocol
+  // <envoy_api_field_core.SocketAddress.protocol>` is :ref:'UDP
+  // <envoy_api_field_core.Protocol.UDP>`, this field specifies the actual udp listener to create,
+  // i.e. :ref:`udp_listener_name
+  // <envoy_api_field_listener.UdpListenerConfig.udp_listener_name>` =
+  // "raw_udp_listener" for creating an ActiveUdpListener. If not present, treat it as
+  // "raw_udp_listener".
+  listener.UdpListenerConfig udp_listener_config = 18;
 }

api/envoy/api/v2/listener/BUILD

@@ -22,3 +22,20 @@ api_go_proto_library(
         "//envoy/api/v2/core:base_go_proto",
     ],
 )
+
+api_proto_library_internal(
+    name = "udp_listener_config",
+    srcs = ["udp_listener_config.proto"],
+    visibility = ["//envoy/api/v2:friends"],
+    deps = [
+        "//envoy/api/v2/core:base",
+    ],
+)
+
+api_go_proto_library(
+    name = "udp_listener_config",
+    proto = ":udp_listener_config",
+    deps = [
+        "//envoy/api/v2/core:base_go_proto",
+    ],
+)

api/envoy/api/v2/listener/udp_listener_config.proto

@@ -0,0 +1,31 @@
+syntax = "proto3";
+
+package envoy.api.v2.listener;
+
+option java_outer_classname = "ListenerProto";
+option java_multiple_files = true;
+option java_package = "io.envoyproxy.envoy.api.v2.listener";
+option go_package = "listener";
+option csharp_namespace = "Envoy.Api.V2.ListenerNS";
+option ruby_package = "Envoy::Api::V2::ListenerNS";
+
+import "google/protobuf/struct.proto";
+import "google/protobuf/any.proto";
+
+// [#protodoc-title: Udp Listener Config]
+// Listener :ref:`configuration overview <config_listeners>`
+
+message UdpListenerConfig {
+  // Used to look up UDP listener factory, matches one of the implementation of
+  // Server::ActiveUdpListenerConfigFactory::name().
+  // If not specified, treat as "raw_udp_listener".
+  string udp_listener_name = 1;
+
+  // Used to create a specific listener factory. To some factory, e.g.
+  // Server::ActiveRawUdpListenerFactory, config is not needed.
+  oneof config_type {
+    google.protobuf.Struct config = 2;
+
+    google.protobuf.Any typed_config = 3;
+  }
+}

docs/build.sh

@@ -82,6 +82,7 @@ PROTO_RST="
   /envoy/api/v2/srds/envoy/api/v2/srds.proto.rst
   /envoy/api/v2/lds/envoy/api/v2/lds.proto.rst
   /envoy/api/v2/listener/listener/envoy/api/v2/listener/listener.proto.rst
+  /envoy/api/v2/listener/udp_listener_config/envoy/api/v2/listener/udp_listener_config.proto.rst
   /envoy/api/v2/ratelimit/ratelimit/envoy/api/v2/ratelimit/ratelimit.proto.rst
   /envoy/config/accesslog/v2/als/envoy/config/accesslog/v2/als.proto.rst
   /envoy/config/accesslog/v2/file/envoy/config/accesslog/v2/file.proto.rst

docs/root/api-v2/listeners/listeners.rst

@@ -7,3 +7,4 @@ Listeners
 
   ../api/v2/lds.proto
   ../api/v2/listener/listener.proto
+  ../api/v2/listener/udp_listener_config.proto

include/envoy/network/listener.h

@@ -11,6 +11,11 @@
 #include "envoy/stats/scope.h"
 
 namespace Envoy {
+
+namespace Server {
+class ActiveUdpListenerFactory;
+} // namespace Server
+
 namespace Network {
 
 class UdpListenerFilterManager;
@@ -90,6 +95,12 @@ class ListenerConfig {
    * @return const std::string& the listener's name.
    */
   virtual const std::string& name() const PURE;
+
+  /**
+   * @return factory pointer if listening on UDP socket, otherwise return
+   * nullptr.
+   */
+  virtual Server::ActiveUdpListenerFactory* udpListenerFactory() PURE;
 };
 
 /**

source/common/network/BUILD

@@ -9,6 +9,16 @@ load(
 
 envoy_package()
 
+envoy_cc_library(
+    name = "active_raw_udp_listener_config",
+    srcs = ["active_raw_udp_listener_config.cc"],
+    hdrs = ["active_raw_udp_listener_config.h"],
+    deps = [
+        "//include/envoy/registry",
+        "//source/server:active_udp_listener_config_interface",
+    ],
+)
+
 envoy_cc_library(
     name = "address_lib",
     srcs = [

source/common/network/active_raw_udp_listener_config.cc

@@ -0,0 +1,17 @@
+#include "common/network/active_raw_udp_listener_config.h"
+
+namespace Envoy {
+namespace Network {
+
+std::unique_ptr<Server::ActiveUdpListenerFactory>
+ActiveRawUdpListenerConfigFactory::createActiveUdpListenerFactory(
+    const Protobuf::Message& /*message*/) {
+  return std::make_unique<Server::ActiveRawUdpListenerFactory>();
+}
+
+std::string ActiveRawUdpListenerConfigFactory::name() { return "raw_udp_listener"; }
+
+REGISTER_FACTORY(ActiveRawUdpListenerConfigFactory, Server::ActiveUdpListenerConfigFactory);
+
+} // namespace Network
+} // namespace Envoy

source/common/network/active_raw_udp_listener_config.h

@@ -0,0 +1,23 @@
+#include "envoy/registry/registry.h"
+
+#include "server/active_udp_listener_config.h"
+#include "server/connection_handler_impl.h"
+
+namespace Envoy {
+namespace Network {
+
+// This class uses a protobuf config to create a UDP listener factory which
+// creates a Server::ConnectionHandlerImpl::ActiveUdpListener.
+// This is the default UDP listener if not specified in config.
+class ActiveRawUdpListenerConfigFactory : public Server::ActiveUdpListenerConfigFactory {
+
+  std::unique_ptr<Server::ActiveUdpListenerFactory>
+  createActiveUdpListenerFactory(const Protobuf::Message&) override;
+
+  std::string name() override;
+};
+
+DECLARE_FACTORY(ActiveRawUdpListenerConfigFactory);
+
+} // namespace Network
+} // namespace Envoy

source/server/BUILD

@@ -10,6 +10,12 @@ load(
 
 envoy_package()
 
+envoy_cc_library(
+    name = "active_udp_listener_config_interface",
+    hdrs = ["active_udp_listener_config.h"],
+    deps = [":connection_handler_lib"],
+)
+
 envoy_cc_library(
     name = "backtrace_lib",
     hdrs = ["backtrace.h"],
@@ -255,6 +261,7 @@ envoy_cc_library(
     srcs = ["listener_manager_impl.cc"],
     hdrs = ["listener_manager_impl.h"],
     deps = [
+        ":active_udp_listener_config_interface",
         ":configuration_lib",
         ":drain_manager_lib",
         ":filter_chain_manager_lib",

source/server/active_udp_listener_config.h

@@ -0,0 +1,20 @@
+#include "server/connection_handler_impl.h"
+
+namespace Envoy {
+namespace Server {
+
+// Interface to create udp listener according to
+// envoy::api::v2::listener::UdpListenerConfig.udp_listener_name.
+class ActiveUdpListenerConfigFactory {
+public:
+  virtual ~ActiveUdpListenerConfigFactory() = default;
+
+  virtual std::unique_ptr<ActiveUdpListenerFactory>
+  createActiveUdpListenerFactory(const Protobuf::Message&) PURE;
+
+  // Used to identify which udp listener to create: quic or raw udp.
+  virtual std::string name() PURE;
+};
+
+} // namespace Server
+} // namespace Envoy

source/server/connection_handler_impl.cc

@@ -27,8 +27,7 @@ void ConnectionHandlerImpl::addListener(Network::ListenerConfig& config) {
   } else {
     ASSERT(socket_type == Network::Address::SocketType::Datagram,
            "Only datagram/stream listener supported");
-    ActiveUdpListenerPtr udp(new ActiveUdpListener(*this, config));
-    listener = std::move(udp);
+    listener = config.udpListenerFactory()->createActiveUdpListener(*this, config);
   }
 
   if (disable_listeners_) {

source/server/connection_handler_impl.h

@@ -59,23 +59,14 @@ class ConnectionHandlerImpl : public Network::ConnectionHandler, NonCopyable {
 
   Network::Listener* findListenerByAddress(const Network::Address::Instance& address) override;
 
-private:
   struct ActiveListenerBase;
   using ActiveListenerBasePtr = std::unique_ptr<ActiveListenerBase>;
 
-  struct ActiveTcpListener;
-  using ActiveTcpListenerPtr = std::unique_ptr<ActiveTcpListener>;
-
   struct ActiveUdpListener;
   using ActiveUdpListenerPtr = std::unique_ptr<ActiveUdpListener>;
 
   ActiveListenerBase* findActiveListenerByAddress(const Network::Address::Instance& address);
 
-  struct ActiveConnection;
-  using ActiveConnectionPtr = std::unique_ptr<ActiveConnection>;
-  struct ActiveSocket;
-  using ActiveSocketPtr = std::unique_ptr<ActiveSocket>;
-
   /**
    * Wrapper for an active listener owned by this handler.
    */
@@ -96,6 +87,7 @@ class ConnectionHandlerImpl : public Network::ConnectionHandler, NonCopyable {
 
   /**
    * Wrapper for an active udp listener owned by this handler.
+   * TODO(danzh): rename to ActiveRawUdpListener.
    */
   struct ActiveUdpListener : public Network::UdpListenerCallbacks,
                              public ActiveListenerBase,
@@ -122,6 +114,14 @@ class ConnectionHandlerImpl : public Network::ConnectionHandler, NonCopyable {
     Network::UdpListenerReadFilterPtr read_filter_;
   };
 
+private:
+  struct ActiveTcpListener;
+  using ActiveTcpListenerPtr = std::unique_ptr<ActiveTcpListener>;
+  struct ActiveConnection;
+  using ActiveConnectionPtr = std::unique_ptr<ActiveConnection>;
+  struct ActiveSocket;
+  using ActiveSocketPtr = std::unique_ptr<ActiveSocket>;
+
   /**
    * Wrapper for an active tcp listener owned by this handler.
    */
@@ -230,5 +230,19 @@ class ConnectionHandlerImpl : public Network::ConnectionHandler, NonCopyable {
   bool disable_listeners_;
 };
 
+class ActiveUdpListenerFactory {
+public:
+  virtual ~ActiveUdpListenerFactory() = default;
+
+  virtual ConnectionHandlerImpl::ActiveListenerBasePtr
+  createActiveUdpListener(ConnectionHandlerImpl& parent, Network::ListenerConfig& config) PURE;
+};
+
+class ActiveRawUdpListenerFactory : public ActiveUdpListenerFactory {
+  ConnectionHandlerImpl::ActiveListenerBasePtr
+  createActiveUdpListener(ConnectionHandlerImpl& parent, Network::ListenerConfig& config) override {
+    return std::make_unique<ConnectionHandlerImpl::ActiveUdpListener>(parent, config);
+  }
+};
 } // namespace Server
 } // namespace Envoy

source/server/http/BUILD

@@ -59,6 +59,7 @@ envoy_cc_library(
         "//source/common/stats:stats_lib",
         "//source/common/upstream:host_utility_lib",
         "//source/extensions/access_loggers/file:file_access_log_lib",
+        "//source/server:connection_handler_lib",
         "@envoy_api//envoy/admin/v2alpha:certs_cc",
         "@envoy_api//envoy/admin/v2alpha:clusters_cc",
         "@envoy_api//envoy/admin/v2alpha:config_dump_cc",

source/server/http/admin.h

@@ -31,6 +31,7 @@
 #include "common/router/scoped_config_impl.h"
 #include "common/stats/isolated_store_impl.h"
 
+#include "server/connection_handler_impl.h"
 #include "server/http/config_tracker_impl.h"
 
 #include "absl/strings/string_view.h"
@@ -319,11 +320,15 @@ class AdminImpl : public Admin,
     Stats::Scope& listenerScope() override { return *scope_; }
     uint64_t listenerTag() const override { return 0; }
     const std::string& name() const override { return name_; }
+    Server::ActiveUdpListenerFactory* udpListenerFactory() override {
+      return udp_listener_factory_.get();
+    }
 
     AdminImpl& parent_;
     const std::string name_;
     Stats::ScopePtr scope_;
     Http::ConnectionManagerListenerStats stats_;
+    std::unique_ptr<Server::ActiveUdpListenerFactory> udp_listener_factory_;
   };
   using AdminListenerPtr = std::unique_ptr<AdminListener>;
 

source/server/listener_manager_impl.cc

@@ -19,6 +19,7 @@
 #include "common/network/utility.h"
 #include "common/protobuf/utility.h"
 
+#include "server/active_udp_listener_config.h"
 #include "server/configuration_impl.h"
 #include "server/drain_manager_impl.h"
 #include "server/filter_chain_manager_impl.h"
@@ -221,6 +222,16 @@ ListenerImpl::ListenerImpl(const envoy::api::v2::Listener& config, const std::st
     addListenSocketOptions(Network::SocketOptionFactory::buildIpPacketInfoOptions());
     // Needed to return receive buffer overflown indicator.
     addListenSocketOptions(Network::SocketOptionFactory::buildRxQueueOverFlowOptions());
+    std::string listener_name =
+        config.has_udp_listener_config() ? config.udp_listener_config().udp_listener_name() : "";
+    if (listener_name.empty()) {
+      listener_name = "raw_udp_listener";
+    }
+    udp_listener_factory_ =
+        Config::Utility::getAndCheckFactory<ActiveUdpListenerConfigFactory>(listener_name)
+            .createActiveUdpListenerFactory(config.has_udp_listener_config()
+                                                ? config.udp_listener_config()
+                                                : envoy::api::v2::listener::UdpListenerConfig());
   }
 
   if (!config.listener_filters().empty()) {

source/server/listener_manager_impl.h

@@ -277,6 +277,7 @@ class ListenerImpl : public Network::ListenerConfig,
   Stats::Scope& listenerScope() override { return *listener_scope_; }
   uint64_t listenerTag() const override { return listener_tag_; }
   const std::string& name() const override { return name_; }
+  ActiveUdpListenerFactory* udpListenerFactory() override { return udp_listener_factory_.get(); }
 
   // Server::Configuration::ListenerFactoryContext
   AccessLog::AccessLogManager& accessLogManager() override {
@@ -376,6 +377,7 @@ class ListenerImpl : public Network::ListenerConfig,
   Network::Socket::OptionsSharedPtr listen_socket_options_;
   const std::chrono::milliseconds listener_filters_timeout_;
   const bool continue_on_listener_filters_timeout_;
+  std::unique_ptr<Server::ActiveUdpListenerFactory> udp_listener_factory_;
   // to access ListenerManagerImpl::factory_.
   friend class ListenerFilterChainFactoryBuilder;
 };