Add basic event listener to allow cpp-rs-driver to implement host listeners

[Lorak-mmk]

cpp-driver has a feature called listener callback. It allows users to register a function that will be called when some node status changes (added / removed / up / down).
As far as we can tell, it is based on events (+ sends artificial add+up events when connecting), without fully considering issues with dropped control connection.

To be able to implement this functionality in cpp-rs-driver, Rust Driver first needs to expose it.
Ultimately, we would like to have such channel exposed to user to:

• Expose more information than topology (see e.g. Provide event for the ScyllaDB schema change #1405 )
• Have "good" semantics: for example handle CC dropping and send artificial events to account for that after reconnecting. Without this, such callback is not really reliable. See Insufficient guarantees regarding ordering of events and control connection breakage. #1409

For now, it will be enough to just have a simple api guarded by cpp-rust-unstable, so that the cpp-rs-driver feature can be implemented. cpp-driver also doesn't consider the above issues, so it should not be a regression.

[wprzytula]

CPP Driver Host Listener Investigation

This analyses the host listener abstraction present in CPP Driver to understand exact events that trigger host listener callbacks.

API

Example of a HostListenerCallback:

void on_host_listener(CassHostListenerEvent event, CassInet inet, void* data) {
  char address[CASS_INET_STRING_LENGTH];

  cass_inet_string(inet, address);
  if (event == CASS_HOST_LISTENER_EVENT_ADD) {
    printf("Host %s has been ADDED\n", address);
  } else if (event == CASS_HOST_LISTENER_EVENT_REMOVE) {
    printf("Host %s has been REMOVED\n", address);
  } else if (event == CASS_HOST_LISTENER_EVENT_UP) {
    printf("Host %s is UP\n", address);
  } else if (event == CASS_HOST_LISTENER_EVENT_DOWN) {
    printf("Host %s is DOWN\n", address);
  }
}

cass_cluster_set_host_listener_callback is used to register the callback.

Code paths

Cluster config

HostListener configured here:

CassError cass_cluster_set_host_listener_callback(CassCluster* cluster,
                                                  CassHostListenerCallback callback, void* data) {
  cluster->config().set_host_listener(
      DefaultHostListener::Ptr(new ExternalHostListener(callback, data)));
  return CASS_OK;
}

HostListener access

1. Session::on_connect.

   for (HostMap::const_iterator it = hosts.begin(), end = hosts.end(); it != end; ++it) {
     const Host::Ptr& host = it->second;
     config().host_listener()->on_host_added(host);
     config().host_listener()->on_host_up(
         host); // If host is down it will be marked down later in the connection process
   }

2. Session::on_host_{up, down, added, removed}.
   These are only called in ClusterEvent::process_event.

Note that there is also event called HOST_MAYBE_UP, but it does not trigger host listener callbacks.

Cluster events machinery

ClusterEvent::process_event called from:

1. ClusterEvent::process_events, only called in Cluster::internal_start_events, caused by Cluster::start_events, only called in SessionBase::notify_connected(). ClusterEvent::process_events processes events that were previously recorded in Cluster::notify_or_record.
2. Cluster::notify_or_record.

So, both paths originate from Cluster::notify_or_record. This is called with relevant events from:

1. Cluster::notify_host_up_after_prepare -> UP, called by Cluster::{internal_}notify_host_up,
   which is called by Session::on_pool_up.
2. Cluster::{internal_}notify_host_down -> DOWN, only if LBP believes the host is UP (not to duplicate events).
3. Cluster::notify_host_add.
   • A bit strange logic here -> REMOVED:

     if (host_it != hosts_.end()) {
       LOG_WARN("Attempting to add host %s that we already have", host->address_string().c_str());
       // If an entry already exists then notify that the node has been removed
       // then re-add it.
       for (LoadBalancingPolicy::Vec::const_iterator it = load_balancing_policies_.begin(),
                                                     end = load_balancing_policies_.end();
           it != end; ++it) {
         (*it)->on_host_removed(host_it->second);
       }
       notify_or_record(ClusterEvent(ClusterEvent::HOST_REMOVE, host));
     }

   • Otherwise -> ADDED
4. Cluster::notify_host_remove -> DOWN, REMOVED

<Session as ConnectionPoolListener>::on_pool_{up,down}

Session::on_pool_up is called by:

1. ConnectionPoolManager::on_pool_up, RequestProcessor::on_pool_up, RequestProcessorInitializer::on_pool_up.
   All these are implementations of ConnectionPoolListener::on_pool_up, which is called only by ConnectionPool::notify_up_or_down.
   The chain of on_pool_up calls is as follows:
   • ConnectionPoolManager::on_pool_up calls ConnectionPoolManagerInitializer::on_pool_up.
   • ConnectionPoolManagerInitializer::on_pool_up calls RequestProcessorInitializer::on_pool_up;
   • RequestProcessorInitializer::on_pool_up calls Session::on_pool_up;
     So finally, Session is the listener.
2. ConnectionPool::notify_up_or_down().

ConnectionPool::notify_up_or_down()

```cpp
void ConnectionPool::notify_up_or_down() {
  if ((notify_state_ == NOTIFY_STATE_NEW || notify_state_ == NOTIFY_STATE_UP) && !has_connections()) {
    notify_state_ = NOTIFY_STATE_DOWN;
    listener_->on_pool_down(host_->address());
  } else if ((notify_state_ == NOTIFY_STATE_NEW || notify_state_ == NOTIFY_STATE_DOWN) && has_connections()) {
    notify_state_ = NOTIFY_STATE_UP;
    listener_->on_pool_up(host_->address());
  }
}
```

When is ConnectionPool::notify_up_or_down() called?

1. ConnectionPool constructor.
2. ConnectionPool::close_connection.
3. ConnectionPool::on_reconnect, when a connection is added.

So basically, everytime the pool may have lost the last connection or gained the first connection, the pool notifies its listener (Session), which may lead to host UP/DOWN notifications.

ControlConnection CQL Events Handling

Reactions to CQL host UP/DOWN events

The following methods are triggered once control connection detects host UP/DOWN events:

void Cluster::on_up(const Address& address) {
  LockedHostMap::const_iterator it = hosts_.find(address);

  if (it == hosts_.end()) {
    LOG_WARN("Received UP event for an unknown host %s", address.to_string().c_str());
    return;
  }

  notify_or_record(ClusterEvent(ClusterEvent::HOST_MAYBE_UP, it->second));
}

void Cluster::on_down(const Address& address) {
  // Ignore down events from the control connection. Use the method
  // `notify_host_down()` to trigger the DOWN status.
}

It's clear that CQL UP events lead to HOST_MAYBE_UP events, which do not trigger host listener callbacks. CQL DOWN events are ignored.
Thus, CQL host UP/DOWN events do not lead to host listener callbacks.

The code above mentions cluster method notify_host_down(), which is called from Session::on_pool_down:

void Session::on_pool_down(const Address& address) { cluster()->notify_host_down(address); }

which is called from ConnectionPool::notify_up_or_down() as explained earlier.

Reactions to CQL host ADDED/REMOVED events

These events cause the cluster event machinery to issue HOST_ADDED and HOST_REMOVED events, respectively.

Summary of events triggering host listener callbacks

Event	Triggering Action
ADDED	When a CQL event ADDED arrives.
REMOVED	When a CQL event REMOVED arrives, or when ADDED comes for a node that we already know (first issue REMOVED, then ADDED).
UP	When a connection pool transitions from having no connections to having at least one connection.
DOWN	When a connection pool transitions from having at least one connection to having no connections, or upon REMOVED if LBPs believe the node is still UP.

Additionally:

• On session connect, for all known hosts, ADDED and UP events are triggered.
• If pool has just been created and has connections, UP event is triggered.
• If pool has just been created and has no connections, DOWN event is triggered.

Deduplication of events

There is no deduplication of ADDED and REMOVED events.

UP and DOWN events are deduplicated based on the load balancing policy's view of host status,
because LBPs are always updated before host listener callbacks are invoked.

void Cluster::internal_notify_host_up(const Address& address) {
  LockedHostMap::const_iterator it = hosts_.find(address);

  if (it == hosts_.end()) {
    LOG_WARN("Attempting to mark host %s that we don't have as UP", address.to_string().c_str());
    return;
  }

  Host::Ptr host(it->second);

  if (load_balancing_policy_->is_host_up(address)) {
    // Already marked up so don't repeat duplicate notifications.
    if (!is_host_ignored(host)) {
      notify_or_record(ClusterEvent(ClusterEvent::HOST_READY, host));
    }
    return;
  }