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/config/address_json.cc

@@ -16,6 +16,13 @@ void AddressJson::translateAddress(const std::string& json_address, bool url, bo
     if (instance->type() == Network::Address::Type::Ip) {
       address.mutable_socket_address()->set_address(instance->ip()->addressAsString());
       address.mutable_socket_address()->set_port_value(instance->ip()->port());
+      if (url) {
+        if (Network::Utility::urlIsTcpScheme(json_address)) {
+          address.mutable_socket_address()->set_protocol(envoy::api::v2::core::SocketAddress::TCP);
+        } else if (Network::Utility::urlIsUdpScheme(json_address)) {
+          address.mutable_socket_address()->set_protocol(envoy::api::v2::core::SocketAddress::UDP);
+        }
+      }
     } else {
       ASSERT(instance->type() == Network::Address::Type::Pipe);
       address.mutable_pipe()->set_path(instance->asString());
@@ -25,12 +32,21 @@ void AddressJson::translateAddress(const std::string& json_address, bool url, bo
 
   // We don't have v1 JSON with unresolved addresses in non-URL form.
   ASSERT(url);
-  // Non-TCP scheme (e.g. Unix scheme) is not supported with unresolved address.
-  if (!Network::Utility::urlIsTcpScheme(json_address)) {
-    throw EnvoyException(fmt::format("unresolved URL must be TCP scheme, got: {}", json_address));
+  if (Network::Utility::urlIsTcpScheme(json_address)) {
+    address.mutable_socket_address()->set_address(Network::Utility::hostFromTcpUrl(json_address));
+    address.mutable_socket_address()->set_port_value(
+        Network::Utility::portFromTcpUrl(json_address));
+    address.mutable_socket_address()->set_protocol(envoy::api::v2::core::SocketAddress::TCP);
+  } else if (Network::Utility::urlIsUdpScheme(json_address)) {
+    address.mutable_socket_address()->set_address(Network::Utility::hostFromUdpUrl(json_address));
+    address.mutable_socket_address()->set_port_value(
+        Network::Utility::portFromUdpUrl(json_address));
+    address.mutable_socket_address()->set_protocol(envoy::api::v2::core::SocketAddress::UDP);
+  } else {
+    // Non-TCP/UDP scheme (e.g. Unix scheme) is not supported with unresolved address.
+    throw EnvoyException(
+        fmt::format("unresolved URL must be TCP or UDP scheme, got: {}", json_address));
   }
-  address.mutable_socket_address()->set_address(Network::Utility::hostFromTcpUrl(json_address));
-  address.mutable_socket_address()->set_port_value(Network::Utility::portFromTcpUrl(json_address));
 }
 
 void AddressJson::translateCidrRangeList(

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 p = proto_address.socket_address().protocol();
+    switch (p) {
+    case envoy::api::v2::core::SocketAddress::TCP:
+      return Address::SocketType::Stream;
+    case envoy::api::v2::core::SocketAddress::UDP:
+      return Address::SocketType::Datagram;
+    default:
+      throw EnvoyException(fmt::format("unknown protocol value: {}", p));
+    }
+  }
+  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

@@ -28,6 +28,20 @@
 
 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:
+    return fmt::format("unknown ({})", static_cast<int>(socket_type));
+  }
+}
+
+} // namespace
 
 std::vector<Network::FilterFactoryCb> ProdListenerComponentFactory::createNetworkFilterFactoryList_(
     const Protobuf::RepeatedPtrField<envoy::api::v2::listener::Filter>& filters,
@@ -82,16 +96,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 +123,22 @@ 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) ? "tcp" : "udp";
+  const std::string addr = fmt::format("{}://{}", 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 +150,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 +806,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/config/BUILD

@@ -9,6 +9,14 @@ load(
 
 envoy_package()
 
+envoy_cc_test(
+    name = "address_json_test",
+    srcs = ["address_json_test.cc"],
+    deps = [
+        "//source/common/config:address_json_lib",
+    ],
+)
+
 envoy_cc_test(
     name = "filesystem_subscription_impl_test",
     srcs = ["filesystem_subscription_impl_test.cc"],

test/common/config/address_json_test.cc

@@ -0,0 +1,94 @@
+#include "envoy/common/exception.h"
+
+#include "common/config/address_json.h"
+
+#include "gtest/gtest.h"
+
+namespace Envoy {
+namespace Config {
+
+TEST(AddressJsonTest, TranslateResolvedUrlAddress) {
+  {
+    envoy::api::v2::core::Address proto_address;
+    AddressJson::translateAddress("tcp://1.2.3.4:5678", /*url=*/true, /*resolved=*/true,
+                                  proto_address);
+    EXPECT_EQ(envoy::api::v2::core::Address::kSocketAddress, proto_address.address_case());
+    EXPECT_EQ(envoy::api::v2::core::SocketAddress::TCP, proto_address.socket_address().protocol());
+    EXPECT_EQ("1.2.3.4", proto_address.socket_address().address());
+    EXPECT_EQ(5678, proto_address.socket_address().port_value());
+  }
+  {
+    envoy::api::v2::core::Address proto_address;
+    AddressJson::translateAddress("udp://1.2.3.4:5678", /*url=*/true, /*resolved=*/true,
+                                  proto_address);
+    EXPECT_EQ(envoy::api::v2::core::Address::kSocketAddress, proto_address.address_case());
+    EXPECT_EQ(envoy::api::v2::core::SocketAddress::UDP, proto_address.socket_address().protocol());
+    EXPECT_EQ("1.2.3.4", proto_address.socket_address().address());
+    EXPECT_EQ(5678, proto_address.socket_address().port_value());
+  }
+  {
+    envoy::api::v2::core::Address proto_address;
+    AddressJson::translateAddress("unix://foo/bar", /*url=*/true, /*resolved=*/true, proto_address);
+    EXPECT_EQ(envoy::api::v2::core::Address::kPipe, proto_address.address_case());
+    EXPECT_EQ("foo/bar", proto_address.pipe().path());
+  }
+  {
+    envoy::api::v2::core::Address proto_address;
+    EXPECT_THROW(AddressJson::translateAddress("invalid://1.2.3.4:5678", /*url=*/true,
+                                               /*resolved=*/true, proto_address),
+                 EnvoyException);
+  }
+}
+
+TEST(AddressJsonTest, TranslateResolvedNonUrlAddress) {
+  {
+    envoy::api::v2::core::Address proto_address;
+    AddressJson::translateAddress("1.2.3.4:5678", /*url=*/false, /*resolved=*/true, proto_address);
+    EXPECT_EQ(envoy::api::v2::core::Address::kSocketAddress, proto_address.address_case());
+    EXPECT_EQ(envoy::api::v2::core::SocketAddress::TCP, proto_address.socket_address().protocol());
+    EXPECT_EQ("1.2.3.4", proto_address.socket_address().address());
+    EXPECT_EQ(5678, proto_address.socket_address().port_value());
+  }
+  {
+    envoy::api::v2::core::Address proto_address;
+    EXPECT_THROW(AddressJson::translateAddress("tcp://1.2.3.4:5678", /*url=*/false,
+                                               /*resolved=*/true, proto_address),
+                 EnvoyException);
+  }
+}
+
+TEST(AddressJsonTest, TranslateUnresolvedUrlAddress) {
+  {
+    envoy::api::v2::core::Address proto_address;
+    AddressJson::translateAddress("tcp://foo.com:5678", /*url=*/true, /*resolved=*/false,
+                                  proto_address);
+    EXPECT_EQ(envoy::api::v2::core::Address::kSocketAddress, proto_address.address_case());
+    EXPECT_EQ(envoy::api::v2::core::SocketAddress::TCP, proto_address.socket_address().protocol());
+    EXPECT_EQ("foo.com", proto_address.socket_address().address());
+    EXPECT_EQ(5678, proto_address.socket_address().port_value());
+  }
+  {
+    envoy::api::v2::core::Address proto_address;
+    AddressJson::translateAddress("udp://bar.com:5678", /*url=*/true, /*resolved=*/false,
+                                  proto_address);
+    EXPECT_EQ(envoy::api::v2::core::Address::kSocketAddress, proto_address.address_case());
+    EXPECT_EQ(envoy::api::v2::core::SocketAddress::UDP, proto_address.socket_address().protocol());
+    EXPECT_EQ("bar.com", proto_address.socket_address().address());
+    EXPECT_EQ(5678, proto_address.socket_address().port_value());
+  }
+  {
+    envoy::api::v2::core::Address proto_address;
+    EXPECT_THROW(AddressJson::translateAddress("unix://foo/bar", /*url=*/true, /*resolved=*/false,
+                                               proto_address),
+                 EnvoyException);
+  }
+  {
+    envoy::api::v2::core::Address proto_address;
+    EXPECT_THROW(AddressJson::translateAddress("invalid://qux.com:5678", /*url=*/true,
+                                               /*resolved=*/false, proto_address),
+                 EnvoyException);
+  }
+}
+
+} // namespace Config
+} // namespace Envoy

test/common/config/utility_test.cc

@@ -188,7 +188,7 @@ TEST(UtilityTest, UnixClusterDns) {
   Stats::StatsOptionsImpl stats_options;
   EXPECT_THROW_WITH_MESSAGE(
       Config::CdsJson::translateCluster(*json_object_ptr, eds_config, cluster, stats_options),
-      EnvoyException, "unresolved URL must be TCP scheme, got: unix:///test.sock");
+      EnvoyException, "unresolved URL must be TCP or UDP scheme, got: unix:///test.sock");
 }
 
 TEST(UtilityTest, UnixClusterStatic) {