CPP-402 - test: Adding unit/integration tests for host listener callback API

* Updates to mockssandra
  * Added topology change events for new and removed nodes
* Added integration and unit tests for external host listener
* Moved `wait_for_logger` into the base integration class
  * Fixed TODO and broken test in control connection
* Removing empty cluster manager test

Author: Michael Fero

cpp-driver/gtests/src/integration/integration.cpp

@@ -26,6 +26,8 @@
 #define SIMPLE_KEYSPACE_FORMAT "CREATE KEYSPACE IF NOT EXISTS %s WITH replication = %s"
 #define REPLICATION_STRATEGY "{ 'class': %s }"
 #define SELECT_SERVER_VERSION "SELECT release_version FROM system.local"
+#define LOGGER_MAXIMUM_WAIT_TIME_MS 10000u
+#define LOGGER_WAIT_FOR_NAP_MS 100
 
 // Initialize static variables
 bool Integration::skipped_message_displayed_ = false;
@@ -434,3 +436,12 @@ void Integration::maybe_shrink_name(std::string& name)
     name = name.substr(0, ENTITY_MAXIMUM_LENGTH - id.size()) + id;
   }
 }
+
+bool Integration::wait_for_logger(size_t expected_count) {
+  start_timer();
+  while (elapsed_time() < LOGGER_MAXIMUM_WAIT_TIME_MS
+         && logger_.count() < expected_count) {
+    msleep(LOGGER_WAIT_FOR_NAP_MS);
+  }
+  return logger_.count() >= expected_count;
+}

cpp-driver/gtests/src/integration/integration.hpp

@@ -595,6 +595,16 @@ class Integration : public testing::Test {
    */
   void maybe_shrink_name(std::string& name);
 
+  /**
+   * Wait for the logger count to reach expected count
+   *
+   * NOTE: This may wait up to LOGGER_MAXIMUM_WAIT_TIME_MS
+   *
+   * @param expected_count Expected logger count
+   * @return True if expected count is equal to logger count; false otherwise
+   */
+  bool wait_for_logger(size_t expected_count);
+
 private:
   /**
    * Keyspace creation query (generated via SetUp)

cpp-driver/gtests/src/integration/objects/cluster.hpp

@@ -189,6 +189,21 @@ class Cluster : public Object<CassCluster, cass_cluster_free> {
     return *this;
   }
 
+  /**
+   * Sets a callback for handling host state changes in the cluster
+   *
+   * @param callback Callback to use for cluster host state changes
+   * @param data User data supplied to the callback (default: NULL)
+   * @return Cluster object
+   */
+  Cluster& with_host_listener_callback(CassHostListenerCallback callback,
+                                       void* data = NULL) {
+    EXPECT_EQ(CASS_OK, cass_cluster_set_host_listener_callback(get(),
+                                                               callback,
+                                                               data));
+    return *this;
+  }
+
   /**
    * Enable/Disable the use of hostname resolution
    *

cpp-driver/gtests/src/integration/tests/test_cluster_manager.cpp

@@ -16,8 +16,6 @@
 
 #include "integration.hpp"
 
-#define LOGGER_WAIT_FOR_NAP 100
-
 /**
  * Control connection integration tests; single node cluster
  */
@@ -84,22 +82,6 @@ class ControlConnectionTests : public Integration {
       logger_.add_critera(prefix + node_ip_address.str() + suffix);
     }
   }
-
-  /**
-   * Wait for the logger count to reach expected count
-   *
-   * NOTE: This may wait up to 10s
-   *
-   * @param expected_count Expected logger count
-   * @return True if expected count is equal to logger count; false otherwise
-   */
-  bool wait_for_logger(size_t expected_count) {
-    start_timer();
-    while (elapsed_time() < 10000u && logger_.count() < expected_count) {
-      msleep(LOGGER_WAIT_FOR_NAP);
-    }
-    return logger_.count() >= expected_count;
-  }
 };
 
 /**
@@ -353,19 +335,20 @@ CASSANDRA_INTEGRATION_TEST_F(ControlConnectionTests, TopologyChange) {
   Session session = cluster.connect();
 
   // Bootstrap a second node and ensure all hosts are actively used
-  unsigned int node_2 = ccm_->bootstrap_node(); // Triggers a `NEW_NODE` event
-  msleep(3000); //TODO: Remove static sleep and check driver logs for reduced wait
+  logger_.add_critera("New node " + (ccm_->get_ip_prefix() + "2") + " added");
+  EXPECT_EQ(2, ccm_->bootstrap_node()); // Triggers a `NEW_NODE` event
+  EXPECT_TRUE(wait_for_logger(1));
   std::set<unsigned short> expected_nodes;
   expected_nodes.insert(1);
-  expected_nodes.insert(node_2);
+  expected_nodes.insert(2);
   check_hosts(session, expected_nodes);
 
   /*
    * Decommission the bootstrapped node and ensure only the first node is
    * actively used
    */
