CCBC-1602: implement Faster Failover

* Announce snappy compression for configuration payloads

* Supply version of the current configuration with GET_CLUSTER_CONFIG
  (0xb5) requests, to allow KV engine to skip the body.

* Support config deduplication, so NotMyVbucket responses allowed to
  have empty body.

* Support config notifications. When notifications are being used, the
  library will not poll for new configuration every 2.5 seconds.

Change-Id: Iabf06f4a9ce8c5bfe70cb8fae8926ec4fc886529
Reviewed-on: https://review.couchbase.org/c/libcouchbase/+/193942
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Brett Lawson <brett19@gmail.com>

Author: avsej

include/memcached/protocol_binary.h

@@ -402,6 +402,10 @@ typedef enum {
     PROTOCOL_BINARY_CMD_INVALID = 0xff
 } protocol_binary_command;
 
+typedef enum {
+    PROTOCOL_BINARY_CMD_CLUSTERMAP_CHANGE_NOTIFICATION = 0x01,
+} protocol_binary_notification;
+
 /**
  * Definition of the data types in the packet
  * See section 3.4 Data Types
@@ -835,38 +839,15 @@ typedef enum {
     PROTOCOL_BINARY_FEATURE_ALT_REQUEST_SUPPORT = 0x10,
     PROTOCOL_BINARY_FEATURE_SYNC_REPLICATION = 0x11,
     PROTOCOL_BINARY_FEATURE_COLLECTIONS = 0x12,
+    PROTOCOL_BINARY_FEATURE_SNAPPY_EVERYWHERE = 0x13,
     PROTOCOL_BINARY_FEATURE_PRESERVE_TTL = 0x14,
-    PROTOCOL_BINARY_FEATURE_CREATE_AS_DELETED = 0x17
+    PROTOCOL_BINARY_FEATURE_CREATE_AS_DELETED = 0x17,
+    PROTOCOL_BINARY_FEATURE_GET_CLUSTER_CONFIG_WITH_KNOWN_VERSION = 0x1d,
+    PROTOCOL_BINARY_FEATURE_DEDUPE_NOT_MY_VBUCKET_CLUSTERMAP = 0x1e,
+    PROTOCOL_BINARY_FEATURE_CLUSTERMAP_CHANGE_NOTIFICATION_BRIEF = 0x1f,
+    MEMCACHED_TOTAL_HELLO_FEATURES
 } protocol_binary_hello_features;
 
-#define MEMCACHED_FIRST_HELLO_FEATURE 0x01
-#define MEMCACHED_TOTAL_HELLO_FEATURES 15
-
-// clang-format off
-#define protocol_feature_2_text(a) \
-    (a == PROTOCOL_BINARY_FEATURE_INVALID) ? "Invalid" : \
-    (a == PROTOCOL_BINARY_FEATURE_TLS) ? "TLS" : \
-    (a == PROTOCOL_BINARY_FEATURE_TCPNODELAY) ? "TCP nodelay" : \
-    (a == PROTOCOL_BINARY_FEATURE_MUTATION_SEQNO) ? "Mutation seqno" : \
-    (a == PROTOCOL_BINARY_FEATURE_TCPDELAY) ? "TCP delay" : \
-    (a == PROTOCOL_BINARY_FEATURE_XATTR) ? "XATTR" : \
-    (a == PROTOCOL_BINARY_FEATURE_XERROR) ? "XERROR": \
-    (a == PROTOCOL_BINARY_FEATURE_SELECT_BUCKET) ? "Select bucket": \
-    (a == PROTOCOL_BINARY_FEATURE_INVALID2) ? "Invalid2": \
-    (a == PROTOCOL_BINARY_FEATURE_SNAPPY) ? "Snappy": \
-    (a == PROTOCOL_BINARY_FEATURE_JSON) ? "JSON": \
-    (a == PROTOCOL_BINARY_FEATURE_DUPLEX) ? "Duplex": \
-    (a == PROTOCOL_BINARY_FEATURE_CLUSTERMAP_CHANGE_NOTIFICATION) ? "Clustermap change notification": \
-    (a == PROTOCOL_BINARY_FEATURE_UNORDERED_EXECUTION) ? "Unordered execution": \
-    (a == PROTOCOL_BINARY_FEATURE_TRACING) ? "Tracing": \
-    (a == PROTOCOL_BINARY_FEATURE_ALT_REQUEST_SUPPORT) ? "Alt request support": \
-    (a == PROTOCOL_BINARY_FEATURE_SYNC_REPLICATION) ? "Synchronous Replication": \
-    (a == PROTOCOL_BINARY_FEATURE_COLLECTIONS) ? "Collections": \
-    (a == PROTOCOL_BINARY_FEATURE_PRESERVE_TTL) ? "Preserve Expiry": \
-    (a == PROTOCOL_BINARY_FEATURE_CREATE_AS_DELETED) ? "Create as Deleted": \
-    "Unknown"
-// clang-format on
-
 /**
  * The HELLO command is used by the client and the server to agree
  * upon the set of features the other end supports. It is initiated

src/bootstrap.cc

@@ -173,7 +173,6 @@ void Bootstrap::check_bgpoll()
 
 void Bootstrap::bgpoll()
 {
-    lcb_log(LOGARGS(parent, TRACE), "Background-polling for new configuration");
     bootstrap(BS_REFRESH_ALWAYS);
     check_bgpoll();
 }
@@ -246,6 +245,7 @@ lcb_STATUS Bootstrap::bootstrap(unsigned options)
         tm.rearm(LCBT_SETTING(parent, config_timeout));
         lcb_aspend_add(&parent->pendops, LCB_PENDTYPE_COUNTER, nullptr);
     } else if (parent->confmon->is_refreshing()) {
+        lcb_log(LOGARGS(parent, TRACE), "already refreshing, exit");
         return LCB_SUCCESS;
     }
 

src/bucketconfig/bc_cccp.cc

@@ -29,8 +29,9 @@
 #include <lcbio/timer-cxx.h>
 #include <lcbio/ssl.h>
 #include "ctx-log-inl.h"
+#include "mc/compress.h"
 
-#include <cstring>
+#include <stdio.h>
 
 #define LOGFMT CTX_LOGFMT
 #define LOGID(p) CTX_LOGID(p->ioctx)
@@ -51,13 +52,14 @@ struct CccpProvider : public Provider {
      *        because we have received a successful response.
      */
     void stop_current_request(bool is_clean);
