Add logic for instantiating UdpListenSocket based on SocketAddress.protocol in listener config

include/envoy/network/listen_socket.h

@@ -35,6 +35,11 @@ class Socket {
    */
   virtual int fd() const PURE;
 
+  /**
+   * @return the type (stream or datagram) of the socket.
+   */
+  virtual Address::SocketType socketType() const PURE;
+
   /**
    * Close the underlying socket.
    */

include/envoy/server/listener_manager.h

@@ -44,12 +44,14 @@ class ListenerComponentFactory {
   /**
    * Creates a socket.
    * @param address supplies the socket's address.
+   * @param socket_type the type of socket (stream or datagram) to create.
    * @param options to be set on the created socket just before calling 'bind()'.
    * @param bind_to_port supplies whether to actually bind the socket.
    * @return Network::SocketSharedPtr an initialized and potentially bound socket.
    */
   virtual Network::SocketSharedPtr
   createListenSocket(Network::Address::InstanceConstSharedPtr address,
+                     Network::Address::SocketType socket_type,
                      const Network::Socket::OptionsSharedPtr& options, bool bind_to_port) PURE;
 
   /**

source/common/network/listen_socket_impl.h

@@ -92,6 +92,8 @@ template <typename T> class NetworkListenSocket : public ListenSocketImpl {
     setListenSocketOptions(options);
   }
 
+  Address::SocketType socketType() const override { return T::type; }
+
 protected:
   void setPrebindSocketOptions();
 };
@@ -106,6 +108,7 @@ class UdsListenSocket : public ListenSocketImpl {
 public:
   UdsListenSocket(const Address::InstanceConstSharedPtr& address);
   UdsListenSocket(int fd, const Address::InstanceConstSharedPtr& address);
+  Address::SocketType socketType() const override { return Address::SocketType::Stream; }
 };
 
 class ConnectionSocketImpl : public SocketImpl, public ConnectionSocket {
@@ -114,6 +117,9 @@ class ConnectionSocketImpl : public SocketImpl, public ConnectionSocket {
                        const Address::InstanceConstSharedPtr& remote_address)
       : SocketImpl(fd, local_address), remote_address_(remote_address) {}
 
+  // Network::Socket
+  Address::SocketType socketType() const override { return Address::SocketType::Stream; }
+
   // Network::ConnectionSocket
   const Address::InstanceConstSharedPtr& remoteAddress() const override { return remote_address_; }
   void setLocalAddress(const Address::InstanceConstSharedPtr& local_address,

source/common/network/utility.cc

@@ -501,5 +501,26 @@ void Utility::addressToProtobufAddress(const Address::Instance& address,
   }
 }
 
+Address::SocketType
+Utility::protobufAddressSocketType(const envoy::api::v2::core::Address& proto_address) {
+  switch (proto_address.address_case()) {
+  case envoy::api::v2::core::Address::kSocketAddress: {
+    auto protocol = proto_address.socket_address().protocol();
+    switch (protocol) {
+    case envoy::api::v2::core::SocketAddress::TCP:
+      return Address::SocketType::Stream;
+    case envoy::api::v2::core::SocketAddress::UDP:
+      return Address::SocketType::Datagram;
+    default:
+      NOT_REACHED_GCOVR_EXCL_LINE;
+    }
+  }
+  case envoy::api::v2::core::Address::kPipe:
+    return Address::SocketType::Stream;
+  default:
+    NOT_REACHED_GCOVR_EXCL_LINE;
+  }
+}
+
 } // namespace Network
 } // namespace Envoy

source/common/network/utility.h

@@ -239,6 +239,15 @@ class Utility {
   static void addressToProtobufAddress(const Address::Instance& address,
                                        envoy::api::v2::core::Address& proto_address);
 
+  /**
+   * Returns socket type corresponding to SocketAddress.protocol value of the
+   * given address, or SocketType::Stream if the address is a pipe address.
+   * @param proto_address the address protobuf
+   * @return socket type
+   */
+  static Address::SocketType
+  protobufAddressSocketType(const envoy::api::v2::core::Address& proto_address);
+
 private:
   static void throwWithMalformedIp(const std::string& ip_address);
 

source/server/config_validation/server.h

@@ -114,6 +114,7 @@ class ValidationInstance : Logger::Loggable<Logger::Id::main>,
     return ProdListenerComponentFactory::createListenerFilterFactoryList_(filters, context);
   }
   Network::SocketSharedPtr createListenSocket(Network::Address::InstanceConstSharedPtr,
+                                              Network::Address::SocketType,
                                               const Network::Socket::OptionsSharedPtr&,
                                               bool) override {
     // Returned sockets are not currently used so we can return nothing here safely vs. a

source/server/listener_manager_impl.cc

@@ -25,9 +25,24 @@
 #include "extensions/transport_sockets/well_known_names.h"
 
 #include "absl/strings/match.h"
+#include "absl/strings/str_cat.h"
 
 namespace Envoy {
 namespace Server {
+namespace {
+
+std::string toString(Network::Address::SocketType socket_type) {
+  switch (socket_type) {
+  case Network::Address::SocketType::Stream:
+    return "SocketType::Stream";
+  case Network::Address::SocketType::Datagram:
+    return "SocketType::Datagram";
+  default:
+    NOT_REACHED_GCOVR_EXCL_LINE;
+  }
+}
+
+} // namespace
 
 std::vector<Network::FilterFactoryCb> ProdListenerComponentFactory::createNetworkFilterFactoryList_(
     const Protobuf::RepeatedPtrField<envoy::api::v2::listener::Filter>& filters,
@@ -82,16 +97,24 @@ ProdListenerComponentFactory::createListenerFilterFactoryList_(
   return ret;
 }
 
-Network::SocketSharedPtr
-ProdListenerComponentFactory::createListenSocket(Network::Address::InstanceConstSharedPtr address,
-                                                 const Network::Socket::OptionsSharedPtr& options,
-                                                 bool bind_to_port) {
+Network::SocketSharedPtr ProdListenerComponentFactory::createListenSocket(
+    Network::Address::InstanceConstSharedPtr address, Network::Address::SocketType socket_type,
+    const Network::Socket::OptionsSharedPtr& options, bool bind_to_port) {
   ASSERT(address->type() == Network::Address::Type::Ip ||
          address->type() == Network::Address::Type::Pipe);
+  ASSERT(socket_type == Network::Address::SocketType::Stream ||
+         socket_type == Network::Address::SocketType::Datagram);
 
   // For each listener config we share a single socket among all threaded listeners.
   // First we try to get the socket from our parent if applicable.
   if (address->type() == Network::Address::Type::Pipe) {
+    if (socket_type != Network::Address::SocketType::Stream) {
+      // This could be implemented in the future, since Unix domain sockets
+      // support SOCK_DGRAM, but there would need to be a way to specify it in
+      // envoy.api.v2.core.Pipe.
+      throw EnvoyException(
+          fmt::format("socket type {} not supported for pipes", toString(socket_type)));
+    }
     const std::string addr = fmt::format("unix://{}", address->asString());
     const int fd = server_.hotRestart().duplicateParentListenSocket(addr);
     if (fd != -1) {
@@ -101,13 +124,24 @@ ProdListenerComponentFactory::createListenSocket(Network::Address::InstanceConst
     return std::make_shared<Network::UdsListenSocket>(address);
   }
 
-  const std::string addr = fmt::format("tcp://{}", address->asString());
+  const std::string scheme = (socket_type == Network::Address::SocketType::Stream)
+                                 ? Network::Utility::TCP_SCHEME
+                                 : Network::Utility::UDP_SCHEME;
+  const std::string addr = absl::StrCat(scheme, address->asString());
   const int fd = server_.hotRestart().duplicateParentListenSocket(addr);
   if (fd != -1) {
     ENVOY_LOG(debug, "obtained socket for address {} from parent", addr);
-    return std::make_shared<Network::TcpListenSocket>(fd, address, options);
+    if (socket_type == Network::Address::SocketType::Stream) {
+      return std::make_shared<Network::TcpListenSocket>(fd, address, options);
+    } else {
+      return std::make_shared<Network::UdpListenSocket>(fd, address, options);
+    }
+  }
+  if (socket_type == Network::Address::SocketType::Stream) {
+    return std::make_shared<Network::TcpListenSocket>(address, options, bind_to_port);
+  } else {
+    return std::make_shared<Network::UdpListenSocket>(address, options, bind_to_port);
   }
-  return std::make_shared<Network::TcpListenSocket>(address, options, bind_to_port);
 }
 
 DrainManagerPtr
@@ -119,6 +153,7 @@ ListenerImpl::ListenerImpl(const envoy::api::v2::Listener& config, const std::st
                            ListenerManagerImpl& parent, const std::string& name, bool modifiable,
                            bool workers_started, uint64_t hash)
     : parent_(parent), address_(Network::Address::resolveProtoAddress(config.address())),
+      socket_type_(Network::Utility::protobufAddressSocketType(config.address())),
       global_scope_(parent_.server_.stats().createScope("")),
       listener_scope_(
           parent_.server_.stats().createScope(fmt::format("listener.{}.", address_->asString()))),
@@ -774,6 +809,7 @@ bool ListenerManagerImpl::addOrUpdateListener(const envoy::api::v2::Listener& co
     new_listener->setSocket(draining_listener_socket
                                 ? draining_listener_socket
                                 : factory_.createListenSocket(new_listener->address(),
+                                                              new_listener->socketType(),
                                                               new_listener->listenSocketOptions(),
                                                               new_listener->bindToPort()));
     if (workers_started_) {

source/server/listener_manager_impl.h

@@ -61,6 +61,7 @@ class ProdListenerComponentFactory : public ListenerComponentFactory,
     return createListenerFilterFactoryList_(filters, context);
   }
   Network::SocketSharedPtr createListenSocket(Network::Address::InstanceConstSharedPtr address,
+                                              Network::Address::SocketType socket_type,
                                               const Network::Socket::OptionsSharedPtr& options,
                                               bool bind_to_port) override;
   DrainManagerPtr createDrainManager(envoy::api::v2::Listener::DrainType drain_type) override;
@@ -227,6 +228,7 @@ class ListenerImpl : public Network::ListenerConfig,
   }
 
   Network::Address::InstanceConstSharedPtr address() const { return address_; }
+  Network::Address::SocketType socketType() const { return socket_type_; }
   const envoy::api::v2::Listener& config() { return config_; }
   const Network::SocketSharedPtr& getSocket() const { return socket_; }
   void debugLog(const std::string& message);
@@ -386,6 +388,7 @@ class ListenerImpl : public Network::ListenerConfig,
 
   ListenerManagerImpl& parent_;
   Network::Address::InstanceConstSharedPtr address_;
+  Network::Address::SocketType socket_type_;
   Network::SocketSharedPtr socket_;
   Stats::ScopePtr global_scope_;   // Stats with global named scope, but needed for LDS cleanup.
   Stats::ScopePtr listener_scope_; // Stats with listener named scope.

test/common/network/listen_socket_impl_test.cc

@@ -70,6 +70,7 @@ class ListenSocketImplTest : public testing::TestWithParam<Address::IpVersion> {
         }
         continue;
       }
+
       // TODO (conqerAtapple): This is unfortunate. We should be able to templatize this
       // instead of if block.
       if (NetworkSocketTrait<Type>::type == Address::SocketType::Stream) {
@@ -78,6 +79,7 @@ class ListenSocketImplTest : public testing::TestWithParam<Address::IpVersion> {
 
       EXPECT_EQ(addr->ip()->port(), socket1->localAddress()->ip()->port());
       EXPECT_EQ(addr->ip()->addressAsString(), socket1->localAddress()->ip()->addressAsString());
+      EXPECT_EQ(Type, socket1->socketType());
 
       auto option2 = std::make_unique<MockSocketOption>();
       auto options2 = std::make_shared<std::vector<Network::Socket::OptionConstSharedPtr>>();
@@ -103,6 +105,7 @@ class ListenSocketImplTest : public testing::TestWithParam<Address::IpVersion> {
     EXPECT_EQ(version_, socket->localAddress()->ip()->version());
     EXPECT_EQ(loopback->ip()->addressAsString(), socket->localAddress()->ip()->addressAsString());
     EXPECT_GT(socket->localAddress()->ip()->port(), 0U);
+    EXPECT_EQ(Type, socket->socketType());
   }
 };
 

test/common/network/utility_test.cc

@@ -320,6 +320,29 @@ TEST(NetworkUtility, AddressToProtobufAddress) {
   }
 }
 
+TEST(NetworkUtility, ProtobufAddressSocketType) {
+  {
+    envoy::api::v2::core::Address proto_address;
+    proto_address.mutable_socket_address();
+    EXPECT_EQ(Address::SocketType::Stream, Utility::protobufAddressSocketType(proto_address));
+  }
+  {
+    envoy::api::v2::core::Address proto_address;
+    proto_address.mutable_socket_address()->set_protocol(envoy::api::v2::core::SocketAddress::TCP);
+    EXPECT_EQ(Address::SocketType::Stream, Utility::protobufAddressSocketType(proto_address));
+  }
+  {
+    envoy::api::v2::core::Address proto_address;
+    proto_address.mutable_socket_address()->set_protocol(envoy::api::v2::core::SocketAddress::UDP);
+    EXPECT_EQ(Address::SocketType::Datagram, Utility::protobufAddressSocketType(proto_address));
+  }
+  {
+    envoy::api::v2::core::Address proto_address;
+    proto_address.mutable_pipe();
+    EXPECT_EQ(Address::SocketType::Stream, Utility::protobufAddressSocketType(proto_address));
+  }
+}
+
 TEST(PortRangeListTest, Errors) {
   {
     std::string port_range_str = "a1";

test/mocks/network/mocks.h

@@ -193,6 +193,7 @@ class MockListenSocket : public Socket {
 
   MOCK_CONST_METHOD0(localAddress, const Address::InstanceConstSharedPtr&());
   MOCK_CONST_METHOD0(fd, int());
+  MOCK_CONST_METHOD0(socketType, Address::SocketType());
   MOCK_METHOD0(close, void());
   MOCK_METHOD1(addOption_, void(const Socket::OptionConstSharedPtr& option));
   MOCK_METHOD1(addOptions_, void(const Socket::OptionsSharedPtr& options));
@@ -238,6 +239,7 @@ class MockConnectionSocket : public ConnectionSocket {
   MOCK_METHOD1(addOptions_, void(const Socket::OptionsSharedPtr&));
   MOCK_CONST_METHOD0(options, const Network::ConnectionSocket::OptionsSharedPtr&());
   MOCK_CONST_METHOD0(fd, int());
+  MOCK_CONST_METHOD0(socketType, Address::SocketType());
   MOCK_METHOD0(close, void());
 
   Address::InstanceConstSharedPtr local_address_;

test/mocks/server/mocks.cc

@@ -83,16 +83,16 @@ MockOverloadManager::~MockOverloadManager() {}
 
 MockListenerComponentFactory::MockListenerComponentFactory()
     : socket_(std::make_shared<NiceMock<Network::MockListenSocket>>()) {
-  ON_CALL(*this, createListenSocket(_, _, _))
-      .WillByDefault(Invoke([&](Network::Address::InstanceConstSharedPtr,
-                                const Network::Socket::OptionsSharedPtr& options,
-                                bool) -> Network::SocketSharedPtr {
-        if (!Network::Socket::applyOptions(options, *socket_,
-                                           envoy::api::v2::core::SocketOption::STATE_PREBIND)) {
-          throw EnvoyException("MockListenerComponentFactory: Setting socket options failed");
-        }
-        return socket_;
-      }));
+  ON_CALL(*this, createListenSocket(_, _, _, _))
+      .WillByDefault(Invoke(
+          [&](Network::Address::InstanceConstSharedPtr, Network::Address::SocketType,
+              const Network::Socket::OptionsSharedPtr& options, bool) -> Network::SocketSharedPtr {
+            if (!Network::Socket::applyOptions(options, *socket_,
+                                               envoy::api::v2::core::SocketOption::STATE_PREBIND)) {
+              throw EnvoyException("MockListenerComponentFactory: Setting socket options failed");
+            }
+            return socket_;
+          }));
 }
 MockListenerComponentFactory::~MockListenerComponentFactory() {}
 

test/mocks/server/mocks.h

@@ -220,8 +220,9 @@ class MockListenerComponentFactory : public ListenerComponentFactory {
                std::vector<Network::ListenerFilterFactoryCb>(
                    const Protobuf::RepeatedPtrField<envoy::api::v2::listener::ListenerFilter>&,
                    Configuration::ListenerFactoryContext& context));
-  MOCK_METHOD3(createListenSocket,
+  MOCK_METHOD4(createListenSocket,
                Network::SocketSharedPtr(Network::Address::InstanceConstSharedPtr address,
+                                        Network::Address::SocketType socket_type,
                                         const Network::Socket::OptionsSharedPtr& options,
                                         bool bind_to_port));
   MOCK_METHOD1(createDrainManager_, DrainManager*(envoy::api::v2::Listener::DrainType drain_type));

test/server/BUILD

@@ -160,6 +160,7 @@ envoy_cc_test(
         "//source/common/network:listen_socket_lib",
         "//source/common/network:socket_option_lib",
         "//source/common/network:utility_lib",
+        "//source/common/protobuf",
         "//source/common/ssl:ssl_socket_lib",
         "//source/extensions/filters/listener/original_dst:config",
         "//source/extensions/filters/listener/tls_inspector:config",