-  force_decommission_node(node_2); // Triggers a `REMOVE_NODE` event
-  expected_nodes.erase(node_2);
+  force_decommission_node(2); // Triggers a `REMOVE_NODE` event
+  expected_nodes.erase(2);
   check_hosts(session, expected_nodes);
 }
 
@@ -474,7 +457,9 @@ CASSANDRA_INTEGRATION_TEST_F(ControlConnectionThreeNodeClusterTests,
    * policy, initial invalid IP addresses, and ensure only the first node is
    * used as the valid contact point for automatic node discovery
    */
-  logger_.add_critera("Unable to establish a control connection to host 192.0.2.");
+  logger_.add_critera("to host 192.0.2.1 closed");
+  logger_.add_critera("to host 192.0.2.2 closed");
+  logger_.add_critera("to host 192.0.2.3 closed");
   Cluster cluster = default_cluster(false) // Do not add the default contact points
     .with_load_balance_round_robin()
     .with_contact_points(generate_contact_points("192.0.2.", 3)) // Invalid IPs

cpp-driver/gtests/src/integration/tests/test_control_connection.cpp

@@ -19,29 +19,116 @@
 
 class SessionTest : public Integration {
 public:
+  SessionTest()
+    : added_(0)
+    , removed_(0)
+    , up_(0)
+    , down_(0) { }
+
   void SetUp() {
-    logger_.add_critera("Attempted to get metrics before connecting session object");
-    logger_.add_critera("Attempted to get speculative execution metrics before connecting session object");
+    is_session_requested_ = false;
+    Integration::SetUp();
   }
 
-  void check_metrics_log_count() {
-    EXPECT_EQ(logger_.count(), 2u);
+  const std::string& address() const { return address_; }
+  int added() const { return added_; }
+  int removed() const { return removed_; }
+  int up() const { return up_; }
+  int down() const { return down_; }
+
+protected:
+  static void on_host_listener(CassHostListenerEvent event,
+                               CassInet inet,
+                               void* data) {
+    SessionTest* instance = static_cast<SessionTest*>(data);
+
+    char address[CASS_INET_STRING_LENGTH];
+    uv_inet_ntop(inet.address_length == CASS_INET_V4_LENGTH ? AF_INET : AF_INET6,
+                 inet.address,
+                 address, CASS_INET_STRING_LENGTH);
+    instance->address_ = address;
+
+    if (event == CASS_HOST_LISTENER_EVENT_ADD) {
+      TEST_LOG("Added node " << address);
+      ++instance->added_;
+    } else if (event == CASS_HOST_LISTENER_EVENT_REMOVE) {
+      TEST_LOG("Removed node " << address);
+      ++instance->removed_;
+    } else if (event == CASS_HOST_LISTENER_EVENT_UP) {
+      TEST_LOG("Up node " << address);
+      ++instance->up_;
+    } else if (event == CASS_HOST_LISTENER_EVENT_DOWN) {
+      TEST_LOG("Down node " << address);
+      ++instance->down_;
+    }
   }
+
+private:
+  std::string address_;
+  int added_;
+  int removed_;
+  int up_;
+  int down_;
 };
 
 CASSANDRA_INTEGRATION_TEST_F(SessionTest, MetricsWithoutConnecting) {
-  Session session(cass_session_new());
+  CHECK_FAILURE;
+
+  Session session;
 
   CassMetrics metrics;
+  logger_.add_critera("Attempted to get metrics before connecting session object");
   cass_session_get_metrics(session.get(), &metrics);
 
   EXPECT_EQ(metrics.requests.min, 0u);
   EXPECT_EQ(metrics.requests.one_minute_rate, 0.0);
+  EXPECT_EQ(logger_.count(), 1u);
 
   CassSpeculativeExecutionMetrics spec_ex_metrics;
+  logger_.reset();
+  logger_.add_critera("Attempted to get speculative execution metrics before connecting session object");
   cass_session_get_speculative_execution_metrics(session.get(), &spec_ex_metrics);
   EXPECT_EQ(spec_ex_metrics.min, 0u);
   EXPECT_EQ(spec_ex_metrics.percentage, 0.0);
+  EXPECT_EQ(logger_.count(), 1u);
+}
+
+CASSANDRA_INTEGRATION_TEST_F(SessionTest, ExternalHostListener) {
+  CHECK_FAILURE;
+  is_test_chaotic_ = true; // Destroy the cluster after the test completes
+
+  Cluster cluster = default_cluster()
+    .with_load_balance_round_robin()
+    .with_host_listener_callback(on_host_listener, this);
+  Session session = cluster.connect();
+
+  // Sanity check
+  EXPECT_EQ(0, added());
+  EXPECT_EQ(0, removed());
+  EXPECT_EQ(0, up());
+  EXPECT_EQ(0, down());
+
+  logger_.add_critera("New node " + (ccm_->get_ip_prefix() + "2") + " added");
+  EXPECT_EQ(2, ccm_->bootstrap_node());
+  EXPECT_TRUE(wait_for_logger(1));
+  EXPECT_EQ(1, added());
+  EXPECT_STREQ((ccm_->get_ip_prefix() + "2").c_str(), address().c_str());
+
+  stop_node(1);
+  EXPECT_EQ(1, down());
+  EXPECT_STREQ((ccm_->get_ip_prefix() + "1").c_str(), address().c_str());
+
+  ccm_->start_node(1);
+  EXPECT_EQ(1, up());
+  EXPECT_STREQ((ccm_->get_ip_prefix() + "1").c_str(), address().c_str());
+
+  force_decommission_node(1);
+  EXPECT_EQ(1, removed());
+  EXPECT_STREQ((ccm_->get_ip_prefix() + "1").c_str(), address().c_str());
 
-  check_metrics_log_count();
+  // Sanity check
+  EXPECT_EQ(1, added());
+  EXPECT_EQ(1, removed());
+  EXPECT_EQ(1, up());
+  EXPECT_EQ(1, down());
 }

cpp-driver/gtests/src/integration/tests/test_session.cpp

@@ -2014,20 +2014,11 @@ void Cluster::init(AddressGenerator& generator,
                    ClientConnectionFactory& factory,
                    size_t num_nodes_dc1,
                    size_t num_nodes_dc2) {
-  MT19937_64 token_rng;
   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);
+    create_and_add_server(generator, factory, "dc1");
   }
   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);