-    lcb_STATUS schedule_next_request(lcb_STATUS err, bool can_rollover);
+    lcb_STATUS schedule_next_request(lcb_STATUS err, bool can_rollover, bool skip_if_push_supported);
+    lcb_STATUS expect_config_with_version(const lcb_host_t *origin, config_version version);
     lcb_STATUS mcio_error(lcb_STATUS err);
     void on_timeout()
     {
         mcio_error(LCB_ERR_TIMEOUT);
     }
-    lcb_STATUS update(const char *host, const char *data);
+    lcb_STATUS update(const char *host, const std::string &config_json);
     void request_config();
     void on_io_read();
 
@@ -96,6 +98,14 @@ struct CccpProvider : public Provider {
     lcb::io::ConnectionRequest *creq{};
     lcbio_CTX *ioctx;
     CccpCookie *cmdcookie;
+    /*
+     * The version, that the library has seen in the notifications, but hasn't fetched yet.
+     *
+     * For example, if KV engine sends clustermap notification while this config provider is waiting for response, the
+     * version number from the notification will be recorded here, and once the current operation will be completed, the
+     * provider will make new request if the expected version is newer than the current one.
+     */
+    config_version expected_config_version{-1, -1};
 };
 
 struct CccpCookie {
@@ -149,17 +159,104 @@ void CccpProvider::stop_current_request(bool is_clean)
     }
 }
 
