cluster_async.c

Go to the documentation of this file.

/*
 *   This program is is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or (at
 *   your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the Free Software
 *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
 */
 
/**
 * $Id: 9f28cd72cbd682d534c702097e81a7bb384d1f90 $
 * @file cluster_async.c
 * @brief conf functions for interacting asynchronously with Redis cluster via Hiredis.
 *
 * @author Arran Cudbard-Bell (a.cudbardb@freeradius.org)
 *
 * @copyright 2026 Network RADIUS (legal@networkradius.com)
 *
 * Overview
 * ========
 *
 * Read and understand this http://redis.io/topics/cluster-spec first, else the text below
 * will not be useful.
 *
 * Using the cluster's public API
 * ------------------------------
 *
 *  The cluster requires use of a coordinator to fetch the cluster map.
 *
 *  Any module using a Redis cluster should register a coordinator which uses
 *  the `redis` virtual server.
 *
 *  In `mod_coord_attach`, #fr_redis_ct_map_bootstrap can be used to initiate
 *  the loading of the cluster map.  Typically this should not be called if
 *  the pool start is set to zero.
 *  In that case, the first attempt to enqueue a command set will indicate
 *  that the cluster map need to be bootstrapped.
 *
 *  At runtime the function #fr_redis_async_cmd_start is used to enqueue a set of
 *  commands, and the statis it returns should be checked with the macro
 *  REDIS_ASYNC_START_RCODE_PROCESS to initiate the cluster bootstrap or get
 *  an updated map if needed.
 *
 *  With calling Redis using it's async API, the majority of results processing has to
 *  be done in a callback called by hiredis - the `redisReply` structure is freed
 *  after the callback is called.
 *
 *  This callback is associated with the individual commands in a Redis command set
 *  as they are added to the command set with the fr_redis_command_*_add functions.
 *
 * Structures
 * ----------
 *
 *   This code maintains a series structures for efficient lookup and lockless operations.
 *
 *   The important ones are:
 *     - An array of #fr_redis_cluster_node_t.  These are pre-allocated on startup and are
 *       never added to, or removed from.
 *     - An #fr_fifo_t.  This contains the queue of nodes that may be re-used.
 *     - An #fr_rb_tree_t.  This contains a tree of nodes which are active.  The tree is built on IP
 *       address and port.
 *
 *   Each #fr_redis_cluster_node_t contains a master ID, and an array of slave IDs.  The IDs are array
 *   indexes in the fr_redis_cluster_t.node array.  We use 8bit unsigned integers instead of
 *   pointers to save space.  Using pointers, the node[] array would need 784K, using IDs
 *   it uses 112K.  Still not light on memory, but a bit more acceptable.
 *   Currently the key_slot array is shadowed by key_slot_pending, used to stage new key_slot
 *   mappings.  This doubles the memory used.  We may want to consider allocating key_slot_pending
 *   only during re-mappings and freeing it after.
 *
 * Mapping/Remapping the cluster
 * -----------------------------
 *
 *   On startup, and during cluster operation, a remap may be performed.  A remap involves
 *   the following steps:
 *
 *     1. Request the cluster map from the coordinator.
 *     2. The coordinator:
 *        a. Checks to see when it last fetched the cluster map.  If it was less than 1 second ago,
 *           replies with the most recently fetched data.
 *        b. Executes the Redis 'cluster info' on all known nodes to establish which  nodes believe
 *           they can see a working cluster, from the `cluster_state` response and which nodes have
 *           the most up to date representation of the cluster, from the `cluster_current_epoch`
 *           response.
 *        c. Nodes reporting the cluster is OK are issued the Redis 'cluster slots' or 'cluster shards'
 *           command depending on the Redis server version.
 *        d. Validating the result of this command.  We need to do extensive validation to
 *           avoid SEGV on invalid data, due to the way libhiredis presents the result.
 *        e. Return the cluster map to the workers.
 *     4. Determining the intersection between nodes described in the result, and those already
 *        in our #fr_rb_tree_t.
 *     5. Creating trunk connections to nodes that were in the result, but not in the tree.
 *     6. Mapping keyslot ranges to nodes in the key_slot_pending array.
 *     7. Verifying there are no holes in the ranges (if there are, we roll back and error out).
 *     8. Applying the new keyslot ranges.
 *     9. Removing nodes no longer used by the key slots, and adding them back to the free
 *        nodes queue.
 *
 *   #fr_redis_ct_map_get is used to request an updated map from the coordinator and
 *   #fr_redis_ct_map_update is used to process the message received from the coordinator to update
 *   the thread local copy of the cluster map.
 *
 *   The cluster client can continue to operate, albeit inefficiently, with a stale cluster map
 *   by following '-ASK' and '-MOVE' redirects.
 *
 *   Remaps are limited to one per second.  If any operation sets the remap_needed flag, or
 *   attempts a remap directly, the remap may be skipped if one occurred recently.
 *
 *
 * Processing '-ASK' and '-MOVE' redirects
 * ---------------------------------------
 *
 *   Resume functions which are run after an async Redis command set has completed should
 *   fetch the rcode with #fr_redis_command_set_rcode.
 *   If the rcode indicates MOVE or ASK, then #fr_redis_async_cmd_redirect should be used
 *   to re-enqueue the command set on the indicated node. In addition, if the response
 *   was MOVE, then #fr_redis_ct_map_get should be used to initiate a refresh of the cluster map.
 *
 *   The data from '-MOVE' responses, is not used to alter the cluster map.  That is only done
 *   on successful remap.
 *
 *
 * Processing '-TRYAGAIN'
 * ----------------------
 *
 *   If the cluster is in a state of flux, a node may return '-TRYAGAIN' to indicated that we
 *   should attempt the operation again.  #fr_redis_async_cmd_resend can be used to re-enqueue
 *   the command set.
 *
 */
 
#include <freeradius-devel/util/debug.h>
 
#include "config.h"
#include "attrs.h"
#include "cluster_async.h"
#include "crc16.h"
 
#ifndef WITH_TLS
#  undef HAVE_REDIS_SSL
#endif
 
#ifdef HAVE_REDIS_SSL
#include <freeradius-devel/tls/strerror.h>
#include <hiredis/hiredis_ssl.h>
#endif
 
