quic: Connect client UDP sockets before usage

contrib/vcl/source/vcl_io_handle.cc

@@ -166,6 +166,8 @@ Api::IoCallUint64Result VclIoHandle::close() {
 
 bool VclIoHandle::isOpen() const { return VCL_SH_VALID(sh_); }
 
+bool VclIoHandle::isConnected() const { return false; }
+
 Api::IoCallUint64Result VclIoHandle::readv(uint64_t max_length, Buffer::RawSlice* slices,
                                            uint64_t num_slice) {
   if (!VCL_SH_VALID(sh_)) {

contrib/vcl/source/vcl_io_handle.h

@@ -31,6 +31,7 @@ class VclIoHandle : public Envoy::Network::IoHandle,
   os_fd_t fdDoNotUse() const override { return fd_; }
   Api::IoCallUint64Result close() override;
   bool isOpen() const override;
+  bool isConnected() const override;
   Api::IoCallUint64Result readv(uint64_t max_length, Buffer::RawSlice* slices,
                                 uint64_t num_slice) override;
   Api::IoCallUint64Result read(Buffer::Instance& buffer,

envoy/network/io_handle.h

@@ -64,6 +64,12 @@ class IoHandle {
    */
   virtual bool isOpen() const PURE;
 
+  /**
+   * Return true if connect() has successfully been called on the socket.
+   * Use isOpen() to check if the socket is still connected or not.
+   */
+  virtual bool isConnected() const PURE;
+
   /**
    * Read data into given slices.
    * @param max_length supplies the maximum length to read.

source/common/http/http3/conn_pool.cc

@@ -148,16 +148,12 @@ Http3ConnPoolImpl::createClientConnection(Quic::QuicStatNames& quic_stat_names,
   auto upstream_local_address_selector = host()->cluster().getUpstreamLocalAddressSelector();
   auto upstream_local_address =
       upstream_local_address_selector->getUpstreamLocalAddress(address, socketOptions());
-  auto source_address = upstream_local_address.address_;
-
-  if (source_address == nullptr) {
-    source_address = Network::Utility::getLocalAddress(address->ip()->version());
-  }
 
   return Quic::createQuicNetworkConnection(
-      quic_info_, std::move(crypto_config), server_id_, dispatcher(), address, source_address,
-      quic_stat_names, rtt_cache, scope, upstream_local_address.socket_options_,
-      transportSocketOptions(), connection_id_generator_, host_->transportSocketFactory());
+      quic_info_, std::move(crypto_config), server_id_, dispatcher(), address,
+      upstream_local_address.address_, quic_stat_names, rtt_cache, scope,
+      upstream_local_address.socket_options_, transportSocketOptions(), connection_id_generator_,
+      host_->transportSocketFactory());
 }
 
 std::unique_ptr<Http3ConnPoolImpl>

source/common/network/io_socket_handle_base_impl.cc

@@ -29,6 +29,8 @@ IoSocketHandleBaseImpl::~IoSocketHandleBaseImpl() {
 
 bool IoSocketHandleBaseImpl::isOpen() const { return SOCKET_VALID(fd_); }
 
+bool IoSocketHandleBaseImpl::isConnected() const { return is_connected_; }
+
 bool IoSocketHandleBaseImpl::supportsMmsg() const {
   return Api::OsSysCallsSingleton::get().supportsMmsg();
 }

source/common/network/io_socket_handle_base_impl.h

@@ -19,6 +19,7 @@ class IoSocketHandleBaseImpl : public IoHandle, protected Logger::Loggable<Logge
   // TODO(sbelair2)  To be removed when the fd is fully abstracted from clients.
   os_fd_t fdDoNotUse() const override { return fd_; }
   bool isOpen() const override;
+  bool isConnected() const override;
   bool supportsMmsg() const override;
   bool supportsUdpGro() const override;
   Api::SysCallIntResult setOption(int level, int optname, const void* optval,
@@ -38,6 +39,7 @@ class IoSocketHandleBaseImpl : public IoHandle, protected Logger::Loggable<Logge
   os_fd_t fd_;
   int socket_v6only_;
   const absl::optional<int> domain_;
+  bool is_connected_ = false;
 };
 
 } // namespace Network

source/common/network/io_socket_handle_impl.cc

@@ -523,7 +523,11 @@ Api::SysCallIntResult IoSocketHandleImpl::connect(Address::InstanceConstSharedPt
   }
 #endif
 
-  return Api::OsSysCallsSingleton::get().connect(fd_, sockaddr_to_use, sockaddr_len_to_use);
+  auto result = Api::OsSysCallsSingleton::get().connect(fd_, sockaddr_to_use, sockaddr_len_to_use);
+  if (result.return_value_ != -1) {
+    is_connected_ = true;
+  }
+  return result;
 }
 
 IoHandlePtr IoSocketHandleImpl::duplicate() {

source/common/network/utility.cc

@@ -513,16 +513,27 @@ Api::IoCallUint64Result Utility::writeToSocket(IoHandle& handle, Buffer::RawSlic
                                                const Address::Instance& peer_address) {
   Api::IoCallUint64Result send_result(
       /*rc=*/0, /*err=*/Api::IoError::none());
+
+  const bool is_connected = handle.isConnected();
   do {
-    send_result = handle.sendmsg(slices, num_slices, 0, local_ip, peer_address);
+    if (is_connected) {
+      // The socket is already connected, so the local and peer addresses should not be specified.
+      // Instead, a writev/send is called.
+      send_result = handle.writev(slices, num_slices);
+    } else {
+      // For non-connected sockets(), calling sendmsg with the peer address specified ensures the
+      // connection happens first.
+      send_result = handle.sendmsg(slices, num_slices, 0, local_ip, peer_address);
+    }
   } while (!send_result.ok() &&
            // Send again if interrupted.
            send_result.err_->getErrorCode() == Api::IoError::IoErrorCode::Interrupt);
 
   if (send_result.ok()) {
-    ENVOY_LOG_MISC(trace, "sendmsg bytes {}", send_result.return_value_);
+    ENVOY_LOG_MISC(trace, "{} bytes {}", is_connected ? "writev" : "sendmsg",
+                   send_result.return_value_);
   } else {
-    ENVOY_LOG_MISC(debug, "sendmsg failed with error code {}: {}",
+    ENVOY_LOG_MISC(debug, "{} failed with error code {}: {}", is_connected ? "writev" : "sendmsg",
                    static_cast<int>(send_result.err_->getErrorCode()),
                    send_result.err_->getErrorDetails());
   }

source/common/quic/envoy_quic_client_connection.cc

@@ -128,7 +128,10 @@ void EnvoyQuicClientConnection::setUpConnectionSocket(Network::ConnectionSocket&
       connection_socket.close();
     }
   }
-  if (!connection_socket.isOpen()) {
+  if (!connection_socket.isOpen() && connectionSocket().get() == &connection_socket) {
+    // Only close the connection if the connection socket is the current one. If it is a probing
+    // socket that isn't the current socket, do not close the connection upon failure, as the
+    // current socket is still usable.
     CloseConnection(quic::QUIC_CONNECTION_CANCELLED, "Fail to set up connection socket.",
                     quic::ConnectionCloseBehavior::SILENT_CLOSE);
   }
@@ -180,8 +183,8 @@ void EnvoyQuicClientConnection::probeWithNewPort(const quic::QuicSocketAddress&
 
   // The probing socket will have the same host but a different port.
   auto probing_socket =
-      createConnectionSocket(connectionSocket()->connectionInfoProvider().remoteAddress(),
-                             new_local_address, connectionSocket()->options(), prefer_gro_);
+      createConnectionSocket(quicAddressToEnvoyAddressInstance(peer_address), new_local_address,
+                             connectionSocket()->options(), prefer_gro_);
   setUpConnectionSocket(*probing_socket, delegate_);
   auto writer = std::make_unique<EnvoyQuicPacketWriter>(
       std::make_unique<Network::UdpDefaultWriter>(probing_socket->ioHandle()));

source/common/quic/envoy_quic_utils.cc

@@ -14,6 +14,18 @@
 namespace Envoy {
 namespace Quic {
 
+namespace {
+
+Network::Address::InstanceConstSharedPtr
+getLoopbackAddress(const Network::Address::IpVersion version) {
+  if (version == Network::Address::IpVersion::v6) {
+    return std::make_shared<Network::Address::Ipv6Instance>("::1");
+  }
+  return std::make_shared<Network::Address::Ipv4Instance>("127.0.0.1");
+}
+
+} // namespace
+
 // TODO(danzh): this is called on each write. Consider to return an address instance on the stack if
 // the heap allocation is too expensive.
 Network::Address::InstanceConstSharedPtr
@@ -168,16 +180,17 @@ createConnectionSocket(const Network::Address::InstanceConstSharedPtr& peer_addr
                        Network::Address::InstanceConstSharedPtr& local_addr,
                        const Network::ConnectionSocket::OptionsSharedPtr& options,
                        const bool prefer_gro) {
-  if (local_addr == nullptr) {
-    local_addr = Network::Utility::getLocalAddress(peer_addr->ip()->version());
-  }
+  ASSERT(peer_addr != nullptr);
   size_t max_addresses_cache_size =
       Runtime::runtimeFeatureEnabled(
           "envoy.reloadable_features.quic_upstream_socket_use_address_cache_for_read")
           ? 4u
           : 0u;
   auto connection_socket = std::make_unique<Network::ConnectionSocketImpl>(
-      Network::Socket::Type::Datagram, local_addr, peer_addr,
+      Network::Socket::Type::Datagram,
+      // Use the loopback address if `local_addr` is null, to pass in the socket interface used to
+      // create the IoHandle, without having to make the more expensive `getifaddrs` call.
+      local_addr ? local_addr : getLoopbackAddress(peer_addr->ip()->version()), peer_addr,
       Network::SocketCreationOptions{false, max_addresses_cache_size});
   connection_socket->setDetectedTransportProtocol("quic");
   if (!connection_socket->isOpen()) {
@@ -201,8 +214,18 @@ createConnectionSocket(const Network::Address::InstanceConstSharedPtr& peer_addr
     ENVOY_LOG_MISC(error, "Fail to apply pre-bind options");
     return connection_socket;
   }
-  connection_socket->bind(local_addr);
-  ASSERT(local_addr->ip());
+
+  if (local_addr != nullptr) {
+    connection_socket->bind(local_addr);
+    ASSERT(local_addr->ip());
+  }
+  if (auto result = connection_socket->connect(peer_addr); result.return_value_ != 0) {
+    connection_socket->close();
+    ENVOY_LOG_MISC(error, "Fail to connect socket: ({}) {}", result.errno_,
+                   errorDetails(result.errno_));
+    return connection_socket;
+  }
+
   local_addr = connection_socket->connectionInfoProvider().localAddress();
   if (!Network::Socket::applyOptions(connection_socket->options(), *connection_socket,
                                      envoy::config::core::v3::SocketOption::STATE_BOUND)) {

source/common/quic/quic_io_handle_wrapper.h

@@ -24,6 +24,7 @@ class QuicIoHandleWrapper : public Network::IoHandle {
     return Api::ioCallUint64ResultNoError();
   }
   bool isOpen() const override { return !closed_; }
+  bool isConnected() const override { return io_handle_.isConnected(); }
   Api::IoCallUint64Result readv(uint64_t max_length, Buffer::RawSlice* slices,
                                 uint64_t num_slice) override {
     if (closed_) {

source/extensions/io_socket/user_space/io_handle_impl.cc

@@ -90,6 +90,8 @@ Api::IoCallUint64Result IoHandleImpl::close() {
 
 bool IoHandleImpl::isOpen() const { return !closed_; }
 
+bool IoHandleImpl::isConnected() const { return false; }
+
 Api::IoCallUint64Result IoHandleImpl::readv(uint64_t max_length, Buffer::RawSlice* slices,
                                             uint64_t num_slice) {
   if (!isOpen()) {

source/extensions/io_socket/user_space/io_handle_impl.h

@@ -49,6 +49,7 @@ class IoHandleImpl final : public Network::IoHandle,
   }
   Api::IoCallUint64Result close() override;
   bool isOpen() const override;
+  bool isConnected() const override;
   Api::IoCallUint64Result readv(uint64_t max_length, Buffer::RawSlice* slices,
                                 uint64_t num_slice) override;
   Api::IoCallUint64Result read(Buffer::Instance& buffer,

test/common/quic/envoy_quic_client_session_test.cc

@@ -99,7 +99,7 @@ class EnvoyQuicClientSessionTest : public testing::TestWithParam<quic::ParsedQui
         self_addr_(Network::Utility::getAddressWithPort(
             *Network::Test::getCanonicalLoopbackAddress(TestEnvironment::getIpVersionsForTest()[0]),
             54321)),
-        peer_socket_(createConnectionSocket(self_addr_, peer_addr_, nullptr)),
+        peer_socket_(createUnconnectedSocket(self_addr_, peer_addr_)),
         crypto_config_(std::make_shared<quic::QuicCryptoClientConfig>(
             quic::test::crypto_test_utils::ProofVerifierForTesting())),
         quic_stat_names_(store_.symbolTable()),
@@ -159,6 +159,19 @@ class EnvoyQuicClientSessionTest : public testing::TestWithParam<quic::ParsedQui
     peer_socket_->close();
   }
 
+  Network::ConnectionSocketPtr
+  createUnconnectedSocket(const Network::Address::InstanceConstSharedPtr& peer_addr,
+                          Network::Address::InstanceConstSharedPtr& local_addr) {
+    auto connection_socket = std::make_unique<Network::ConnectionSocketImpl>(
+        Network::Socket::Type::Datagram, local_addr, peer_addr,
+        Network::SocketCreationOptions{false, 0});
+    connection_socket->setDetectedTransportProtocol("quic");
+    ASSERT(connection_socket->isOpen());
+    connection_socket->bind(local_addr);
+    local_addr = connection_socket->connectionInfoProvider().localAddress();
+    return connection_socket;
+  }
+
   EnvoyQuicClientStream& sendGetRequest(Http::ResponseDecoder& response_decoder,
                                         Http::StreamCallbacks& stream_callbacks) {
     auto& stream =

test/common/quic/envoy_quic_utils_test.cc

@@ -282,5 +282,39 @@ TEST(EnvoyQuicUtilsTest, QuicResetErrorToEnvoyResetReason) {
             Http::StreamResetReason::ConnectError);
 }
 
+TEST(EnvoyQuicUtilsTest, CreateConnectionSocket) {
+  Network::Address::InstanceConstSharedPtr local_addr =
+      std::make_shared<Network::Address::Ipv4Instance>("127.0.0.1");
+  Network::Address::InstanceConstSharedPtr peer_addr =
+      std::make_shared<Network::Address::Ipv4Instance>("127.0.0.1", 54321, nullptr);
+  auto connection_socket = createConnectionSocket(peer_addr, local_addr, nullptr);
+  EXPECT_TRUE(connection_socket->isOpen());
+  EXPECT_TRUE(connection_socket->ioHandle().isConnected());
+  connection_socket->close();
+
+  Network::Address::InstanceConstSharedPtr no_local_addr = nullptr;
+  connection_socket = createConnectionSocket(peer_addr, no_local_addr, nullptr);
+  EXPECT_TRUE(connection_socket->isOpen());
+  EXPECT_TRUE(connection_socket->ioHandle().isConnected());
+  EXPECT_EQ("127.0.0.1", no_local_addr->ip()->addressAsString());
+  connection_socket->close();
+
+  Network::Address::InstanceConstSharedPtr local_addr_v6 =
+      std::make_shared<Network::Address::Ipv6Instance>("::1");
+  Network::Address::InstanceConstSharedPtr peer_addr_v6 =
+      std::make_shared<Network::Address::Ipv6Instance>("::1", 54321, nullptr);
+  connection_socket = createConnectionSocket(peer_addr_v6, local_addr_v6, nullptr);
+  EXPECT_TRUE(connection_socket->isOpen());
+  EXPECT_TRUE(connection_socket->ioHandle().isConnected());
+  connection_socket->close();
+
+  Network::Address::InstanceConstSharedPtr no_local_addr_v6 = nullptr;
+  connection_socket = createConnectionSocket(peer_addr_v6, no_local_addr_v6, nullptr);
+  EXPECT_TRUE(connection_socket->isOpen());
+  EXPECT_TRUE(connection_socket->ioHandle().isConnected());
+  EXPECT_EQ("::1", no_local_addr_v6->ip()->addressAsString());
+  connection_socket->close();
+}
+
 } // namespace Quic
 } // namespace Envoy

test/integration/filters/test_socket_interface.cc

@@ -50,7 +50,7 @@ Api::IoCallUint64Result TestIoSocketHandle::writev(const Buffer::RawSlice* slice
   Address::InstanceConstSharedPtr dnat_peer_address;
   if (write_override_) {
     auto result = write_override_(this, slices, num_slice, dnat_peer_address);
-    ENVOY_BUG(dnat_peer_address == nullptr, "Only works for sendmsg, not writev");
+    peer_address_override_.reset();
     if (result.has_value()) {
       return std::move(result).value();
     }
@@ -81,9 +81,10 @@ IoHandlePtr TestIoSocketHandle::duplicate() {
 
 Api::SysCallIntResult TestIoSocketHandle::connect(Address::InstanceConstSharedPtr address) {
   if (connect_override_) {
-    auto result = connect_override_(this);
-    if (result.has_value())
+    auto result = connect_override_(this, address);
+    if (result.has_value()) {
       return Api::SysCallIntResult{-1, EINPROGRESS};
+    }
   }
 
   return Test::IoSocketHandlePlatformImpl::connect(address);

test/integration/filters/test_socket_interface.h

@@ -27,8 +27,8 @@ class TestIoSocketHandle : public Test::IoSocketHandlePlatformImpl {
       Address::InstanceConstSharedPtr& peer_address_override_out);
   using WriteOverrideProc = std::function<WriteOverrideType>;
   using ReadOverrideProc = std::function<void(RecvMsgOutput& output)>;
-  using ConnectOverrideProc =
-      std::function<absl::optional<Api::IoCallUint64Result>(TestIoSocketHandle* io_handle)>;
+  using ConnectOverrideProc = std::function<absl::optional<Api::IoCallUint64Result>(
+      TestIoSocketHandle* io_handle, Address::InstanceConstSharedPtr& peer_address_override_out)>;
 
   TestIoSocketHandle(ConnectOverrideProc connect_override_proc,
                      WriteOverrideProc write_override_proc, ReadOverrideProc read_override_proc,

test/integration/quic_http_integration_test.cc

@@ -2100,7 +2100,7 @@ TEST_P(QuicHttpIntegrationSPATest, UsesPreferredAddressDNAT) {
   }
   auto listener_port = lookupPort("http");
 
-  // Setup DNAT for 0.0.0.0:12345-->127.0.0.2:listener_port
+  // Setup DNAT for 1.2.3.4:12345-->127.0.0.2:listener_port
   socket_swap.write_matcher_->setDnat(
       Network::Utility::parseInternetAddressNoThrow("1.2.3.4", 12345),
       Network::Utility::parseInternetAddressNoThrow("127.0.0.2", listener_port));

test/integration/socket_interface_swap.cc

@@ -5,9 +5,12 @@ namespace Envoy {
 SocketInterfaceSwap::SocketInterfaceSwap(Network::Socket::Type socket_type)
     : write_matcher_(std::make_shared<IoHandleMatcher>(socket_type)),
       test_socket_interface_loader_(std::make_unique<Envoy::Network::TestSocketInterface>(
-          [write_matcher = write_matcher_](Envoy::Network::TestIoSocketHandle* io_handle)
+          [write_matcher =
+               write_matcher_](Envoy::Network::TestIoSocketHandle* io_handle,
+                               Network::Address::InstanceConstSharedPtr& peer_address_override_out)
               -> absl::optional<Api::IoCallUint64Result> {
-            Api::IoErrorPtr error_override = write_matcher->returnConnectOverride(io_handle);
+            Api::IoErrorPtr error_override =
+                write_matcher->returnConnectOverride(io_handle, peer_address_override_out);
             if (error_override) {
               return Api::IoCallUint64Result(0, std::move(error_override));
             }

test/integration/socket_interface_swap.h

@@ -39,13 +39,22 @@ class SocketInterfaceSwap {
       return Api::IoError::none();
     }
 
-    Api::IoErrorPtr returnConnectOverride(Envoy::Network::TestIoSocketHandle* io_handle) {
+    Api::IoErrorPtr
+    returnConnectOverride(Envoy::Network::TestIoSocketHandle* io_handle,
+                          Network::Address::InstanceConstSharedPtr& peer_address_override_out) {
       absl::MutexLock lock(&mutex_);
       if (block_connect_ && socket_type_ == io_handle->getSocketType() &&
           (io_handle->localAddress()->ip()->port() == src_port_ ||
            (dst_port_ && io_handle->peerAddress()->ip()->port() == dst_port_))) {
         return Network::IoSocketError::getIoSocketEagainError();
       }
+
+      if (orig_dnat_address_ != nullptr && peer_address_override_out != nullptr &&
+          *orig_dnat_address_ == *peer_address_override_out) {
+        ASSERT(translated_dnat_address_ != nullptr);
+        peer_address_override_out = translated_dnat_address_;
+      }
+
       return Api::IoError::none();
     }
 

test/mocks/network/io_handle.h

@@ -22,6 +22,7 @@ class MockIoHandle : public IoHandle {
   MOCK_METHOD(os_fd_t, fdDoNotUse, (), (const));
   MOCK_METHOD(Api::IoCallUint64Result, close, ());
   MOCK_METHOD(bool, isOpen, (), (const));
+  MOCK_METHOD(bool, isConnected, (), (const));
   MOCK_METHOD(Api::IoCallUint64Result, readv,
               (uint64_t max_length, Buffer::RawSlice* slices, uint64_t num_slice));
   MOCK_METHOD(Api::IoCallUint64Result, read,