-lcb_STATUS CccpProvider::schedule_next_request(lcb_STATUS err, bool can_rollover)
+/**
+ * Tell the provider that the newer configuration might be existing on the server side.
+ *
+ * If there is no request in-flight, the library will do PROTOCOL_BINARY_CMD_GET_CLUSTER_CONFIG (0xb5) immediately,
+ * otherwise update internal state with the version number and return;
+ *
+ * @param origin the hostname of the connection that was notification about configuration update
+ * @param requested the version of the configuration from the KV notification
+ * @return
+ */
+lcb_STATUS CccpProvider::expect_config_with_version(const lcb_host_t *origin, config_version requested)
+{
+
+    auto current = parent->get_current_version();
+    auto previous_expected = expected_config_version;
+    if (current < requested && previous_expected < requested) {
+        /*
+         * The requested version is newer than we've seen so far (both current and previously requested)
+         */
+        expected_config_version = requested;
+        if (has_pending_request()) {
+            /*
+             * Only one configuration request could be in-flight, but the version is stored in expected_config_version
+             * already and will be checked once current request completes.
+             */
+            lcb_log(LOGARGS(this, DEBUG),
+                    "Configuration request is in flight " LCB_HOST_FMT ": expected=%" PRId64 ":%" PRId64
+                    ", current=%" PRId64 ":%" PRId64 ", requested=%" PRId64 ":%" PRId64,
+                    LCB_HOST_ARG(parent->settings, origin), previous_expected.epoch, previous_expected.revision,
+                    current.epoch, current.revision, requested.epoch, requested.revision);
+            return LCB_SUCCESS;
+        }
+        /*
+         * Request new configuration immediately
+         */
+        return schedule_next_request(LCB_SUCCESS, /* can_rollover */ true, /* skip_if_push_supported */ false);
+    } else {
+        /*
+         * The config provider already seen this revision and probably already applied it and using as the current.
+         */
+        lcb_log(LOGARGS(this, TRACE),
+                "Ignore configuration request " LCB_HOST_FMT ": expected=%" PRId64 ":%" PRId64 ", current=%" PRId64
+                ":%" PRId64 ", requested=%" PRId64 ":%" PRId64,
+                LCB_HOST_ARG(parent->settings, origin), previous_expected.epoch, previous_expected.revision,
+                current.epoch, current.revision, requested.epoch, requested.revision);
+    }
+    return LCB_SUCCESS;
+}
+
+lcb_STATUS CccpProvider::schedule_next_request(lcb_STATUS err, bool can_rollover, bool skip_if_push_supported)
 {
-    lcb_host_t *next_host = nodes->next(can_rollover);
+    if (nodes->empty()) {
+        timer.cancel();
+        parent->provider_failed(this, err);
+        return err;
+    }
+
+    if (skip_if_push_supported && nodes->all_hosts_support_config_push()) {
+        /* all nodes support configuration push, and this function invoked from periodic poller, so nothing has to be
+         * done here */
+        parent->stop();
+        return LCB_SUCCESS;
+    }
+
+    lcb::Server *server{nullptr};
+    lcb_host_t *next_host = nodes->next(can_rollover, skip_if_push_supported);
     if (!next_host) {
         timer.cancel();
         parent->provider_failed(this, err);
         return err;
     }
 
-    lcb::Server *server = instance->find_server(*next_host);
-    if (server) {
+    do {
+        /* try to find connected socket that corresponds the hostname */
+        server = instance->find_server(*next_host);
+        if (server != nullptr && server->supports_config_push()) {
+            /* mark the address in the list, that it supports configuration push, so that it could be skipped later */
+            next_host->supports_config_push = true;
+            if (skip_if_push_supported) {
+                next_host = nodes->next(can_rollover, skip_if_push_supported);
+            } else {
+                break;
+            }
+        } else {
+            break;
+        }
+    } while (next_host != nullptr);
+
+    /* there is no connected sockets */
+    if (server != nullptr) {
+        if (skip_if_push_supported && server->supports_config_push()) {
+            /* we found the server, but at the same time all sockets support push, so we can stop polling and just
+             * expect notifications from KV engine */
+            lcb_log(LOGARGS(this, DEBUG), "Stop background polling, as all nodes support configuration push");
+            parent->stop();
+            return LCB_SUCCESS;
+        }
+
         cmdcookie = new CccpCookie(this);
         lcb_log(LOGARGS(this, TRACE), "Re-Issuing CCCP Command on server struct %p (" LCB_HOST_FMT ")", (void *)server,
                 LCB_HOST_ARG(this->parent->settings, next_host));
@@ -169,10 +266,10 @@ lcb_STATUS CccpProvider::schedule_next_request(lcb_STATUS err, bool can_rollover
             instance->select_bucket(cmdcookie, server);
         }
         cmdcookie->incref();
-        instance->request_config(cmdcookie, server);
+        instance->request_config(cmdcookie, server, parent->get_current_version());
 
     } else {
-
+        /* initiate new connection */
         lcb_log(LOGARGS(this, INFO), "Requesting connection to node " LCB_HOST_FMT " for CCCP configuration",
                 LCB_HOST_ARG(this->parent->settings, next_host));
         creq = instance->memd_sockpool->get(*next_host, settings().config_node_timeout, on_connected, this);
@@ -194,18 +291,28 @@ lcb_STATUS CccpProvider::mcio_error(lcb_STATUS err)
         parent->provider_failed(this, err);
         return err;
     } else {
-        return schedule_next_request(err, false);
+        return schedule_next_request(err, /* can_rollover */ false, /* skip_if_push_supported */ false);
     }
 }
 
 /** Update the configuration from a server. */
-lcb_STATUS lcb::clconfig::cccp_update(Provider *provider, const char *host, const char *data)
+lcb_STATUS lcb::clconfig::cccp_update(Provider *provider, const char *host, const std::string &config_json)
+{
+    return static_cast<CccpProvider *>(provider)->update(host, config_json);
+}
+
+lcb_STATUS lcb::clconfig::schedule_get_config(Provider *provider, const lcb_host_t *origin, config_version version)
 {
-    return static_cast<CccpProvider *>(provider)->update(host, data);
+    return static_cast<CccpProvider *>(provider)->expect_config_with_version(origin, version);
 }
 
-lcb_STATUS CccpProvider::update(const char *host, const char *data)
+lcb_STATUS CccpProvider::update(const char *host, const std::string &config_json)
 {
+    if (config_json.empty()) {
+        // ignore empty payloads, in case of brief mode
+        parent->stop();
+        return LCB_SUCCESS;
+    }
     lcbvb_CONFIG *vbc;
     int rv;
     ConfigInfo *new_config;
@@ -214,11 +321,11 @@ lcb_STATUS CccpProvider::update(const char *host, const char *data)
     if (!vbc) {
         return LCB_ERR_NO_MEMORY;
     }
-    rv = lcbvb_load_json_ex(vbc, data, host, &LCBT_SETTING(this->parent, network));
+    rv = lcbvb_load_json_ex(vbc, config_json.c_str(), host, &LCBT_SETTING(this->parent, network));
 
     if (rv) {
         lcb_log(LOGARGS(this, ERROR), LOGFMT "Failed to parse config", LOGID(this));
-        lcb_log_badconfig(LOGARGS(this, ERROR), vbc, data);
+        lcb_log_badconfig(LOGARGS(this, ERROR), vbc, config_json.c_str());
         lcbvb_destroy(vbc);
         return LCB_ERR_PROTOCOL_ERROR;
     }
@@ -238,6 +345,10 @@ lcb_STATUS CccpProvider::update(const char *host, const char *data)
     /** TODO: Figure out the comparison vector */
     config = new_config;
     parent->provider_got_config(this, new_config);
+
+    if (parent->get_current_version() < expected_config_version) {
+        return schedule_next_request(LCB_SUCCESS, /* can_rollover */ true, /* skip_if_push_supported */ false);
+    }
     return LCB_SUCCESS;
 }
 
@@ -248,8 +359,8 @@ void lcb::clconfig::select_status(const void *cookie_, lcb_STATUS err)
     cookie->decref();
 }
 