#define MAX_REPLICAS            5                       //!< Maximum number of replicas associated
                                                        //!< with a keyslot.
struct fr_redis_ct_key_slot_s {
        uint8_t         replica[MAX_REPLICAS];          //!< Array of ids of replica nodes
        uint8_t         num_replicas;                   //!< Number of replica nodes
        uint8_t         master;                         //!< id of the master node.
};
 
typedef enum {
        CLUSTER_INIT = 0,                               //!< Cluster has been initialised.
        CLUSTER_MAP_FETCHING,                           //!< The cluster map is currently being fetched.
        CLUSTER_READY,                                  //!< The cluster is available to handle requests.
        CLUSTER_FAIL,                                   //!< The coordinator reported a failed cluster map update.
} fr_redis_ct_state_t;
 
/** Thread local state for a cluster
 *
 */
struct fr_redis_ct_s {
        uint16_t                        cluster_id;     //!< Number assigned to the cluster by coordinator.
        fr_event_list_t                 *el;
        trunk_conf_t    const           *tconf;         //!< Configuration for all trunks in the cluster.
        bool                            delay_start;    //!< Prevent connections from spawning immediately.
        fr_redis_conf_t const           *conf;          //!< Redis configuration for the cluster.
        CONF_SECTION const              *tls_cs;        //!< TLS CONF_SECTION
 
        fr_redis_trunk_active_t         active;         //!< Callback to run when the trunk becomes active.
        void                            *active_uctx;   //!< Uctx to pass to active callback.
        bool                            active_oneshot; //!< Should the callback only be called once.
        uint8_t                         max_redirects;  //!< maximum number of redirects.
 
#ifdef HAVE_REDIS_SSL
        SSL_CTX                         *ssl_ctx;       //!< SSL context.
#endif
 
        fr_redis_ct_node_t              *node;          //!< Array of nodes in this cluster.
        fr_fifo_t                       *free_nodes;    //!< Nodes not currently active.
        fr_rb_tree_t                    *used_nodes;    //!< Active nodes.
 
        fr_redis_ct_key_slot_t          key_slot[KEY_SLOTS];
 
        fr_redis_ct_state_t             state;          //!< State of the cluster.
        fr_dlist_head_t                 pend_cmds;      //!< Commands awaiting cluster map.
        fr_dlist_head_t                 pend_reqs;      //!< Requests awaiting cluster map.
        fr_time_t                       map_updated;    //!< Time the cluster last updated.
};
 
struct fr_redis_ct_node_s {
        fr_rb_node_t                    rbnode;         //!< Entry in the tree of used nodes
        char                            name[INET6_ADDRSTRLEN];
        uint8_t                         id;             //!< Array offset in the array of available nodes.
 
        bool                            is_active;      //!< Is this node currently active.
        bool                            is_master;      //!< Is this node currently a master.
 
        fr_redis_ct_t                   *rtcluster;     //!< Cluster this node belongs to
        fr_redis_io_conf_t              ioconf;         //!< Connection config for this node.
        fr_redis_trunk_t                *trunk;         //!< Trunk connection to this node.
        fr_pair_list_t                  trigger_args;   //!< Pairs to pass to trigger functions.
};
 
/** Structure for holding the state of an async redis command set.
 *
 */
struct fr_redis_async_cmd_s {
        request_t                       *request;       //!< Request this command set relates to.
        fr_redis_ct_t                   *rtcluster;     //!< Cluster this command set is running on.
        fr_redis_trunk_t                *rtrunk;        //!< Trunk the command set is currently running on.
        fr_redis_command_set_t          *cmds;          //!< Command set to run.
        uint8_t const                   *key;           //!< Key used to identify key slot.
        size_t                          key_len;        //!< Length of key.
        fr_redis_ct_key_slot_t const    *key_slot;      //!< Key slot identified from the command key.
        bool                            read_only;      //!< Should this command be run read only.
        uint8_t                         replica_no;     //!< Current replica number being used.
        fr_dlist_t                      entry;          //!< Entry in the list of commands waiting for a cluster remap.
        fr_redis_ct_node_t              *node;          //!< Specific node to run command set on.
};
 
/** Structure to record that a request is waiting for the cluster map.
 *
 */
typedef struct {
        request_t                       *request;       //!< The request waiting for the map.
        fr_dlist_t                      entry;          //!< Entry in the list of pending requests.
        fr_redis_ct_t                   *rtcluster;     //!< Cluster the request is waiting for.
} fr_redis_ct_pend_req_t;
 
#define CONFIGURE_NODE(_node, _addr, _port) \
do {  \
        _node->ioconf = (fr_redis_io_conf_t) { \
                .port = _port, \
                .database = rtcluster->conf->database, \
                .username = rtcluster->conf->username, \
                .password = rtcluster->conf->password, \
                .use_tls = rtcluster->conf->use_tls, \
        }; \
        _node->ioconf.hostname = talloc_strdup(rtcluster, _addr); \
        _node->ioconf.log_prefix = talloc_asprintf(rtcluster, "%s %s:%d", rtcluster->conf->log_prefix, \
                                                   _addr, _node->ioconf.port); \
        if (rtcluster->conf->trunk_conf.conn_triggers) { \
                module_trigger_args_build(rtcluster, &_node->trigger_args, NULL, \
                                        &(module_trigger_args_t) { \
                                                .module = rtcluster->conf->module_name, \
                                                .name = rtcluster->conf->inst_name, \
                                                .server = _addr, \
                                                .port = _node->ioconf.port \
                                        }); \
        } \
        _node->trunk = fr_redis_trunk_alloc(rtcluster, &_node->ioconf, &_node->trigger_args, rtcluster->active, \
                                            rtcluster->active_uctx, rtcluster->active_oneshot); \
        if (!_node->trunk) goto error; \
} while (0)
 
/** Resolve key to key slot index
 *
 * Identical to the example implementation, except it uses memchr which will
 * be faster, and isn't so needlessly complex.
 *
 * @param[in] key to resolve.
 * @param[in] key_len length of key.
 * @return key slot index for the key.
 */