+    create_and_add_server(generator, factory, "dc2");
   }
 }
 
@@ -2118,17 +2109,30 @@ int Cluster::add(cass::EventLoopGroup* event_loop_group, size_t node) {
     return -1;
   }
   Server& server = servers_[node - 1];
-  server.is_removed.store(false);
+  bool is_removed = server.is_removed.exchange(false);
   server.connection->listen(event_loop_group);
-  return server.connection->wait_listen();
+  int rc = server.connection->wait_listen();
+
+  // Send the added node event after starting the socket
+  if (is_removed) { // Only send topology change event if node was previously removed
+    event(TopologyChangeEvent::new_node(server.connection->address()));
+  }
+
+  return rc;
 }
 
 void Cluster::remove(size_t node) {
   if (node < 1 || node > servers_.size()) {
     return;
   }
   Server& server = servers_[node - 1];
-  server.is_removed.store(true);
+  bool is_removed = server.is_removed.exchange(true);
+
+  // Send the remove node event before closing the socket
+  if (!is_removed) { // Only send the topology change event if node was previously active
+    event(TopologyChangeEvent::removed_node(server.connection->address()));
+  }
+
   server.connection->close();
   server.connection->wait_close();
 }
@@ -2140,6 +2144,7 @@ const Host& Cluster::host(const Address& address) const {
       return it->host;
     }
   }
+
   throw Exception(ERROR_PROTOCOL_ERROR,
                   "Unable to find host " + address.to_string());
 }
@@ -2156,6 +2161,18 @@ Hosts Cluster::hosts() const {
   return hosts;
 }
 
+int Cluster::create_and_add_server(AddressGenerator& generator,
+                                   ClientConnectionFactory& factory,
+                                   const String& dc) {
+  Address address(generator.next());
+  Server server(Host(address, dc, "rack1", token_rng_),
+                internal::ServerConnection::Ptr(
+                Memory::allocate<internal::ServerConnection>(address, factory)));
+
+  servers_.push_back(server);
+  return static_cast<int>(servers_.size());
+}
+
 void Cluster::event(const Event::Ptr& event) {
   for (Servers::const_iterator it = servers_.begin(),
        end = servers_.end(); it != end; ++it) {

cpp-driver/gtests/src/unit/mockssandra.cpp

@@ -1275,8 +1275,13 @@ class Cluster {
 
   typedef Vector<Server> Servers;
 
+  int create_and_add_server(AddressGenerator& generator,
+                            ClientConnectionFactory& factory,
+                            const String& dc);
+
 private:
   Servers servers_;
+  MT19937_64 token_rng_;
 };
 
 class SimpleEventLoopGroup : public cass::RoundRobinEventLoopGroup {