CPP-794 Add domain name resolution to SocketConnector (#271)

Author: mpenick

cpp-driver/gtests/src/unit/tests/test_socket.cpp

@@ -20,9 +20,9 @@
 #include "socket_connector.hpp"
 #include "ssl.hpp"
 
-#define SSL_VERIFY_PEER_DNS_RELATIVE_HOSTNAME "cpp-driver.hostname"
-#define SSL_VERIFY_PEER_DNS_ABSOLUTE_HOSTNAME SSL_VERIFY_PEER_DNS_RELATIVE_HOSTNAME "."
-#define SSL_VERIFY_PEER_DNS_IP_ADDRESS "127.254.254.254"
+#define DNS_RELATIVE_HOSTNAME "cpp-driver.hostname"
+#define DNS_ABSOLUTE_HOSTNAME DNS_RELATIVE_HOSTNAME "."
+#define DNS_IP_ADDRESS "127.254.254.254"
 
 using mockssandra::internal::ClientConnection;
 using mockssandra::internal::ClientConnectionFactory;
@@ -152,6 +152,11 @@ class SocketUnitTest : public LoopTest {
     close();
   }
 
+  bool verify_dns() {
+    verify_dns_check(); // Verify address can be resolved
+    return HasFailure();
+  }
+
   static void on_socket_connected(SocketConnector* connector, String* result) {
     Socket::Ptr socket = connector->release_socket();
     if (connector->error_code() == SocketConnector::SOCKET_OK) {
@@ -190,15 +195,24 @@ class SocketUnitTest : public LoopTest {
   static void on_request(uv_getnameinfo_t* handle, int status, const char* hostname,
                          const char* service) {
     if (status) {
-      FAIL() << "Unable to Execute Test SocketUnitTest.SslVerifyIdentityDns: "
-             << "Add /etc/hosts entry " << SSL_VERIFY_PEER_DNS_IP_ADDRESS << "\t"
-             << SSL_VERIFY_PEER_DNS_ABSOLUTE_HOSTNAME;
-    } else if (String(hostname) != String(SSL_VERIFY_PEER_DNS_ABSOLUTE_HOSTNAME)) {
-      FAIL() << "Invalid /etc/hosts entry for: '" << hostname << "' != '"
-             << SSL_VERIFY_PEER_DNS_ABSOLUTE_HOSTNAME << "'";
+      FAIL() << "Unable to Execute Test: "
+             << "Add /etc/hosts entry " << DNS_IP_ADDRESS << "\t" << DNS_ABSOLUTE_HOSTNAME;
+    } else if (String(hostname) != String(DNS_ABSOLUTE_HOSTNAME)) {
+      FAIL() << "Invalid /etc/hosts entry for: '" << hostname << "' != '" << DNS_ABSOLUTE_HOSTNAME
+             << "'";
     }
   }
 
+private:
+  void verify_dns_check() {
+    Address verify_entry(DNS_IP_ADDRESS, 8888);
+    uv_getnameinfo_t request;
+    Address::SocketStorage storage;
+    ASSERT_EQ(0,
+              uv_getnameinfo(loop(), &request, on_request, verify_entry.to_sockaddr(&storage), 0));
+    uv_run(loop(), UV_RUN_DEFAULT);
+  }
+
 private:
   mockssandra::SimpleEchoServer server_;
 };
@@ -217,6 +231,22 @@ TEST_F(SocketUnitTest, Simple) {
   EXPECT_EQ(result, "The socket is successfully connected and wrote data - Closed");
 }
 
+TEST_F(SocketUnitTest, SimpleDns) {
+  if (!verify_dns()) return;
+
+  listen(Address(DNS_IP_ADDRESS, 8888));
+
+  String result;
+  SocketConnector::Ptr connector(new SocketConnector(Address(DNS_RELATIVE_HOSTNAME, 8888),
+                                                     bind_callback(on_socket_connected, &result)));
+
+  connector->connect(loop());
+
+  uv_run(loop(), UV_RUN_DEFAULT);
+
+  EXPECT_EQ(result, "The socket is successfully connected and wrote data - Closed");
+}
+
 TEST_F(SocketUnitTest, Ssl) {
   SocketSettings settings(use_ssl());
 
@@ -355,25 +385,16 @@ TEST_F(SocketUnitTest, SslVerifyIdentity) {
 }
 
 TEST_F(SocketUnitTest, SslVerifyIdentityDns) {
-  // Verify address can be resolved
-  Address verify_entry(SSL_VERIFY_PEER_DNS_IP_ADDRESS, 8888);
-  uv_getnameinfo_t request;
-  Address::SocketStorage storage;
-  ASSERT_EQ(0, uv_getnameinfo(loop(), &request, on_request, verify_entry.to_sockaddr(&storage), 0));
-  uv_run(loop(), UV_RUN_DEFAULT);
-  if (this->HasFailure()) { // Make test fail due to DNS not configured
-    return;
-  }
+  if (!verify_dns()) return;
 
-  SocketSettings settings(use_ssl(SSL_VERIFY_PEER_DNS_RELATIVE_HOSTNAME));
+  SocketSettings settings(use_ssl(DNS_RELATIVE_HOSTNAME));
 
-  listen(Address(SSL_VERIFY_PEER_DNS_IP_ADDRESS,
-                 8888)); // Ensure the echo server is listening on the correct address
+  listen(Address(DNS_IP_ADDRESS, 8888));
 
   settings.ssl_context->set_verify_flags(CASS_SSL_VERIFY_PEER_IDENTITY_DNS);
 
   String result;
-  SocketConnector::Ptr connector(new SocketConnector(Address(SSL_VERIFY_PEER_DNS_IP_ADDRESS, 8888),
+  SocketConnector::Ptr connector(new SocketConnector(Address(DNS_RELATIVE_HOSTNAME, 8888),
                                                      bind_callback(on_socket_connected, &result)));
 
   connector->with_settings(settings)->connect(loop());

cpp-driver/src/connection.cpp

@@ -116,8 +116,6 @@ Connection::Connection(const Socket::Ptr& socket, const Host::Ptr& host,
     , heartbeat_interval_secs_(heartbeat_interval_secs)
     , heartbeat_outstanding_(false) {
   inc_ref(); // For the event loop
-
-  assert(host_->address() == socket_->address() && "Host doesn't match socket address");
   host_->increment_connection_count();
 }
 

cpp-driver/src/resolver.hpp

@@ -55,7 +55,7 @@ class Resolver : public RefCounted<Resolver> {
     req_.data = this;
   }
 
-  ~Resolver() {}
+  uv_loop_t* loop() { return req_.loop; }
 
   const String& hostname() { return hostname_; }
   int port() { return port_; }

cpp-driver/src/socket_connector.cpp

@@ -22,10 +22,25 @@
 #define SSL_HANDSHAKE_MAX_BUFFER_SIZE (16 * 1024 + 5)
 
 using namespace datastax;
+using namespace datastax::internal;
 using namespace datastax::internal::core;
 
 namespace datastax { namespace internal { namespace core {
 
+namespace {
+
+// Used for debugging resolved addresses.
+String to_string(const AddressVec& addresses) {
+  String result;
+  for (AddressVec::const_iterator it = addresses.begin(), end = addresses.end(); it != end; ++it) {
+    if (!result.empty()) result.append(", ");
+    result.append(it->to_string());
+  }
+  return result;
+}
+
+} // namespace
+
 /**
  * A socket handler that handles the SSL handshake process.
  */
@@ -85,6 +100,8 @@ SocketSettings::SocketSettings(const Config& config)
     , max_reusable_write_objects(config.max_reusable_write_objects())
     , local_address(config.local_address()) {}
 
+Atomic<size_t> SocketConnector::resolved_address_offset_(0);
+
 SocketConnector::SocketConnector(const Address& address, const Callback& callback)
     : address_(address)
     , callback_(callback)
@@ -99,23 +116,34 @@ SocketConnector* SocketConnector::with_settings(const SocketSettings& settings)
 void SocketConnector::connect(uv_loop_t* loop) {
   inc_ref(); // For the event loop
 
-  if (settings_.hostname_resolution_enabled) {
-    // Run hostname resolution then connect.
-    resolver_.reset(new NameResolver(address_, bind_callback(&SocketConnector::on_resolve, this)));
+  if (!address_.is_resolved()) { // Address not resolved
+    hostname_ = address_.to_string();
+
+    resolver_.reset(new Resolver(hostname_, address_.port(),
+                                 bind_callback(&SocketConnector::on_resolve, this)));
     resolver_->resolve(loop, settings_.resolve_timeout_ms);
   } else {
-    // Postpone the connection process until after this method ends because it
-    // can call the callback (via on_error() when when the socket fails to
-    // init/bind) and destroy its parent.
-    no_resolve_timer_.start(loop,
-                            0, // Run connect immediately after.
-                            bind_callback(&SocketConnector::on_no_resolve, this));
+    resolved_address_ = address_;
+
+    if (settings_.hostname_resolution_enabled) { // Run hostname resolution then connect.
+      name_resolver_.reset(
+          new NameResolver(address_, bind_callback(&SocketConnector::on_name_resolve, this)));
+      name_resolver_->resolve(loop, settings_.resolve_timeout_ms);
+    } else {
+      // Postpone the connection process until after this method ends because it
+      // can call the callback (via on_error() when when the socket fails to
+      // init/bind) and destroy its parent.
+      no_resolve_timer_.start(loop,
+                              0, // Run connect immediately after.
+                              bind_callback(&SocketConnector::on_no_resolve, this));
+    }
   }
 }
 
 void SocketConnector::cancel() {
   error_code_ = SOCKET_CANCELED;
   if (resolver_) resolver_->cancel();
+  if (name_resolver_) name_resolver_->cancel();
   if (connector_) connector_->cancel();
   if (socket_) socket_->close();
 }
@@ -127,7 +155,7 @@ Socket::Ptr SocketConnector::release_socket() {
 }
 
 void SocketConnector::internal_connect(uv_loop_t* loop) {
-  Socket::Ptr socket(new Socket(address_, settings_.max_reusable_write_objects));
+  Socket::Ptr socket(new Socket(resolved_address_, settings_.max_reusable_write_objects));
 
   if (uv_tcp_init(loop, socket->handle()) != 0) {
     on_error(SOCKET_ERROR_INIT, "Unable to initialize TCP object");
@@ -159,11 +187,11 @@ void SocketConnector::internal_connect(uv_loop_t* loop) {
   }
 
   if (settings_.ssl_context) {
-    ssl_session_.reset(
-        settings_.ssl_context->create_session(address_, hostname_, address_.server_name()));
+    ssl_session_.reset(settings_.ssl_context->create_session(resolved_address_, hostname_,
+                                                             address_.server_name()));
   }
 
-  connector_.reset(new TcpConnector(address_));
+  connector_.reset(new TcpConnector(resolved_address_));
   connector_->connect(socket_->handle(), bind_callback(&SocketConnector::on_connect, this));
 }
 
@@ -253,7 +281,25 @@ void SocketConnector::on_connect(TcpConnector* tcp_connector) {
   }
 }
 
-void SocketConnector::on_resolve(NameResolver* resolver) {
+void SocketConnector::on_resolve(Resolver* resolver) {
+  if (resolver->is_success()) {
+    const AddressVec& addresses(resolver->addresses());
+    LOG_DEBUG("Resolved the addresses %s for hostname %s", to_string(addresses).c_str(),
+              hostname_.c_str());
+    resolved_address_ =
+        addresses[resolved_address_offset_.fetch_add(MEMORY_ORDER_RELAXED) % addresses.size()];
+    internal_connect(resolver->loop());
+  } else if (is_canceled() || resolver->is_canceled()) {
+    finish();
+  } else if (resolver->is_timed_out()) {
+    on_error(SOCKET_ERROR_RESOLVE_TIMEOUT, "Timed out attempting to resolve hostname");
+  } else {
+    on_error(SOCKET_ERROR_RESOLVE,
+             "Unable to resolve hostname '" + String(uv_strerror(resolver->uv_status())) + "'");
+  }
+}
+
+void SocketConnector::on_name_resolve(NameResolver* resolver) {
   if (resolver->is_success()) {
     LOG_DEBUG("Resolved the hostname %s for address %s", resolver->hostname().c_str(),
               resolver->address().to_string().c_str());

cpp-driver/src/socket_connector.hpp

@@ -17,8 +17,10 @@
 #ifndef DATASTAX_INTERNAL_SOCKET_CONNECTOR_HPP
 #define DATASTAX_INTERNAL_SOCKET_CONNECTOR_HPP
 
+#include "atomic.hpp"
 #include "callback.hpp"
 #include "name_resolver.hpp"
+#include "resolver.hpp"
 #include "socket.hpp"
 #include "tcp_connector.hpp"
 
@@ -141,17 +143,23 @@ class SocketConnector : public RefCounted<SocketConnector> {
 
   void on_error(SocketError code, const String& message);
   void on_connect(TcpConnector* tcp_connecter);
-  void on_resolve(NameResolver* resolver);
+  void on_resolve(Resolver* resolver);
+  void on_name_resolve(NameResolver* resolver);
   void on_no_resolve(Timer* timer);
 
+private:
+  static Atomic<size_t> resolved_address_offset_;
+
 private:
   Address address_;
+  Address resolved_address_;
   String hostname_;
   Callback callback_;
 
   Socket::Ptr socket_;
   TcpConnector::Ptr connector_;
-  NameResolver::Ptr resolver_;
+  Resolver::Ptr resolver_;
+  NameResolver::Ptr name_resolver_;
   Timer no_resolve_timer_;
 
   SocketError error_code_;