static uint16_t cluster_key_hash(uint8_t const *key, size_t key_len)
{
        uint8_t *p, *q;
 
        p = memchr(key, '{', key_len);
        if (!p) {
        all:
                return fr_crc16_xmodem(key, key_len) & (KEY_SLOTS - 1);
        }
 
        q = memchr(p, '}', key_len - (p - key)); /* look for } after { */
        if (!q || (q == p + 1)) goto all; /* no } or {}, hash everything */
 
        p++;    /* skip '{' */
 
        return fr_crc16_xmodem(p, q - p) & (KEY_SLOTS - 1);     /* hash stuff between { and } */
}
 
/** Resolve key to key slot
 *
 * @param[in] rtcluster to resolve the key slot in.
 * @param[in] request Current request (for debugging).
 * @param[in] key to resolve.
 * @param[in] key_len length of key.
 * @return key slot for the key.
 */
fr_redis_ct_key_slot_t const *fr_redis_ct_slot_by_key(fr_redis_ct_t *rtcluster, request_t *request,
                                                      uint8_t const *key, size_t key_len)
{
        fr_redis_ct_key_slot_t *key_slot;
 
        if (!key || (key_len == 0)) {
                key_slot = &rtcluster->key_slot[(uint16_t)(fr_rand() & (KEY_SLOTS - 1))];
                ROPTIONAL(RDEBUG2, DEBUG2, "Key rand() -> slot %zu", key_slot - rtcluster->key_slot);
 
                return key_slot;
        }
 
        /*
         *      Avoid CRC16 if we're operating with one cluster node or
         *      without clustering.
         */
        if (fr_rb_num_elements(rtcluster->used_nodes) > 1) {
                key_slot = &rtcluster->key_slot[cluster_key_hash(key, key_len)];
                ROPTIONAL(RDEBUG2, DEBUG2, "Key \"%pV\" -> slot %zu",
                          fr_box_strvalue_len((char const *)key, key_len), key_slot - rtcluster->key_slot);
 
                return key_slot;
        }
        ROPTIONAL(RDEBUG3, DEBUG3, "Single node available, skipping key selection");
 
        return &rtcluster->key_slot[0];
}
 
/** Return the master node that would be used for a particular key slot
 *
 * @param[in] rtcluster         To resolve key slot in.
 * @param[in] key_slot          to resolve to node.
 * @return
 *      - The current master node.
 *      - NULL if no master node is currently assigned to a particular key slot.
 */
fr_redis_ct_node_t const *fr_redis_ct_master(fr_redis_ct_t *rtcluster,
                                             fr_redis_ct_key_slot_t const *key_slot)
{
        return &rtcluster->node[key_slot->master];
}
 
/** Return the replica node that would be used for a particular key slot
 *
 * @param[in] rtcluster         To resolve key slot in.
 * @param[in] key_slot          To resolve to node.
 * @param[in] replica_num               0..n.
 * @return
 *      - A replica node.
 *      - NULL if no replica node is assigned, or is at the specific key slot.
 *
 */
fr_redis_ct_node_t const *fr_redis_ct_replica(fr_redis_ct_t *rtcluster,
                                              fr_redis_ct_key_slot_t const *key_slot, uint8_t replica_num)
{
        if (replica_num >= key_slot->num_replicas) return NULL; /* No replica available */
 
        return &rtcluster->node[key_slot->replica[replica_num]];
}
 
/** Return the ipaddr of a particular node
 *
 * @param[in] node      to get ip address from.
 * @return
 *      - IP address of node
 *      - NULL on failure (node is NULL).
 */
char const * fr_redis_ct_ipaddr(fr_redis_ct_node_t const *node)
{
        if (!node) return NULL;
 
        return node->ioconf.hostname;
}
 
/** Return the port of a particular node
 *
 * @param[out] out      Port of the node.
 * @param[in] node      to get ip address from.
 * @return
 *      - 0 on success.
 *      - -1 on failure (node is NULL).
 */
int fr_redis_ct_port(uint16_t *out, fr_redis_ct_node_t const *node)
{
        if (!node) return -1;
 
        *out = node->ioconf.port;
 
        return 0;
}
 
/** Enqueue a command set on a node identified by the key.
 *
 */
static fr_redis_async_rcode_t fr_redis_async_cmd_enqueue(fr_redis_async_cmd_t *cmd)
{
        fr_redis_ct_t   *rtcluster = cmd->rtcluster;
        fr_redis_pipeline_status_t      ret;
        bool                            dst_unavail = false;
 
        if (likely(!cmd->node)) cmd->key_slot = fr_redis_ct_slot_by_key(rtcluster, cmd->request, cmd->key, cmd->key_len);
 
        if (unlikely(cmd->node != NULL)) {
                cmd->rtrunk = cmd->node->trunk;
        }
        /*
         *      Read only commands start on the first replica, if there are any.
         */
        else if (cmd->read_only && cmd->key_slot->num_replicas) {
                cmd->rtrunk = rtcluster->node[cmd->key_slot->replica[0]].trunk;
        } else {
                cmd->rtrunk = rtcluster->node[cmd->key_slot->master].trunk;
        }
 
        if (unlikely(!cmd->rtrunk)) return REDIS_ASYNC_RCODE_ERROR;
 
again:
        ret = redis_command_set_enqueue(cmd->rtrunk, cmd->cmds);
 
        switch (ret) {
        case FR_REDIS_PIPELINE_OK:
                /*
                 *      If one or more nodes reported failed to enqueue with
                 *      destination unavailable, tell the caller that the cluster
                 *      map should be updated.
                 */
                return dst_unavail ? REDIS_ASYNC_RCODE_GETMAP : REDIS_ASYNC_RCODE_SUCCESS;
 
        case FR_REDIS_PIPELINE_DST_UNAVAILABLE:
                if (cmd->node) return REDIS_ASYNC_RCODE_ERROR;
                dst_unavail = true;
                if (cmd->replica_no < cmd->key_slot->num_replicas) {
                        cmd->rtrunk = rtcluster->node[cmd->key_slot->replica[cmd->replica_no]].trunk;
                        cmd->replica_no++;
                        goto again;
                }
                /*
                 *      Read only commands can also try the master node.
                 *      Non-read only first tried the master.
                 */
                if (cmd->read_only && (cmd->rtrunk != rtcluster->node[cmd->key_slot->master].trunk)) {
                        cmd->rtrunk = rtcluster->node[cmd->key_slot->master].trunk;
                        goto again;
                }
                return REDIS_ASYNC_RCODE_ERROR;
 
        default:
                return REDIS_ASYNC_RCODE_ERROR;
        }
 
}
 
