CPP-687 - Fix recovery of remote DC host (#427)

The DC-aware load balancing policy depends on an internal host map to determine host distance. The problem was the host distance was being checked to determine whether a node should be added or marked as up.  This fix always adds and marks hosts up even if they're ignored. The distance is now only used to determine if a host should have a connection pool.

Author: mpenick

gtests/src/unit/mockssandra.cpp

@@ -1807,15 +1807,23 @@ cass::String SchemaChangeEvent::encode(SchemaChangeEvent::Target target, SchemaC
 
 void Cluster::init(AddressGenerator& generator,
                    ClientConnectionFactory& factory,
-                   size_t num_nodes) {
+                   size_t num_nodes_dc1,
+                   size_t num_nodes_dc2) {
   MT19937_64 token_rng;
-  for (size_t i = 0; i < num_nodes; ++i) {
+  for (size_t i = 0; i < num_nodes_dc1; ++i) {
     Address address(generator.next());
     Server server(Host(address, "dc1", "rack1", token_rng),
                   internal::ServerConnection::Ptr(
                     Memory::allocate<internal::ServerConnection>(address, factory)));
     servers_.push_back(server);
   }
+  for (size_t i = 0; i < num_nodes_dc2; ++i) {
+    Address address(generator.next());
+    Server server(Host(address, "dc2", "rack1", token_rng),
+                  internal::ServerConnection::Ptr(
+                    Memory::allocate<internal::ServerConnection>(address, factory)));
+    servers_.push_back(server);
+  }
 }
 
 Cluster::~Cluster() {

gtests/src/unit/mockssandra.hpp

@@ -1119,7 +1119,8 @@ class Cluster {
 protected:
   void init(AddressGenerator& generator,
             ClientConnectionFactory& factory,
-            size_t num_nodes);
+            size_t num_nodes_dc1,
+            size_t num_nodes_dc2);
 
 public:
   ~Cluster();
@@ -1187,10 +1188,11 @@ class AuthRequestHandlerBuilder : public SimpleRequestHandlerBuilder {
 class SimpleCluster : public Cluster {
 public:
   SimpleCluster(const RequestHandler* request_handler,
-                size_t num_nodes = 1)
+                size_t num_nodes_dc1 = 1,
+                size_t num_nodes_dc2 = 0)
     : factory_(request_handler, this)
     , event_loop_group_(1) {
-    init(generator_, factory_, num_nodes);
+    init(generator_, factory_, num_nodes_dc1, num_nodes_dc2);
   }
 
   ~SimpleCluster() {

gtests/src/unit/tests/test_cluster.cpp

@@ -189,6 +189,32 @@ class ClusterUnitTest : public EventLoopTest {
     OutagePlan* outage_plan_;
   };
 
+  class RecoverClusterListener : public UpDownListener {
+  public:
+    typedef SharedRefPtr<RecoverClusterListener> Ptr;
+
+    RecoverClusterListener(const Future::Ptr& close_future,
+                           const Future::Ptr& up_future,
+                           const Future::Ptr& recover_future)
+      : UpDownListener(close_future, up_future, Future::Ptr())
+      , recover_future_(recover_future) { }
+
+    const HostVec& connected_hosts() const {
+      return connected_hosts_;
+    }
+
+    virtual void on_reconnect(Cluster* cluster) {
+      connected_hosts_.push_back(cluster->connected_host());
+      if (connected_hosts_.size() > 1 && recover_future_) {
+        recover_future_->set();
+      }
+    }
+
+  private:
+    HostVec connected_hosts_;
+    Future::Ptr recover_future_;
+  };
+
   static void on_connection_connected(ClusterConnector* connector, Future* future) {
     if (connector->is_ok()) {
       future->set();
@@ -208,7 +234,7 @@ class ClusterUnitTest : public EventLoopTest {
           break;
         case ClusterConnector::CLUSTER_ERROR_NO_HOSTS_AVAILABLE:
           future->set_error(CASS_ERROR_LIB_NO_HOSTS_AVAILABLE,
-                            "Unable to connect to any contact points");
+                            connector->error_message());
           break;
         case ClusterConnector::CLUSTER_CANCELED:
           future->set_error(CASS_ERROR_LIB_UNABLE_TO_CONNECT,
@@ -668,3 +694,88 @@ TEST_F(ClusterUnitTest, InvalidSsl) {
   ASSERT_TRUE(connect_future->error());
   EXPECT_EQ(CASS_ERROR_SSL_INVALID_PEER_CERT, connect_future->error()->code);
 }
+
+TEST_F(ClusterUnitTest, DCAwareRecoverOnRemoteHost) {
+  mockssandra::SimpleCluster cluster(simple(), 1, 1); // 2 DCs with a single node each
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  Address local_address("127.0.0.1", 9042);
+  Address remote_address("127.0.0.2", 9042);
+
+  ContactPointList contact_points;
+  contact_points.push_back(local_address.to_string());
+
+  Future::Ptr close_future(Memory::allocate<Future>());
+  Future::Ptr connect_future(Memory::allocate<Future>());
+  ClusterConnector::Ptr connector(Memory::allocate<ClusterConnector>(contact_points,
+                                                                     PROTOCOL_VERSION,
+                                                                     bind_callback(on_connection_reconnect, connect_future.get())));
+
+  Future::Ptr up_future(Memory::allocate<Future>());
+  Future::Ptr recover_future(Memory::allocate<Future>());
+  RecoverClusterListener::Ptr listener(
+        Memory::allocate<RecoverClusterListener>(close_future, up_future, recover_future));
+
+  ClusterSettings settings;
+  settings.load_balancing_policy.reset(Memory::allocate<DCAwarePolicy>("dc1", 1, false)); // Allow connection to a single remote host
+  settings.load_balancing_policies.clear();
+  settings.load_balancing_policies.push_back(settings.load_balancing_policy);
+  settings.reconnect_timeout_ms = 1; // Reconnect immediately
+  settings.control_connection_settings.connection_settings.connect_timeout_ms = 200; // Give enough time for the connection to complete
+
+  connector
+      ->with_settings(settings)
+      ->with_listener(listener.get())
+      ->connect(event_loop());
+
+  ASSERT_TRUE(connect_future->wait_for(WAIT_FOR_TIME));
+  EXPECT_FALSE(connect_future->error());
+
+  // Notify every host as down
+  connect_future->cluster()->notify_down(local_address);
+  connect_future->cluster()->notify_down(remote_address);
+
+  // Notify the remote host as up
+  connect_future->cluster()->notify_up(remote_address);
+
+  // Verify that the remote host was marked as up
+  ASSERT_TRUE(up_future->wait_for(WAIT_FOR_TIME));
+  EXPECT_EQ(remote_address, listener->address());
+
+  cluster.stop(1); // Stop local node to verify that remote host is tried for reconnection.
+  ASSERT_TRUE(recover_future->wait_for(WAIT_FOR_TIME));
+
+  connect_future->cluster()->close();
+  ASSERT_TRUE(close_future->wait_for(WAIT_FOR_TIME));
+
+  ASSERT_EQ(listener->connected_hosts().size(), 2u);
+  EXPECT_EQ(listener->connected_hosts()[0]->address(), Address("127.0.0.1", PORT));
+  EXPECT_EQ(listener->connected_hosts()[1]->address(), Address("127.0.0.2", PORT)); // Connected to remote host.
+}
+
+TEST_F(ClusterUnitTest, InvalidDC) {
+  mockssandra::SimpleCluster cluster(simple());
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  ContactPointList contact_points;
+  contact_points.push_back("127.0.0.1");
+
+  Future::Ptr connect_future(Memory::allocate<Future>());
+  ClusterConnector::Ptr connector(Memory::allocate<ClusterConnector>(contact_points,
+                                                                     PROTOCOL_VERSION,
+                                                                     bind_callback(on_connection_connected, connect_future.get())));
+
+  ClusterSettings settings;
+  settings.load_balancing_policy.reset(Memory::allocate<DCAwarePolicy>("invalid_dc", 0, false)); // Invalid DC and not using remote hosts
+  settings.load_balancing_policies.clear();
+  settings.load_balancing_policies.push_back(settings.load_balancing_policy);
+
+  connector
+      ->with_settings(settings)
+      ->connect(event_loop());
+
+  ASSERT_TRUE(connect_future->wait_for(WAIT_FOR_TIME));
+  ASSERT_TRUE(connect_future->error());
+  EXPECT_EQ(CASS_ERROR_LIB_NO_HOSTS_AVAILABLE, connect_future->error()->code);
+  EXPECT_TRUE(connect_future->error()->message.find("Check to see if the configured local datacenter is valid") != String::npos);
+}

src/cluster.cpp

@@ -213,19 +213,28 @@ void Cluster::prepared(const String& id,
   prepared_metadata_.set(id, entry);
 }
 
+HostMap Cluster::available_hosts() const {
+  HostMap available;
+  for (HostMap::const_iterator it = hosts_.begin(),
+       end = hosts_.end(); it != end; ++it) {
+    if (!is_host_ignored(it->second)) {
+      available[it->first] = it->second;
+    }
+  }
+  return available;
+}
+
 void Cluster::update_hosts(const HostMap& hosts) {
   // Update the hosts and properly notify the listener
   HostMap existing(hosts_);
 
   for (HostMap::const_iterator it = hosts.begin(),
        end = hosts.end(); it != end; ++it) {
-    if (!is_host_ignored(it->second)) {
-      HostMap::iterator find_it = existing.find(it->first);
-      if (find_it != existing.end()) {
-        existing.erase(find_it); // Already exists mark as visited
-      } else {
-        notify_add(it->second); // A new host has been added
-      }
+    HostMap::iterator find_it = existing.find(it->first);
+    if (find_it != existing.end()) {
+      existing.erase(find_it); // Already exists mark as visited
+    } else {
+      notify_add(it->second); // A new host has been added
     }
   }
 
@@ -305,6 +314,10 @@ void Cluster::update_token_map(const HostMap& hosts,
   }
 }
 
+// All hosts from the cluster are included in the host map and in the load
+// balancing policies (LBP) so that LBPs return the correct host distance (esp.
+// important for DC-aware). This method prevents connection pools from being
+// created to ignored hosts.
 bool Cluster::is_host_ignored(const Host::Ptr& host) const {
   return cass::is_host_ignored(load_balancing_policies_, host);
 }
@@ -413,10 +426,6 @@ void Cluster::internal_notify_up(const Address& address, const Host::Ptr& refres
 
   Host::Ptr host(it->second);
 
-  if (is_host_ignored(host)) {
-    return; // Ignore host
-  }
-
   if (refreshed){
     if (token_map_) {
       token_map_ = token_map_->copy();
@@ -437,6 +446,10 @@ void Cluster::internal_notify_up(const Address& address, const Host::Ptr& refres
     (*it)->on_up(host);
   }
 
+  if (is_host_ignored(host)) {
+    return; // Ignore host
+  }
+
   if (!prepare_host(host,
                     bind_callback(&Cluster::on_prepare_host_up, this))) {
     notify_up_after_prepare(host);
@@ -487,16 +500,16 @@ void Cluster::notify_add(const Host::Ptr& host) {
     listener_->on_remove(host_it->second);
   }
 
-  if (is_host_ignored(host)) {
-    return; // Ignore host
-  }
-
   hosts_[host->address()] = host;
   for (LoadBalancingPolicy::Vec::const_iterator it = load_balancing_policies_.begin(),
        end = load_balancing_policies_.end(); it != end; ++it) {
     (*it)->on_add(host);
   }
 
+  if (is_host_ignored(host)) {
+    return; // Ignore host
+  }
+
   if (!prepare_host(host,
                     bind_callback(&Cluster::on_prepare_host_add, this))) {
     notify_add_after_prepare(host);

src/cluster.hpp

@@ -255,10 +255,17 @@ class Cluster : public RefCounted<Cluster>
   void prepared(const String& id,
                 const PreparedMetadata::Entry::Ptr& entry);
 
+  /**
+   * Get available hosts (determined by host distance). This filters out ignored
+   * hosts (*NOT* thread-safe).
+   *
+   * @return A mapping of available hosts.
+   */
+  HostMap available_hosts() const;
+
 public:
   int protocol_version() const { return connection_->protocol_version(); }
   const Host::Ptr& connected_host() const { return connected_host_; }
-  const HostMap& hosts() const { return hosts_; }
   const TokenMap::Ptr& token_map() const { return token_map_; }
 
 private:

src/cluster_connector.cpp

@@ -15,6 +15,7 @@
 */
 
 #include "cluster_connector.hpp"
+#include "dc_aware_policy.hpp"
 #include "random.hpp"
 #include "round_robin_policy.hpp"
 
@@ -230,28 +231,26 @@ void ClusterConnector::on_connect(ControlConnector* connector) {
       policy->register_handles(event_loop_->loop());
     }
 
-    // Gather the available hosts for the load balancing policies
-    HostMap available_hosts;
-    for (HostMap::const_iterator it = hosts.begin(), end = hosts.end();
-         it != end; ++it) {
-      if (!is_host_ignored(policies, it->second)) {
-        available_hosts[it->first] = it->second;
+    ScopedPtr<QueryPlan> query_plan(default_policy->new_query_plan("", NULL, NULL));
+    if (!query_plan->compute_next()) { // No hosts in the query plan
+      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 " \
+                  "DC-aware load balancing policy. " \
+                  "Check to see if the configured local datacenter is valid";
+      } else {
+        message = "No hosts available for the control connection using the " \
+                  "configured load balancing policy";
       }
-    }
-
-    if (available_hosts.empty()) {
-      //TODO(fero): Check for DC aware policy to give more informative message (e.g. invalid DC)
-      on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE,
-               "No hosts available for connection using the current load " \
-               "balancing policy(s)");
+      on_error(CLUSTER_ERROR_NO_HOSTS_AVAILABLE, message);
       return;
     }
 
     cluster_.reset(Memory::allocate<Cluster>(connector->release_connection(),
                                              listener_,
                                              event_loop_,
                                              connected_host,
-                                             available_hosts,
+                                             hosts,
                                              connector->schema(),
                                              default_policy,
                                              policies,

src/dc_aware_policy.hpp

@@ -93,6 +93,7 @@ class DCAwarePolicy : public LoadBalancingPolicy {
   const CopyOnWriteHostVec& get_local_dc_hosts() const;
   void get_remote_dcs(PerDCHostMap::KeySet* remote_dcs) const;
 
+public:
   class DCAwareQueryPlan : public QueryPlan {
   public:
     DCAwareQueryPlan(const DCAwarePolicy* policy,
@@ -111,6 +112,7 @@ class DCAwarePolicy : public LoadBalancingPolicy {
     size_t index_;
   };
 
+private:
   String local_dc_;
   size_t used_hosts_per_remote_dc_;
   bool skip_remote_dcs_for_local_cl_;

src/session_base.cpp

@@ -165,7 +165,7 @@ void SessionBase::on_initialize(ClusterConnector* connector) {
     cluster_ = connector->release_cluster();
     on_connect(cluster_->connected_host(),
                cluster_->protocol_version(),
-               cluster_->hosts(),
+               cluster_->available_hosts(),
                cluster_->token_map());
   } else {
     assert(!connector->is_canceled() && "Cluster connection process canceled");