CPP-510/CPP-183 - Try all contact points until one succeeds

Michael Penick · Michael Penick · commit 4cd6c8238f5b · 2018-10-31T14:19:06.000-04:00
* Also, connects to all contact points concurrently and cancels the
  others after the first successful connection.
* Fixed error message format (added a ":") in control connector.
* Start the connection settings with a default authentication provider.
diff --git a/cpp-driver/src/cluster_connector.cpp b/cpp-driver/src/cluster_connector.cpp
@@ -57,14 +57,16 @@ class RunCancelCluster : public Task {
 ClusterConnector::ClusterConnector(const ContactPointList& contact_points,
                                    ProtocolVersion protocol_version,
                                    const Callback& callback)
-  : contact_points_(contact_points)
+  : remaining_connector_count_(0)
+  , contact_points_(contact_points)
   , protocol_version_(protocol_version)
   , listener_(NULL)
   , event_loop_(NULL)
   , random_(NULL)
   , metrics_(NULL)
   , callback_(callback)
-  , error_code_(CLUSTER_OK) { }
+  , error_code_(CLUSTER_OK)
+  , ssl_error_code_(CASS_OK) { }
 
 void ClusterConnector::connect(EventLoop* event_loop) {
   event_loop_ = event_loop;
@@ -113,29 +115,42 @@ void ClusterConnector::internal_resolve_and_connect() {
   }
 
   if (!resolver_) {
-    contact_points_resolved_it_ = contact_points_resolved_.begin();
-    internal_connect();
+    internal_connect_all();
   }
 }
 
-void ClusterConnector::internal_connect() {
-  if (contact_points_resolved_it_ == contact_points_resolved_.end()) {
-    on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE, "Unable to connect to any contact points");
-    return;
-  }
-  connector_.reset(Memory::allocate<ControlConnector>(*contact_points_resolved_it_,
-                                                      protocol_version_,
-                                                      bind_callback(&ClusterConnector::on_connect, this)));
-  connector_
+void ClusterConnector::internal_connect(const Address& address, ProtocolVersion version) {
+  ControlConnector::Ptr connector(Memory::allocate<ControlConnector>(address,
+                                                                     version,
+                                                                     bind_callback(&ClusterConnector::on_connect, this)));
+  connectors_[address] = connector; // Keep track of the connectors so they can be canceled.
+  connector
       ->with_metrics(metrics_)
       ->with_settings(settings_.control_connection_settings)
       ->connect(event_loop_->loop());
 }
 
+void ClusterConnector::internal_connect_all() {
+  if (contact_points_resolved_.empty()) {
+    error_code_ = CLUSTER_ERROR_NO_HOSTS_AVAILABLE;
+    error_message_ = "Unable to connect to any contact points";
+    finish();
+    return;
+  }
+  remaining_connector_count_ = contact_points_resolved_.size();
+  for (AddressVec::const_iterator it = contact_points_resolved_.begin(),
+       end = contact_points_resolved_.end(); it != end; ++it) {
+    internal_connect(*it, protocol_version_);
+  }
+}
+
 void ClusterConnector::internal_cancel() {
   error_code_ = CLUSTER_CANCELED;
   if (resolver_) resolver_->cancel();
-  if (connector_) connector_->cancel();
+  for (ConnectorMap::iterator it = connectors_.begin(),
+       end = connectors_.end(); it != end; ++it) {
+    it->second->cancel();
+  }
   if (cluster_) cluster_->close();
 }
 
@@ -144,19 +159,23 @@ void ClusterConnector::finish() {
   if (cluster_) cluster_->close();
   // Explicitly release resources on the event loop thread.
   resolver_.reset();
-  connector_.reset();
+  connectors_.clear();
   cluster_.reset();
   dec_ref();
 }
 
+void ClusterConnector::maybe_finish() {
+  if (remaining_connector_count_ > 0 && --remaining_connector_count_ == 0) {
+    finish();
+  }
+}
+
 void ClusterConnector::on_error(ClusterConnector::ClusterError code,
                                 const String& message) {
   assert(code != CLUSTER_OK && "Notified error without an error");
-  if (error_code_ == CLUSTER_OK) { // Only perform this once
-    error_message_ = message;
-    error_code_ = code;
-    finish();
-  }
+  error_message_ = message;
+  error_code_ = code;
+  maybe_finish();
 }
 
 void ClusterConnector::on_resolve(MultiResolver* resolver) {
@@ -189,16 +208,24 @@ void ClusterConnector::on_resolve(MultiResolver* resolver) {
     }
   }
 
-  contact_points_resolved_it_ = contact_points_resolved_.begin();
-  internal_connect();
+  internal_connect_all();
 }
 
 void ClusterConnector::on_connect(ControlConnector* connector) {
-  if (is_canceled()) {
-    finish();
+  if (!connector->is_ok() && !connector->is_canceled()) {
+    LOG_ERROR("Unable to establish a control connection to host %s because of the following error: %s",
+              connector->address().to_string().c_str(),
+              connector->error_message().c_str());
+  }
+
+  // If the cluster object is successfully initialized or if the connector is
+  // canceled then attempt to finish the connection process.
+  if (cluster_ || is_canceled()) {
+    maybe_finish();
     return;
   }
 
+  // Otherwise, initialize the cluster and handle errors.
   if (connector->is_ok()) {
     const HostMap& hosts(connector->hosts());
     LoadBalancingPolicy::Vec policies;
@@ -209,9 +236,9 @@ void ClusterConnector::on_connect(ControlConnector* connector) {
       // metadata is corrupt or missing and the control connection process
       // would've probably failed before this happens.
       LOG_ERROR("Current control connection host %s not found in hosts metadata",
-                connector_->address().to_string().c_str());
-      ++contact_points_resolved_it_; // Move to the next contact point
-      internal_connect();
+                connector->address().to_string().c_str());
+      on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE,
+               "Control connection host is not found in hosts metadata");
       return;
     }
     Host::Ptr connected_host(host_it->second);
@@ -234,6 +261,11 @@ void ClusterConnector::on_connect(ControlConnector* connector) {
 
     ScopedPtr<QueryPlan> query_plan(default_policy->new_query_plan("", NULL, NULL));
     if (!query_plan->compute_next()) { // No hosts in the query plan
+      LOG_ERROR("Current control connection host %s has no hosts available in "
+                "it's query plan for the configured load balancing policy. If "
+                "using DC-aware check to see if the local datacenter is valid.",
+                connector->address().to_string().c_str());
+
       const char* message;
       if (dynamic_cast<DCAwarePolicy::DCAwareQueryPlan*>(query_plan.get()) != NULL) { // Check if DC-aware
         message = "No hosts available for the control connection using the " \
@@ -256,25 +288,38 @@ void ClusterConnector::on_connect(ControlConnector* connector) {
                                              default_policy,
                                              policies,
                                              settings_));
-    finish();
+
+    // Clear any connection errors and set the final negotiated protocol version.
+    error_code_ = CLUSTER_OK;
+    error_message_.clear();
+    protocol_version_ = connector->protocol_version();
+
+    // The cluster is initialized so the rest of the connectors can be canceled.
+    for (ConnectorMap::iterator it = connectors_.begin(),
+         end = connectors_.end(); it != end; ++it) {
+      if (it->first != connector->address()) { // Not the current connector.
+        it->second->cancel();
+      }
+    }
+
+    maybe_finish();
   } else if (connector->is_invalid_protocol()) {
-    if (!protocol_version_.attempt_lower_supported(contact_points_resolved_it_->to_string())) {
-      on_error(CLUSTER_ERROR_INVALID_PROTOCOL, "Unable to find supported protocol version");
+    ProtocolVersion lower_version(connector->protocol_version());
+    if (!lower_version.attempt_lower_supported(connector->address().to_string())) {
+      on_error(CLUSTER_ERROR_INVALID_PROTOCOL,
+               "Unable to find supported protocol version");
       return;
     }
-    internal_connect();
+    internal_connect(connector->address(), lower_version);
   } else if(connector->is_ssl_error()) {
+    ssl_error_code_ = connector->ssl_error_code();
     on_error(CLUSTER_ERROR_SSL_ERROR, connector->error_message());
   } else if(connector->is_auth_error()) {
     on_error(CLUSTER_ERROR_AUTH_ERROR, connector->error_message());
   } else {
-    LOG_ERROR("Unable to establish a control connection to host %s because of the following error: %s",
-              connector->address().to_string().c_str(),
-              connector->error_message().c_str());
-
-    // Possibly a recoverable error or timeout try the next host
-    ++contact_points_resolved_it_; // Move to the next contact point
-    internal_connect();
+    assert(!connector->is_canceled() &&
+           "The control connector should have an error and not be canceled");
+    on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE, connector->error_message());
   }
 }
 
diff --git a/cpp-driver/src/cluster_connector.hpp b/cpp-driver/src/cluster_connector.hpp
@@ -137,30 +137,41 @@ class ClusterConnector : public RefCounted<ClusterConnector> {
 
   ClusterError error_code() const { return error_code_; }
   const String& error_message() const { return error_message_; }
-  CassError ssl_error_code() { return connector_->ssl_error_code(); }
+  CassError ssl_error_code() { return ssl_error_code_; }
 
 private:
   friend class RunResolveAndConnectCluster;
   friend class RunCancelCluster;
 
 private:
   void internal_resolve_and_connect();
-  void internal_connect();
+  void internal_connect(const Address& address, ProtocolVersion version);
+  void internal_connect_all();
   void internal_cancel();
 
   void finish();
+  void maybe_finish();
 
   void on_error(ClusterError code, const String& message);
   void on_resolve(MultiResolver* resolver);
   void on_connect(ControlConnector* connector);
 
+private:
+  class ConnectorMap : public DenseHashMap<Address, ControlConnector::Ptr, AddressHash> {
+  public:
+    ConnectorMap() {
+      set_empty_key(Address::EMPTY_KEY);
+      set_deleted_key(Address::DELETED_KEY);
+    }
+  };
+
 private:
   Cluster::Ptr cluster_;
   MultiResolver::Ptr resolver_;
-  ControlConnector::Ptr connector_;
+  ConnectorMap connectors_;
+  size_t remaining_connector_count_;
   ContactPointList contact_points_;
   AddressVec contact_points_resolved_;
-  AddressVec::const_iterator contact_points_resolved_it_;
   ProtocolVersion protocol_version_;
   ClusterListener* listener_;
   EventLoop* event_loop_;
@@ -172,6 +183,7 @@ class ClusterConnector : public RefCounted<ClusterConnector> {
 
   ClusterError error_code_;
   String error_message_;
+  CassError ssl_error_code_;
 };
 
 } // namespace cass
diff --git a/cpp-driver/src/connector.cpp b/cpp-driver/src/connector.cpp
@@ -152,6 +152,7 @@ void StartupCallback::on_result_response(ResponseMessage* response) {
 
 ConnectionSettings::ConnectionSettings()
   : connect_timeout_ms(CASS_DEFAULT_CONNECT_TIMEOUT_MS)
+  , auth_provider(Memory::allocate<AuthProvider>())
   , idle_timeout_secs(CASS_DEFAULT_IDLE_TIMEOUT_SECS)
   , heartbeat_interval_secs(CASS_DEFAULT_HEARTBEAT_INTERVAL_SECS)
   , no_compact(CASS_DEFAULT_NO_COMPACT) { }
diff --git a/cpp-driver/src/connector.hpp b/cpp-driver/src/connector.hpp
@@ -164,6 +164,7 @@ class Connector
   uv_loop_t* loop() { return loop_; }
 
   const Address& address() const { return socket_connector_->address(); }
+  const ProtocolVersion protocol_version() const { return protocol_version_; }
 
 
   bool is_ok() const {
diff --git a/cpp-driver/src/control_connector.cpp b/cpp-driver/src/control_connector.cpp
@@ -38,7 +38,7 @@ class HostsConnectorRequestCallback : public ChainedRequestCallback {
 
   virtual void on_chain_error(CassError code, const String& message) {
     connector_->on_error(ControlConnector::CONTROL_CONNECTION_ERROR_HOSTS,
-                         "Error running host queries on control connection " + message);
+                         "Error running host queries on control connection: " + message);
   }
 
   virtual void on_chain_timeout() {
diff --git a/cpp-driver/src/control_connector.hpp b/cpp-driver/src/control_connector.hpp
@@ -201,6 +201,10 @@ class ControlConnector : public RefCounted<ControlConnector>
     return connector_->address();
   }
 
+  const ProtocolVersion protocol_version() const {
+    return connector_->protocol_version();
+  }
+
   bool is_ok() const { return error_code_ == CONTROL_CONNECTION_OK; }
   bool is_canceled() const { return error_code_ == CONTROL_CONNECTION_CANCELED; }
   bool is_invalid_protocol() const {

Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ class HostsConnectorRequestCallback : public ChainedRequestCallback {`
`38`	`38`
`39`	`39`	`virtual void on_chain_error(CassError code, const String& message) {`
`40`	`40`	`connector_->on_error(ControlConnector::CONTROL_CONNECTION_ERROR_HOSTS,`
`41`		`- "Error running host queries on control connection " + message);`
	`41`	`+ "Error running host queries on control connection: " + message);`
`42`	`42`	`}`
`43`	`43`
`44`	`44`	`virtual void on_chain_timeout() {`