static int _fr_redis_async_cmd_free(fr_redis_async_cmd_t *cmd)
{
        if (!fr_dlist_entry_in_list(&cmd->entry)) return 0;
        fr_dlist_remove(&cmd->rtcluster->pend_cmds, cmd);
        return 0;
}
 
/** Start running a command set on an async redis cluster
 *
 * @param ctx           to allocate tracking structure.
 * @param request       current request.
 * @param rcode         Where to write the result code.
 * @param rtcluster     to start the command set on
 * @param key           to identify the cluster slot.
 * @param key_len       Length of key.
 * @param cmds          Command set to run.
 * @param read_only     Should the command set be run on read only nodes.
 * @param node          Specific node to run the command set on.
 * @return The async redis command
 */
fr_redis_async_cmd_t *fr_redis_async_cmd_start(TALLOC_CTX *ctx, request_t *request, fr_redis_async_rcode_t *rcode,
                                               fr_redis_ct_t *rtcluster, uint8_t const *key, size_t key_len,
                                               fr_redis_command_set_t *cmds, bool read_only, fr_redis_ct_node_t *node)
{
        fr_redis_async_cmd_t            *cmd;
 
        MEM(cmd = talloc(ctx, fr_redis_async_cmd_t));
 
        *cmd = (fr_redis_async_cmd_t) {
                .request = request,
                .rtcluster = rtcluster,
                .cmds = cmds,
                .read_only = read_only,
                .key = key,
                .key_len = key_len,
                .node = node,
        };
 
        switch (rtcluster->state) {
        case CLUSTER_INIT:
                /*
                 *      If the cluster has not bootstrapped, that must be done first.
                 */
                fr_dlist_insert_tail(&rtcluster->pend_cmds, cmd);
                talloc_set_destructor(cmd, _fr_redis_async_cmd_free);
                *rcode = REDIS_ASYNC_RCODE_BOOTSTRAP;
                break;
 
        case CLUSTER_MAP_FETCHING:
                fr_dlist_insert_tail(&rtcluster->pend_cmds, cmd);
                talloc_set_destructor(cmd, _fr_redis_async_cmd_free);
                *rcode = REDIS_ASYNC_RCODE_SUCCESS;
                break;
 
        case CLUSTER_FAIL:
                /*
                 *      The coordinator reported a failed cluster.
                 */
                *rcode = REDIS_ASYNC_RCODE_FAIL;
                fr_strerror_const("Cluster failed");
                talloc_free(cmd);
                return NULL;
 
        default:
                *rcode = fr_redis_async_cmd_enqueue(cmd);
                break;
        }
 
        return cmd;
}
 
/** Cancel a Redis async command.
 *
 */
void fr_redis_async_cmd_cancel(fr_redis_async_cmd_t *cmd)
{
        fr_redis_command_set_cancel(cmd->cmds);
}
 
/** Fetch the redis trunk a command is associated with.
 *
 */
fr_redis_trunk_t *fr_redis_async_cmd_trunk(fr_redis_async_cmd_t *cmd)
{
        return cmd->rtrunk;
}
 
/** Fetch the cluster node a command was last sent to
 */
fr_redis_ct_node_t *fr_redis_async_cmd_node(fr_redis_async_cmd_t *cmd) {
        if (!cmd->key_slot) return NULL;
        if (cmd->replica_no == 0) return &cmd->rtcluster->node[cmd->key_slot->master];
        return &cmd->rtcluster->node[cmd->key_slot->replica[cmd->replica_no - 1]];
}
 
/** Re-submit a redis async command set on a different node
 *
 * Using the node returned by a MOVED / ASK response.
 * @param cmd   Async command set to redirect
 * @return fr_redis_async_rcode_t
 */
fr_redis_async_rcode_t fr_redis_async_cmd_redirect(fr_redis_async_cmd_t *cmd)
{
        fr_redis_ct_node_t      find, *cluster_node;
 
        fr_redis_command_set_next_node(cmd->cmds, &find.ioconf);
 
        fr_rb_find((void **)&cluster_node, cmd->rtcluster->used_nodes, &find);
        if (!cluster_node) {
                ERROR("Asked to redirect to a node not in the current cluster map");
                return REDIS_ASYNC_RCODE_ERROR;
        }
 
        fr_redis_command_set_reset(cmd->cmds);
        cmd->node = cluster_node;
        return fr_redis_async_cmd_enqueue(cmd);
}
 
/** Re-submit a redis async command set
 *
 * To be used following TRYAGAIN responses
 * @param cmd   Async command set to redirect
 * @return fr_redis_async_rcode_t
 */
fr_redis_async_rcode_t fr_redis_async_cmd_resend(fr_redis_async_cmd_t *cmd)
{
        fr_redis_command_set_reset(cmd->cmds);
        return fr_redis_async_cmd_enqueue(cmd);
}
 
/** Compare two redis nodes to check equality
 *
 * @param[in] one first node.
 * @param[in] two second node.
 * @return CMP(one, two)
 */
static fr_cmp_ret_t _cluster_thread_node_cmp(void const *one, void const *two)
{
        fr_redis_ct_node_t const *a = one;
        fr_redis_ct_node_t const *b = two;
        int ret;
 
        ret = strcmp(a->ioconf.hostname, b->ioconf.hostname);
        if (ret != 0) return CMP(ret, 0);
 
        return CMP(a->ioconf.port, b->ioconf.port);
}
 
#ifdef HAVE_REDIS_SSL
static int _redis_cluster_thread_free(fr_redis_ct_t *rtcluster)
{
        if (rtcluster->ssl_ctx) SSL_CTX_free(rtcluster->ssl_ctx);
        return 0;
}
#endif
 
/** Allocate per-thread, per-cluster instance
 *
 * This structure represents all the connections for a given thread for a given cluster.
 * The structures holds the trunk connections to talk to each cluster member.
 *
 */