-void lcb::clconfig::cccp_update(const void *cookie_, lcb_STATUS err, const void *bytes, size_t nbytes,
-                                const lcb_host_t *origin)
+void lcb::clconfig::cccp_update(const void *cookie_, lcb_STATUS err, const lcb_host_t *origin,
+                                const std::string &config_json)
 {
     auto *cookie = reinterpret_cast<CccpCookie *>(const_cast<void *>(cookie_));
     CccpProvider *cccp = cookie->parent;
@@ -269,8 +380,7 @@ void lcb::clconfig::cccp_update(const void *cookie_, lcb_STATUS err, const void
     }
 
     if (err == LCB_SUCCESS) {
-        std::string ss(reinterpret_cast<const char *>(bytes), nbytes);
-        err = cccp->update(origin->host, ss.c_str());
+        err = cccp->update(origin->host, config_json);
     }
 
     if (err != LCB_SUCCESS && was_active) {
@@ -312,7 +422,7 @@ lcb_STATUS CccpProvider::refresh()
         return LCB_ERR_BUSY;
     }
 
-    return schedule_next_request(LCB_SUCCESS, true);
+    return schedule_next_request(LCB_SUCCESS, /* can_rollover */ true, /* skip_if_push_supported */ true);
 }
 
 bool CccpProvider::pause()