fr_redis_ct_t *fr_redis_ct_alloc(TALLOC_CTX *ctx, CONF_SECTION *tls_cs, fr_event_list_t *el, fr_redis_conf_t *conf,
                                 fr_redis_trunk_active_t active, void *active_uctx, bool active_oneshot)
{
        fr_redis_ct_t   *rtcluster;
        trunk_conf_t                    *our_tconf;
        uint8_t                         i;
        uint32_t                        s, num_nodes;
 
        MEM(rtcluster = talloc_zero(ctx, fr_redis_ct_t));
        *rtcluster = (fr_redis_ct_t) {
                .el = el,
                .conf = conf,
                .tls_cs = tls_cs,
                .active = active,
                .active_uctx = active_uctx,
                .active_oneshot = active_oneshot
        };
        MEM(our_tconf = talloc_memdup(rtcluster, &conf->trunk_conf, sizeof(conf->trunk_conf)));
        our_tconf->always_writable = true;
 
        rtcluster->tconf = our_tconf;
        fr_dlist_talloc_init(&rtcluster->pend_cmds, fr_redis_async_cmd_t, entry);
        fr_dlist_talloc_init(&rtcluster->pend_reqs, fr_redis_ct_pend_req_t, entry);
 
        if (conf->max_nodes == UINT8_MAX) {
                ERROR("%s - Maximum number of connected nodes allowed is %i", conf->log_prefix, UINT8_MAX - 1);
        error:
                talloc_free(rtcluster);
                return NULL;
        }
 
        if (conf->max_nodes == 0) {
                ERROR("%s - Minimum number of nodes allowed is 1", conf->log_prefix);
                goto error;
        }
 
        MEM(rtcluster->node = talloc_zero_array(rtcluster, fr_redis_ct_node_t, conf->max_nodes + 1));
        MEM(rtcluster->used_nodes = fr_rb_inline_alloc(rtcluster, fr_redis_ct_node_t, rbnode, _cluster_thread_node_cmp, NULL));
        MEM(rtcluster->free_nodes = fr_fifo_create(rtcluster, conf->max_nodes, NULL));
 
        /*
         *      Node id 0 is reserved, so we can detect misconfigured
         *      clusters.
         */
        for (i = 1; i <= conf->max_nodes; i++) {
                rtcluster->node[i].id = i;
                rtcluster->node[i].rtcluster = rtcluster;
                fr_pair_list_init(&rtcluster->node[i].trigger_args);
 
                /* Push them all into the queue */
                fr_fifo_push(rtcluster->free_nodes, &rtcluster->node[i]);
        }
 
        if (conf->use_tls) {
#ifdef HAVE_REDIS_SSL
                fr_tls_conf_t *tls_conf;
                if (!tls_cs) {
                        ERROR("%s - Missing TLS configuration", conf->log_prefix);
                        goto error;
                }
 
                tls_conf = fr_tls_conf_parse_client(tls_cs);
                if (!tls_conf) {
                        ERROR("%s - Failed to parse TLS configuration", conf->log_prefix);
                        goto error;
                }
 
                rtcluster->ssl_ctx = fr_tls_ctx_alloc(tls_conf, true);
                if (!rtcluster->ssl_ctx) {
                        ERROR("%s - Failed to allocate SSL context", conf->log_prefix);
                        goto error;
                }
                talloc_set_destructor(rtcluster, _redis_cluster_thread_free);
#else
                WARN("%s - No redis SSL support, ignoring \"use_tls = yes\"", conf->log_prefix);
#endif
        }
 
        /*
         *      cmds->redirected is uint8_t, so we limit the number of redirects here.
         */
        if (rtcluster->conf->max_redirects > UINT8_MAX) {
                rtcluster->max_redirects = UINT8_MAX;
        } else {
                rtcluster->max_redirects = rtcluster->conf->max_redirects;
        }
 
        if (conf->use_cluster_map) return rtcluster;
 
        /*
         *      If we are not using a cluster map, just configure nodes from
         *      the bootstrap list and distribute them through the key slots.
         */
        for (s = 0; s < talloc_array_length(conf->hostname); s++) {
                fr_redis_ct_node_t      *cluster_node;
                fr_ipaddr_t             addr;
                uint16_t                port;
                char                    buff[FR_IPADDR_STRLEN];
 
                cluster_node = fr_fifo_pop(rtcluster->free_nodes);
                if (!cluster_node) {
                        ERROR("Reached maximum connected nodes");
                        goto error;
                }
                if (fr_inet_pton_port(&addr, &port,
                                      conf->hostname[s], talloc_strlen(conf->hostname[s]), AF_UNSPEC, true, true) < 0) {
                        PERROR("Failed parsing %s", conf->hostname[s]);
                        goto error;
                }
                if (port == 0) port = conf->port;
                fr_inet_ntop(buff, sizeof(buff), &addr);
                CONFIGURE_NODE(cluster_node, buff, port);
                fr_rb_insert(rtcluster->used_nodes, cluster_node);
                cluster_node->is_active = true;
                cluster_node->is_master = true;
        }
 
        num_nodes = fr_rb_num_elements(rtcluster->used_nodes);
        if (!num_nodes) {
                ERROR("%s - No bootstrap servers configured", conf->log_prefix);
                goto error;
        }
 
        for (s = 0; s < KEY_SLOTS; s++) rtcluster->key_slot[s].master = (s % (uint16_t) num_nodes) + 1;
 
        rtcluster->state = CLUSTER_READY;
 
        return rtcluster;
}
 
fr_event_list_t *fr_redis_ct_el(fr_redis_ct_t *rtcluster)
{
        return rtcluster->el;
}
 
trunk_conf_t const *fr_redis_ct_trunk_conf(fr_redis_ct_t *rtcluster)
{
        return rtcluster->tconf;
}
 
/** How many times a command set running on this cluster may be redirected
 *
 * @param[in] rtcluster to return the configured limit for.
 * @return the value of the max_redirects configuration item.
 */
uint8_t fr_redis_ct_max_redirects(fr_redis_ct_t *rtcluster)
{
        return rtcluster->max_redirects;
}
 
#ifdef HAVE_REDIS_SSL
SSL_CTX *fr_redis_ct_ssl_ctx(fr_redis_ct_t *rtcluster)
{
        return rtcluster->ssl_ctx;
}
#endif
 
/** Update a Redis cluster map from a pair list returned from a coordinator
 *
 * @param rtcluster     Cluster to update
 * @param list          pairs sent by a coordinator
 * @return
 *      - 0 om success
 *      - -1 on error
 */
int fr_redis_ct_map_update(fr_redis_ct_t *rtcluster, fr_pair_list_t const *list)
{
        fr_pair_t       *vp, *shard = NULL, *slot, *start, *end, *node, *role, *node_ip, *node_port;
        uint16_t        i;
        uint8_t         r = 0;
        uint8_t         rollback[UINT8_MAX];            // Set of nodes to re-add to the queue on failure.
        bool            active[UINT8_MAX];              // Set of nodes active in the new cluster map.
        bool            master[UINT8_MAX];              // Master nodes.
 
        fr_redis_ct_node_t      find, *cluster_node;
        fr_redis_ct_key_slot_t  tmp_slot;
        fr_redis_ct_key_slot_t  key_slot_pending[KEY_SLOTS];
        fr_redis_async_cmd_t    *cmd;
        fr_redis_ct_pend_req_t  *pend_req;
 
#define SET_INACTIVE(_node) \
do { \
        (_node)->is_active = false; \
        (_node)->is_master = false; \
        talloc_const_free((_node)->ioconf.log_prefix); \
        (_node)->ioconf.log_prefix = NULL; \
        TALLOC_FREE((_node)->trunk); \
        fr_pair_list_free(&(_node)->trigger_args); \
        fr_rb_delete(rtcluster->used_nodes, _node); \
        fr_fifo_push(rtcluster->free_nodes, _node); \
} while (0)
 
#define SET_ACTIVE(_node) \
do { \
        fr_rb_insert(rtcluster->used_nodes, _node); \
        fr_fifo_pop(rtcluster->free_nodes); \
        (_node)->is_active = true; \
        active[(_node)->id] = true; \
        rollback[r++] = (_node)->id; \
} while (0)
 
        vp = fr_pair_find_by_da(list, NULL, attr_redis_cluster_id);
        if (unlikely(!vp)) {
                ERROR("Missing cluster ID");
                return -1;
        }
        if (rtcluster->cluster_id == 0) rtcluster->cluster_id = vp->vp_uint16;
 
        if (rtcluster->cluster_id != vp->vp_uint16) {
                ERROR("Got map for cluster ID %d, expected ID %d", vp->vp_uint16, rtcluster->cluster_id);
                return -1;
        }
 
        DEBUG3("Updating cluster %d", rtcluster->cluster_id);
 
        memset(&key_slot_pending, 0, sizeof(key_slot_pending));
        memset(active, 0, sizeof(active));
        memset(master, 0, sizeof(master));
 
        while ((shard = fr_pair_find_by_da(list, shard, attr_redis_shard))) {
                cluster_node = NULL;
                memset(&tmp_slot, 0, sizeof(fr_redis_ct_key_slot_t));
                node = NULL;
                while ((node = fr_pair_find_by_da(&shard->vp_group, node, attr_redis_node))) {
                        role = fr_pair_find_by_da(&node->vp_group, NULL, attr_redis_node_role);
                        if (unlikely(!role)) continue;
                        if (role->vp_uint8 == 1) {
                                DEBUG3("Master node %pP", node);
 
                                node_ip = fr_pair_find_by_da(&node->vp_group, NULL, attr_redis_node_endpoint);
                                if (unlikely(!node_ip)) continue;
                                find.ioconf.hostname = node_ip->vp_strvalue;
                                node_port = fr_pair_find_by_da(&node->vp_group, NULL, attr_redis_node_port);
                                if (unlikely(!node_port)) continue;
                                find.ioconf.port = node_port->vp_uint16;
 
                                fr_rb_find((void **)&cluster_node, rtcluster->used_nodes, &find);
                                break;
                        }
                }
 
                if (!node) {
                        ERROR("Missing master node");
                error:
                        for (i = 0; i < r; i++) SET_INACTIVE(&rtcluster->node[rollback[i]]);
                        return -1;
                }
 
                if (!cluster_node) {
                        cluster_node = fr_fifo_peek(rtcluster->free_nodes);
                        if (!cluster_node) {
                        out_of_nodes:
                                fr_strerror_const("Reached maximum connected nodes");
                                goto error;
                        }
                        CONFIGURE_NODE(cluster_node, find.ioconf.hostname, find.ioconf.port);
                        SET_ACTIVE(cluster_node);
                } else {
                        active[cluster_node->id] = true;
                }
                master[cluster_node->id] = true;
                tmp_slot.master = cluster_node->id;
 
                node = NULL;
                while ((node = fr_pair_find_by_da(&shard->vp_group, node, attr_redis_node))) {
                        role = fr_pair_find_by_da(&node->vp_group, NULL, attr_redis_node_role);
                        if (unlikely(!role)) continue;
                        if (tmp_slot.num_replicas >= MAX_REPLICAS) break;
                        if (role->vp_uint8 != 2) continue;
 
                        DEBUG3("Replica node %pP", node);
                        node_ip = fr_pair_find_by_da(&node->vp_group, NULL, attr_redis_node_endpoint);
                        if (unlikely(!node_ip)) continue;
                        find.ioconf.hostname = node_ip->vp_strvalue;
                        node_port = fr_pair_find_by_da(&node->vp_group, NULL, attr_redis_node_port);
                        if (unlikely(!node_port)) continue;
                        find.ioconf.port = node_port->vp_uint16;
 
                        fr_rb_find((void **)&cluster_node, rtcluster->used_nodes, &find);
 
                        if (cluster_node) {
                                tmp_slot.replica[tmp_slot.num_replicas++] = cluster_node->id;
                                active[cluster_node->id] = true;
                                continue;
                        }
 
                        cluster_node = fr_fifo_peek(rtcluster->free_nodes);
                        if (!cluster_node) goto out_of_nodes;
 
                        CONFIGURE_NODE(cluster_node, find.ioconf.hostname, find.ioconf.port);
                        tmp_slot.replica[tmp_slot.num_replicas++] = cluster_node->id;
                        SET_ACTIVE(cluster_node);
                }
 
                slot = NULL;
                while ((slot = fr_pair_find_by_da(&shard->vp_group, slot, attr_redis_slot))) {
                        start = fr_pair_find_by_da(&slot->vp_group, NULL, attr_redis_slot_start);
                        if (unlikely(!start)) {
                                ERROR("Missing slot start");
                                goto error;
                        }
                        if (unlikely(start->vp_uint16 >= KEY_SLOTS)) {
                                ERROR("Value of %d for slot start greater than expected maximum %d",
                                      start->vp_uint16, KEY_SLOTS);
                                goto error;
                        }
                        end = fr_pair_find_by_da(&slot->vp_group, NULL, attr_redis_slot_end);
                        if (unlikely(!end)) {
                                ERROR("Missing slot end");
                                goto error;
                        }
                        if (unlikely(end->vp_uint16 >= KEY_SLOTS)) {
                                ERROR("Value of %d for slot end greater than expected maximum %d",
                                      end->vp_uint16, KEY_SLOTS);
                                goto error;
                        }
                        if (unlikely(end->vp_uint16 < start->vp_uint16)) {
                                ERROR("Value of %d for slot end less than value of %d for slot start",
                                      end->vp_uint16, start->vp_uint16);
                                goto error;
                        }
                        DEBUG4("Setting nodes for slots %d to %d", start->vp_uint16, end->vp_uint16);
                        for (i = start->vp_uint16; i <= end->vp_uint16; i++) {
                                memcpy(&key_slot_pending[i], &tmp_slot, sizeof(*key_slot_pending));
                        }
                }
        }
 
        memcpy(&rtcluster->key_slot, &key_slot_pending, sizeof(rtcluster->key_slot));
 
        /*
         *      Anything not in the active set of nodes gets
         *      added back into the queue, to be re-used.
         *
         *      We start at 1, as node 0 is reserved.
         */
        for (i = 1; i <= rtcluster->conf->max_nodes; i++) {
#ifndef NDEBUG
                fr_redis_ct_node_t *found;
 
                if (rtcluster->node[i].is_active) {
                        /* Sanity check for duplicates that are active */
                        fr_rb_find((void **)&found, rtcluster->used_nodes, &rtcluster->node[i]);
                        fr_assert(found);
                        fr_assert(found->is_active);
                        fr_assert(found->id == i);
                }
#endif
 
                if (!active[i] && rtcluster->node[i].is_active) {
                        SET_INACTIVE(&rtcluster->node[i]);
 
                /*
                 *      Only change the masters once we've successfully
                 *      remapped the cluster.
                 */
                } else if (master[i]) {
                        rtcluster->node[i].is_master = true;
                } else {
                        rtcluster->node[i].is_master = false;
                }
        }
 
        rtcluster->state = CLUSTER_READY;
        rtcluster->map_updated = fr_time();
 
        /*
         *      Enqueue any commands which were waiting for the cluster remap.
         */
        while ((cmd = fr_dlist_pop_head(&rtcluster->pend_cmds))) {
                fr_redis_async_cmd_enqueue(cmd);
        }
 
        /*
         *      Resume any requests which were waiting for the cluster remap.
         */
        while ((pend_req = fr_dlist_pop_head(&rtcluster->pend_reqs))) {
                unlang_interpret_mark_runnable(pend_req->request);
                talloc_free(pend_req);
        }
 
        return 0;
}
 