@@ -339,6 +449,9 @@ CccpProvider::~CccpProvider()
 
 void CccpProvider::configure_nodes(const lcb::Hostlist &nodes_)
 {
+    /* note, that provider assumes that none of the nodes supports configuration push. It will be checked later, when
+     * the address will be selected to fetch configuration. It allows to handle downgrade scenario, when newer server
+     * version replaced with older, that do not support configuration push */
     nodes->assign(nodes_);
     if (parent->settings->randomize_bootstrap_nodes) {
         nodes->randomize();
@@ -422,7 +535,7 @@ void CccpProvider::on_io_read()
         return_error(LCB_ERR_PROTOCOL_ERROR);
     }
 
-    std::string jsonstr(resp.value(), resp.vallen());
+    auto jsonstr = resp.inflated_value();
     std::string hoststr(lcbio_get_host(lcbio_ctx_sock(ioctx))->host);
 
     resp.release(ioctx);
@@ -433,14 +546,17 @@ void CccpProvider::on_io_read()
     if (err == LCB_SUCCESS) {
         timer.cancel();
     } else {
-        schedule_next_request(LCB_ERR_PROTOCOL_ERROR, false);
+        schedule_next_request(LCB_ERR_PROTOCOL_ERROR, /* can_rollover */ false, /* skip_if_push_supported */ false);
     }
 
 #undef return_error
 }
 
 void CccpProvider::request_config()
 {
+    lcb_log(LOGARGS(this, TRACE), "Attempting to retrieve cluster map via CCCP (timeout=%uus)",
+            settings().config_node_timeout);
+
     lcb::MemcachedRequest req(PROTOCOL_BINARY_CMD_GET_CLUSTER_CONFIG);
     req.opaque(0xF00D);
     lcbio_ctx_put(ioctx, req.data(), req.size());

src/bucketconfig/bc_file.cc

@@ -189,6 +189,7 @@ ConfigInfo *FileProvider::get_cached()
 
 void FileProvider::reload_cache()
 {
+    lcb_log(LOGARGS(this, TRACE), "Attempting to retrieve cluster map via FILE (path=\"%s\")", filename.c_str());
     if (load_cache() == UPDATED) {
         parent->provider_got_config(this, config);
     } else {

src/bucketconfig/bc_http.cc

@@ -415,6 +415,7 @@ lcb_STATUS HttpProvider::refresh()
 
     /** If we need a new socket, we do connect_next. */
     if (ioctx == nullptr && creq == nullptr) {
+        lcb_log(LOGARGS(this, TRACE), "Reconnect socket to retrieve cluster map via HTTP");
         as_reconnect.signal();
     }
     disconn_timer.cancel();