/** Process a cluster map fail message from the coordinator.
 *
 * @param rtcluster     Cluster to update
 * @param list          pairs sent by a coordinator
 */
int fr_redis_ct_map_fail(fr_redis_ct_t *rtcluster, UNUSED fr_pair_list_t const *list)
{
        fr_redis_async_cmd_t    *cmd;
        fr_redis_ct_pend_req_t  *pend_req;
 
        DEBUG3("Cluster %d failed", rtcluster->cluster_id);
        rtcluster->state = CLUSTER_FAIL;
 
        /*
         *      Inform any pending requests that the cluster has failed.
         */
        while ((cmd = fr_dlist_pop_head(&rtcluster->pend_cmds))) {
                fr_redis_command_set_rcode_set(cmd->cmds, REDIS_ASYNC_RCODE_FAIL);
                unlang_interpret_mark_runnable(cmd->request);
        }
 
        /*
         *      Resume any requests which were waiting for the cluster remap.
         */
        while ((pend_req = fr_dlist_pop_head(&rtcluster->pend_reqs))) {
                unlang_interpret_mark_runnable(pend_req->request);
                talloc_free(pend_req);
        }
 
        return 0;
}
 
/** Initiate bootstrapping of the cluster map
 *
 * To be used when a module first wants to fetch a cluster map
 *
 * @param rtcluster             Cluster to fetch map for
 * @param cw                    Coord worker to launch request
 * @param coord_pair_reg        Coord pair registration
 * @return
 *      - 0 on success.
 *      - -1 on failure.
 */
int fr_redis_ct_map_bootstrap(fr_redis_ct_t *rtcluster, fr_coord_worker_t *cw, fr_coord_pair_reg_t *coord_pair_reg)
{
        fr_redis_conf_t const   *conf = rtcluster->conf;
        fr_pair_list_t          list;
        fr_pair_t               *vp;
        TALLOC_CTX              *local = talloc_new(NULL);
        int                     ret;
        size_t                  i;
 
        fr_pair_list_init(&list);
        fr_pair_list_append_by_da(local, vp, &list, attr_redis_packet_type, (uint32_t)FR_REDIS_CLUSTER_MAP_BOOTSTRAP, false);
        if (!vp) {
        error:
                talloc_free(local);
                return -1;
        }
 
        if (fr_pair_append_by_da(local, &vp, &list, attr_redis_log_prefix) < 0) goto error;
        if (fr_value_box_strdup(vp, &vp->data, NULL, conf->log_prefix, false) < 0) goto error;
 
        fr_pair_list_append_by_da(local, vp, &list, attr_redis_max_nodes, conf->max_nodes, false);
        if (!vp) goto error;
 
        for (i = 0; i < talloc_array_length(conf->hostname); i++) {
                if (fr_pair_append_by_da(local, &vp, &list, attr_redis_bootstrap_node) < 0) goto error;
                if (fr_value_box_strdup(vp, &vp->data, NULL, conf->hostname[i], false) < 0) goto error;
        }
 
        fr_pair_list_append_by_da(local, vp, &list, attr_redis_bootstrap_port, conf->port, false);
        if (!vp) goto error;
 
        if (conf->password) {
                if (fr_pair_append_by_da(local, &vp, &list, attr_redis_password) < 0) goto error;
                if (fr_value_box_strdup(vp, &vp->data, NULL, conf->password, false) < 0) goto error;
                if (conf->username) {
                        if (fr_pair_append_by_da(local, &vp, &list, attr_redis_username) < 0) goto error;
                        if (fr_value_box_strdup(vp, &vp->data, NULL, conf->username, false) < 0) goto error;
                }
        }
 
        if (conf->use_tls) {
                uintptr_t tls_conf = (uintptr_t)rtcluster->tls_cs;
                fr_pair_list_append_by_da(local, vp, &list, attr_redis_use_tls, false, false);
                if (!vp) goto error;
                fr_pair_list_append_by_da(local, vp, &list, attr_redis_tls_conf, (uint64_t)tls_conf, false);
                if (!vp) goto error;
        }
 
        ret = fr_worker_to_coord_pair_send(cw, coord_pair_reg, &list);
        talloc_free(local);
 
        if (ret < 0) return -1;
        rtcluster->state = CLUSTER_MAP_FETCHING;
 
        return 0;
}
 
/** Initiate updating of the cluster map
 *
 * To be used when a command returns MOVED
 */
fr_redis_async_rcode_t fr_redis_ct_map_get(fr_redis_ct_t *rtcluster, fr_coord_worker_t *cw,
                                           fr_coord_pair_reg_t *coord_pair_reg, bool force)
{
        fr_pair_list_t          list;
        fr_pair_t               *vp;
        TALLOC_CTX              *local;
        int                     ret;
 
        if (rtcluster->cluster_id == 0) return REDIS_ASYNC_RCODE_BOOTSTRAP;
 
        /*
         *      The update request has already been sent.
         */
        if ((rtcluster->state == CLUSTER_MAP_FETCHING) && !force) return REDIS_ASYNC_RCODE_SUCCESS;
 
        /*
         *      If the cluster was updated less than 1 sec ago, don't ask.
         */
        if (((fr_time_to_sec(fr_time()) == fr_time_to_sec(rtcluster->map_updated))) && !force) return REDIS_ASYNC_RCODE_SUCCESS;
 
        DEBUG3("Requesting updated map for cluster %d", rtcluster->cluster_id);
 
        local = talloc_new(NULL);
        fr_pair_list_init(&list);
        fr_pair_list_append_by_da(local, vp, &list, attr_redis_packet_type, (uint32_t)FR_REDIS_CLUSTER_MAP_GET, false);
        if (!vp) {
        error:
                talloc_free(local);
                return REDIS_ASYNC_RCODE_ERROR;
        }
 
        fr_pair_list_append_by_da(local, vp, &list, attr_redis_cluster_id, rtcluster->cluster_id, false);
        if (!vp) goto error;
 
        if (force) {
                fr_pair_list_append_by_da(local, vp, &list, attr_redis_force_update, true, false);
        }
 
        ret = fr_worker_to_coord_pair_send(cw, coord_pair_reg, &list);
        talloc_free(local);
 
        if (ret < 0) return REDIS_ASYNC_RCODE_ERROR;
        rtcluster->state = CLUSTER_MAP_FETCHING;
 
        return REDIS_ASYNC_RCODE_SUCCESS;
}
 
fr_redis_ct_node_t *fr_redis_ct_node_by_addr(fr_redis_ct_t *rtcluster, fr_redis_io_conf_t *ioconf)
{
        fr_redis_ct_node_t      find, *found;
 
        find.ioconf.hostname = ioconf->hostname;
        find.ioconf.port = ioconf->port;
        fr_rb_find((void **)&found, rtcluster->used_nodes, &find);
        return found;
}
 
fr_redis_async_rcode_t fr_redis_ct_node_addr_by_role(TALLOC_CTX *ctx, fr_redis_io_conf_t *out[], uint8_t *count_out,
                                                     fr_redis_ct_t *rtcluster, bool is_master, bool is_replica)
{
        uint64_t                in_use = fr_rb_num_elements(rtcluster->used_nodes);
        fr_rb_iter_inorder_t    iter;
        fr_redis_ct_node_t      *node;
        uint8_t                 count = 0;
        fr_redis_io_conf_t      *found;
 
        switch (rtcluster->state) {
        case CLUSTER_INIT:
                return REDIS_ASYNC_RCODE_BOOTSTRAP;
 
        case CLUSTER_MAP_FETCHING:
                return REDIS_ASYNC_RCODE_TRY_AGAIN;
 
        default:
                break;
        }
 
        if (in_use == 0) {
                *out = NULL;
                *count_out = 0;
                return REDIS_ASYNC_RCODE_SUCCESS;
        }
 
        found = talloc_zero_array(ctx, fr_redis_io_conf_t, in_use);
        if (!found) {
                fr_strerror_const("Out of memory");
                return REDIS_ASYNC_RCODE_ERROR;
        }
 
        for (node = fr_rb_iter_init_inorder(rtcluster->used_nodes, &iter);
             node;
             node = fr_rb_iter_next_inorder(rtcluster->used_nodes, &iter)) {
                if ((is_master && node->is_master) || (is_replica && !node->is_master)) found[count++] = node->ioconf;
        }
 
        if (count == 0) {
                *out = NULL;
                talloc_free(found);
        } else {
                *out = found;
        }
        *count_out = count;
        return REDIS_ASYNC_RCODE_SUCCESS;
}
 
/** Ensure pending request is removed from the list on freeing.
 */
static int _fr_redis_ct_pend_req_free(fr_redis_ct_pend_req_t *pend_req)
{
        if (!fr_dlist_entry_in_list(&pend_req->entry)) return 0;
        fr_dlist_remove(&pend_req->rtcluster->pend_reqs, pend_req);
        return 0;
}
 
/** Add a request to the list of those waiting for the cluster map
 *
 */
void fr_redis_ct_request_yield(TALLOC_CTX *ctx, fr_redis_ct_t *rtcluster, request_t *request)
{
        fr_redis_ct_pend_req_t  *pend_req;
 
        MEM(pend_req = talloc_zero(ctx, fr_redis_ct_pend_req_t));
        pend_req->request = request;
        pend_req->rtcluster = rtcluster;
        fr_dlist_insert_tail(&rtcluster->pend_reqs, pend_req);
        talloc_set_destructor(pend_req, _fr_redis_ct_pend_req_free);
}