/*
 *   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: b9168ca817664a51f6a310fa5655eee0fc006d05 $
 * @file redis.c
 * @brief conf functions for interacting with Redis cluster via Hiredis.
 *
 * @author Arran Cudbard-Bell
 *
 * @copyright 2015 Arran Cudbard-Bell (a.cudbardb@freeradius.org)
 * @copyright 2015 Network RADIUS (legal@networkradius.com)
 * @copyright 2015 The FreeRADIUS server project
 *
 * 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 two functions used at runtime to issue commands are #fr_redis_cluster_state_init
 * and #fr_redis_cluster_state_next.
 *
 * #fr_redis_cluster_state_init initialises the structure we use to track which node
 * we're communicating with, and uses a key (string) to determine which node we should
 * try and contact first.
 *
 * #fr_redis_cluster_state_next examines the result of the last command, and either
 * gets a new connection, or errors out.
 *
 * In both cases the connection should not be released by the caller, only by
 * #fr_redis_cluster_state_next.
 *
 * Below is the sequence of calls required to use the cluster:
 *
 *     1. During initialization allocate a cluster configuration structure with
 *	  #fr_redis_cluster_alloc.  This holds the cluster configuration and state.
 *     2. Declare a #fr_redis_cluster_state_t in the function that needs to issue
 *        commands against the cluster.
 *     3. Call #fr_redis_cluster_state_init with relevant arguments including a pointer
 *        to the #fr_redis_cluster_state_t struct, and a key/key_len used for initial
 *        lookup.  For most commands the key is the value of the second argument.
 *        #fr_redis_cluster_state_init will then and reserve/pass back a connection for
 *        the pool associated with the node associated with the key.
 *     4. Use the connection that was passed back, to issue a Redis command against.
 *     5. Use #fr_redis_command_status to translate the result of the command into
 *        a #fr_redis_rcode_t value.
 *     6. Call #fr_redis_cluster_state_next with relevant arguments including a pointer
 *        to the #fr_redis_cluster_state_t struct and the #fr_redis_rcode_t value for the
 *        last command.
 *     7. If #fr_redis_cluster_state_next returns 0, repeat steps 4-7.  Otherwise
 *        stop and analyse the return value.
 *
 * See #fr_redis_cluster_state_init for example code.
 *
 * 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 remappings 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. Executing the Redis 'cluster slots' command.
 *     2. 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.
 *     3. Determining the intersection between nodes described in the result, and those already
 *        in our #fr_rb_tree_t.
 *     4. Connecting to nodes that were in the result, but not in the tree.
 *        Note: If we can't connect to any of the masters, we count the map as invalid, roll
 *        back any newly connected nodes, and error out. Slave failure is OK.
 *     5. Mapping keyslot ranges to nodes in the key_slot_pending array.
 *     6. Verifying there are no holes in the ranges (if there are, we roll back and error out).
 *     7. Applying the new keyslot ranges.
 *     8. Removing nodes no longer used by the key slots, and adding them back to the free
 *        nodes queue.
 *
 *   #cluster_map_get and #cluster_map_apply, perform the operations described
 *   above. The get function, issues the 'cluster slots' command and performs validation, the
 *   apply function processes and applies the map.
 *
 *   Failing to apply a map is not a fatal error at runtime, and is only fatal on startup if
 *   pool.start > 0.
 *
 *   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
 * ---------------------------------------
 *
 *   The code treats '-ASK' (temporary redirect) and '-MOVE' (permanent redirect) responses
 *   similarly.  If the node is known, then a connection is reserved from its pool, if the node
 *   is not known, a new pool is established, and a connection reserved.
 *
 *   The difference between '-ASK' and '-MOVE' is that '-MOVE' attempts a cluster remap before
 *   following the redirect.
 *
 *   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.  The cluster spec says we should attempt the operation
 *   after some time.  This time is configurable.
 *
 */

#include <freeradius-devel/util/debug.h>
#include <freeradius-devel/server/cf_parse.h>

#include <freeradius-devel/util/fifo.h>
#include <freeradius-devel/util/misc.h>
#include <freeradius-devel/util/rand.h>
#include <freeradius-devel/util/time.h>

#include "config.h"
#include "base.h"
#include "cluster.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 KEY_SLOTS		16384			//!< Maximum number of keyslots (should not change).

#define MAX_SLAVES		5			//!< Maximum number of slaves associated
							//!< with a keyslot.

/*
 *	Periods and weights for live node selection
 */
#define CLOSED_PERIOD		10000			//!< How recently must the closed have
							//!< occurred for us to care.

#define CLOSED_WEIGHT		1			//!< What weight to give to nodes that
							//!< had a connection closed recently.

#define FAILED_PERIOD		10000			//!< How recently must the spawn failure
							//!< occurred for us to care.

#define FAILED_WEIGHT		1			//!< What weight to give to nodes that
							//!< had a spawn failure recently.

#define RELEASED_PERIOD		10000			//!< Period after which we don't care
							//!< about when the last connection was
							//!< released.

#define RELEASED_MIN_WEIGHT	1000			//!< Minimum weight to assign to node.

/** Live nodes data, used to perform weighted random selection of alternative nodes
 */
typedef struct {
	struct {
		uint8_t					id;		//!< Node ID.
		fr_pool_state_t const	*pool_state;	//!< Connection pool stats.
		unsigned int				cumulative;	//!< Cumulative weight.
	} node[UINT8_MAX - 1];				//!< Array of live node IDs (and weights).
	uint8_t next;					//!< Next index in live.
	uint8_t skip;
} cluster_nodes_live_t;

/** A Redis cluster node
 *
 * Passed as opaque data to pools which open connection to nodes.
 */
struct fr_redis_cluster_node_s {
	fr_rb_node_t		rbnode;			//!< Entry into the tree of redis nodes.

	char			name[INET6_ADDRSTRLEN];	//!< Buffer to hold IP string.
							//!< text for debug messages.
	uint8_t			id;			//!< Node ID (index in node array).

	fr_socket_t		addr;			//!< Current node address.
	fr_socket_t		pending_addr;		//!< New node address to be applied when the pool
							//!< is reconnected.

	fr_redis_cluster_t	*cluster;		//!< Common configuration (database number,
							//!< password, etc..).
	fr_pool_t		*pool;			//!< Pool associated with this node.
	CONF_SECTION		*pool_cs;		//!< Pool configuration section associated with node.

	bool			is_active;		//!< Whether this node is in the active node set.
	bool			is_master;		//!< Whether this node is a master.
							//!< This is needed for commands like 'KEYS', which
							//!< we need to issue to every master in the cluster.
};

/** Indexes in the fr_redis_cluster_node_t array for a single key slot
 *
 * When dealing with 16K entries, space is a concern. It's significantly
 * more memory efficient to use 8bit indexes than 64bit pointers for each
 * of the key slot to node mappings.
 */
struct fr_redis_cluster_key_slot_s {
	uint8_t			slave[MAX_SLAVES];	//!< R/O node (slave) for this key slot.
	uint8_t			slave_num;		//!< Number of slaves associated with this key slot.
	uint8_t			master;			//!< R/W node (master) for this key slot.
};

/** A redis cluster
 *
 * Holds all the structures and collections of nodes, to represent a Redis cluster.
 */
struct fr_redis_cluster {
	char const		*log_prefix;		//!< What to prepend to log messages.
	char const		*trigger_prefix;	//!< Trigger path.
	fr_pair_list_t		trigger_args;		//!< Arguments to pass to triggers.
	bool			triggers_enabled;	//!< Whether triggers are enabled.

	bool			remapping;		//!< True when cluster is being remapped.
	bool			remap_needed;		//!< Set true if at least one cluster node is definitely
							//!< unreachable. Set false on successful remap.
	fr_time_t      		last_updated;		//!< Last time the cluster mappings were updated.
	CONF_SECTION		*module;		//!< Module configuration.

	fr_redis_conf_t		*conf;			//!< Base configuration data such as the database number
							//!< and passwords.
#ifdef HAVE_REDIS_SSL
	SSL_CTX			*ssl_ctx;		//!< SSL context.
#endif

	fr_redis_cluster_node_t	*node;			//!< Structure containing a node id, its address and
							//!< a pool of its connections.

	fr_fifo_t		*free_nodes;		//!< Queue of free nodes (or nodes waiting to be reused).
	fr_rb_tree_t		*used_nodes;		//!< Tree of used nodes.

	fr_redis_cluster_key_slot_t	key_slot[KEY_SLOTS];		//!< Lookup table of slots to pools.
	fr_redis_cluster_key_slot_t	key_slot_pending[KEY_SLOTS];	//!< Pending key slot table.

	pthread_mutex_t		mutex;			//!< Mutex to synchronise cluster operations.
};

fr_table_num_sorted_t const fr_redis_cluster_rcodes_table[] = {
	{ L("bad-input"),	FR_REDIS_CLUSTER_RCODE_BAD_INPUT	},
	{ L("failed"),		FR_REDIS_CLUSTER_RCODE_FAILED		},
	{ L("ignored"),		FR_REDIS_CLUSTER_RCODE_IGNORED		},
	{ L("no-connection"),	FR_REDIS_CLUSTER_RCODE_NO_CONNECTION	},
	{ L("success"),		FR_REDIS_CLUSTER_RCODE_SUCCESS		}
};
size_t fr_redis_cluster_rcodes_table_len = NUM_ELEMENTS(fr_redis_cluster_rcodes_table);

/** Resolve key to key slot
 *
 * 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 } */
}

/** Compare two redis nodes to check equality
 *
 * @param[in] one first node.
 * @param[in] two second node.
 * @return CMP(one, two)
 */
static int8_t _cluster_node_cmp(void const *one, void const *two)
{
	fr_redis_cluster_node_t const *a = one;
	fr_redis_cluster_node_t const *b = two;
	int ret;

	ret = fr_ipaddr_cmp(&a->addr.inet.dst_ipaddr, &b->addr.inet.dst_ipaddr);
	if (ret != 0) return ret;

	return CMP(a->addr.inet.dst_port, b->addr.inet.dst_port);
}

/** Reconnect callback to apply new pool config
 *
 * @param[in] pool to apply new configuration to.
 * @param[in] opaque data passed to the connection pool.
 */
static void _cluster_node_conf_apply(fr_pool_t *pool, void *opaque)
{
	fr_pair_list_t	args;
	fr_redis_cluster_node_t	*node = opaque;

	fr_pair_list_init(&args);
	node->addr = node->pending_addr;

	if (node->cluster->triggers_enabled) {
		trigger_args_afrom_server(pool, &args, node->name, node->addr.inet.dst_port);
		if (fr_pair_list_empty(&args)) return;

		if (!fr_pair_list_empty(&node->cluster->trigger_args)) MEM(fr_pair_list_copy(node->cluster, &args,
								      &node->cluster->trigger_args) >= 0);

		fr_pool_enable_triggers(pool, node->cluster->trigger_prefix, &args);

		fr_pair_list_free(&args);
	}
}

/** Establish a connection to a cluster node
 *
 * @note Must be called with the cluster mutex locked.
 * @note Configuration to use for the connection must be set in node->pending_addr, not node->cluster->conf.
 *
 * @param[in] cluster to search in.
 * @param[in] node config.
 * @return
 *	 - FR_REDIS_CLUSTER_RCODE_SUCCESS on success.
 *	 - FR_REDIS_CLUSTER_RCODE_FAILED if the operation failed.
 */
static fr_redis_cluster_rcode_t cluster_node_connect(fr_redis_cluster_t *cluster, fr_redis_cluster_node_t *node)
{
	char const *p;

	fr_assert(node->pending_addr.inet.dst_ipaddr.af);

	/*
	 *	Write out the IP address and Port in string form
	 */
	p = inet_ntop(node->pending_addr.inet.dst_ipaddr.af, &node->pending_addr.inet.dst_ipaddr.addr,
		      node->name, sizeof(node->name));
	if (!fr_cond_assert(p)) return FR_REDIS_CLUSTER_RCODE_FAILED;

	/*
	 *	Node has never been used before, needs a pool allocated for it.
	 */
	if (!node->pool) {
		char		buffer[256];
		fr_pair_list_t	args;
		CONF_SECTION	*pool;

		fr_pair_list_init(&args);
		snprintf(buffer, sizeof(buffer), "%s [%i]", cluster->log_prefix, node->id);

		pool = cf_section_find(cluster->module, "pool", NULL);
		/*
		 *	Dup so we can re-parse, and have unique CONF_DATA
		 */
		node->pool_cs = cf_section_dup(cluster, NULL, pool, "pool", NULL, true);
		node->addr = node->pending_addr;
		node->pool = fr_pool_init(cluster, node->pool_cs, node,
					  fr_redis_cluster_conn_create, NULL, buffer);
		if (!node->pool) {
		error:
			TALLOC_FREE(node->pool_cs);
			TALLOC_FREE(node->pool);
			return FR_REDIS_CLUSTER_RCODE_FAILED;
		}
		fr_pool_reconnect_func(node->pool, _cluster_node_conf_apply);

		if (trigger_enabled() && cluster->triggers_enabled) {
			trigger_args_afrom_server(node->pool, &args, node->name, node->addr.inet.dst_port);
			if (fr_pair_list_empty(&args)) goto error;

			if (!fr_pair_list_empty(&cluster->trigger_args)) MEM(fr_pair_list_copy(cluster, &args, &cluster->trigger_args) >= 0);

			fr_pool_enable_triggers(node->pool, node->cluster->trigger_prefix, &args);

			fr_pair_list_free(&args);
		}

		if (fr_pool_start(node->pool) < 0) goto error;

		return FR_REDIS_CLUSTER_RCODE_SUCCESS;
	}

	/*
	 *	Apply the new config to the possibly live pool
	 */
	if (fr_pool_reconnect(node->pool, NULL) < 0) goto error;

	return FR_REDIS_CLUSTER_RCODE_SUCCESS;
}

/** Parse a -MOVED or -ASK redirect
 *
 * Converts the body of the -MOVED or -ASK error into an IPv4/6 address and port.
 *
 * @param[out] key_slot value extracted from redirect string (may be NULL).
 * @param[out] node_addr Redis node ipaddr and port extracted from redirect string.
 * @param[in] redirect to process.
 * @return
 *	- FR_REDIS_CLUSTER_RCODE_SUCCESS on success.
 *	- FR_REDIS_CLUSTER_RCODE_BAD_INPUT if the server returned an invalid redirect.
 */
static fr_redis_cluster_rcode_t cluster_node_conf_from_redirect(uint16_t *key_slot, fr_socket_t *node_addr,
						       redisReply *redirect)
{
	char		*p, *q;
	unsigned long	key;
	uint16_t	port;
	fr_ipaddr_t	ipaddr;

	if (!redirect || (redirect->type != REDIS_REPLY_ERROR)) {
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	p = redirect->str;
	if (strncmp(REDIS_ERROR_MOVED_STR, redirect->str, sizeof(REDIS_ERROR_MOVED_STR) - 1) == 0) {
		q = p + sizeof(REDIS_ERROR_MOVED_STR);	/* not a typo, skip space too */
	} else if (strncmp(REDIS_ERROR_ASK_STR, redirect->str, sizeof(REDIS_ERROR_ASK_STR) - 1) == 0) {
		q = p + sizeof(REDIS_ERROR_ASK_STR);	/* not a typo, skip space too */
	} else {
		fr_strerror_const("No '-MOVED' or '-ASK' log_prefix");
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}
	if ((size_t)(q - p) >= (size_t)redirect->len) {
		fr_strerror_const("Truncated");
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}
	p = q;
	key = strtoul(p, &q, 10);
	if (key > KEY_SLOTS) {
		fr_strerror_printf("Key %lu outside of redis slot range", key);
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}
	p = q;

	if (*p != ' ') {
		fr_strerror_const("Missing key/host separator");
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}
	p++;			/* Skip the ' ' */

	if (fr_inet_pton_port(&ipaddr, &port, p, redirect->len - (p - redirect->str), AF_UNSPEC, true, true) < 0) {
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}
	fr_assert(ipaddr.af);

	if (key_slot) *key_slot = key;
	if (node_addr) {
		node_addr->inet.dst_ipaddr = ipaddr;
		node_addr->inet.dst_port = port;
	}

	return FR_REDIS_CLUSTER_RCODE_SUCCESS;
}

/** Apply a cluster map received from a cluster node
 *
 * @note Errors may be retrieved with fr_strerror().
 * @note Must be called with the cluster mutex held.
 *
 * Key slot range structure
 @verbatim
   [0] -> key slot range 0
       [0] -> key_slot_start
       [1] -> key_slot_end
       [2] -> master_node
           [0] -> master 0 ip (string)
           [1] -> master 0 port (number)
       [3..n] -> slave_node(s)
   [1] -> key slot range 1)
       [0]  -> key_slot_start
       [1] -> key_slot_end
       [2] -> master_node
           [0] -> master 1 ip (string)
           [1] -> master 1 port (number)
       [3..n] -> slave_node(s)
   [n] -> key slot range n
       [0] -> key_slot_start
       [1] -> key_slot_end
       [2] -> master_node
           [0] -> master n ip (string)
           [1] -> master n port (number)
       [3..n] -> slave_node(s)
 @endverbatim
 *
 * @param[in,out] cluster to apply map to.
 * @param[in] reply from #cluster_map_get.
 * @return
 *	- FR_REDIS_CLUSTER_RCODE_SUCCESS on success.
 *	- FR_REDIS_CLUSTER_RCODE_FAILED on failure.
  *	- FR_REDIS_CLUSTER_RCODE_NO_CONNECTION connection failure.
 *	- FR_REDIS_CLUSTER_RCODE_BAD_INPUT if the map didn't provide nodes for all keyslots.
 */
static fr_redis_cluster_rcode_t cluster_map_apply(fr_redis_cluster_t *cluster, redisReply *reply)
{
	size_t		i;
	uint8_t		r = 0;

	fr_redis_cluster_rcode_t	rcode;

	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.
#ifndef NDEBUG
#  define SET_ADDR(_addr, _map) \
do { \
	int _ret; \
	_ret = fr_inet_pton(&_addr.inet.dst_ipaddr, _map->element[0]->str, _map->element[0]->len, AF_UNSPEC, true, true);\
	fr_assert(_ret == 0);\
	_addr.inet.dst_port = _map->element[1]->integer; \
} while (0)
#else
#  define SET_ADDR(_addr, _map) \
do { \
	fr_inet_pton(&_addr.inet.dst_ipaddr, _map->element[0]->str, _map->element[0]->len, AF_UNSPEC, true, true);\
	_addr.inet.dst_port = _map->element[1]->integer; \
} while (0)
#endif

#define SET_INACTIVE(_node) \
do { \
	(_node)->is_active = false; \
	(_node)->is_master = false; \
	fr_rb_delete(cluster->used_nodes, _node); \
	fr_fifo_push(cluster->free_nodes, _node); \
} while (0)

#define SET_ACTIVE(_node) \
do { \
	(_node)->is_active = true; \
	fr_rb_insert(cluster->used_nodes, _node); \
	fr_fifo_pop(cluster->free_nodes); \
	active[(_node)->id] = true; \
	rollback[r++] = (_node)->id; \
} while (0)

	fr_assert(reply->type == REDIS_REPLY_ARRAY);

	memset(&rollback, 0, sizeof(rollback));
	memset(active, 0, sizeof(active));
	memset(master, 0, sizeof(master));

	cluster->remapping = true;

	/*
	 *	Must be cleared with the mutex held
	 */
	memset(&cluster->key_slot_pending, 0, sizeof(cluster->key_slot_pending));

	/*
	 *	Insert new nodes and markup the keyslot indexes
	 *	in our temporary keyslot_array.
	 *
	 *	A map consists of an array with the following indexes:
	 *	  [0]    -> key_slot_start
	 *	  [1]    -> key_slot_end
	 *	  [2]    -> master_node
	 *	  [3..n] -> slave_node(s)
	 */
	for (i = 0; i < reply->elements; i++) {
		size_t			j;
		long long int		k;
		int			slaves = 0;
		fr_redis_cluster_node_t		*found, *spare;
		fr_redis_cluster_node_t		find = {}; /* Initialise unused fields to stop Coverity complaints */
		fr_redis_cluster_key_slot_t	tmpl_slot;
		redisReply		*map = reply->element[i];

		memset(&tmpl_slot, 0, sizeof(tmpl_slot));

		SET_ADDR(find.addr, map->element[2]);
		found = fr_rb_find(cluster->used_nodes, &find);
		if (found) {
			active[found->id] = true;
			goto reuse_master_node;
		}

		/*
		 *	Process the master
		 *
		 *      A master node consists of any array with the following indexes:
		 *	  [0] -> node ip (as string)
		 *	  [1] -> node port
		 */
		spare = fr_fifo_peek(cluster->free_nodes);
		if (!spare) {
		out_of_nodes:
			fr_strerror_const("Reached maximum connected nodes");
			rcode = FR_REDIS_CLUSTER_RCODE_FAILED;
		error:
			cluster->remapping = false;
			cluster->last_updated = fr_time();
			/* Re-insert new nodes back into the free_nodes queue */
			for (i = 0; i < r; i++) SET_INACTIVE(&cluster->node[rollback[i]]);
			return rcode;
		}

		spare->pending_addr = find.addr;
		rcode = cluster_node_connect(cluster, spare);
		if (rcode < 0) goto error;

		/*
		 *	Check to see if the node we just configured
		 *	already exists in the tree.  If it does we
		 *	use that, else we add it to the array of
		 *	nodes to rollback on failure.
		 */
		SET_ACTIVE(spare);
		found = spare;

	reuse_master_node:
		tmpl_slot.master = found->id;
		master[found->id] = true;	/* Mark this node as a master */

		/*
		 *	Process the slaves
		 *
		 *      A slave node consists of any array with the following indexes:
		 *	  [0] -> node ip (as string)
		 *	  [1] -> node port
		 */
		for (j = 3; (j < map->elements); j++) {
			SET_ADDR(find.addr, map->element[j]);
			found = fr_rb_find(cluster->used_nodes, &find);
			if (found) {
				active[found->id] = true;
				goto next;
			}

			spare = fr_fifo_peek(cluster->free_nodes);
			if (!spare) goto out_of_nodes;

			spare->pending_addr = find.addr;
			if (cluster_node_connect(cluster, spare) < 0) continue;	/* Slave failure is non-fatal */

			SET_ACTIVE(spare);
			found = spare;

		next:
			tmpl_slot.slave[slaves++] = found->id;

			/* Hit the maximum number of slaves we allow */
			if (slaves >= MAX_SLAVES) break;
		}
		tmpl_slot.slave_num = slaves;

		/*
		 *	Copy our tmpl key slot to each of the key slots
		 *	specified by the range for this map.
		 */
		for (k = map->element[0]->integer; k <= map->element[1]->integer; k++) {
			memcpy(&cluster->key_slot_pending[k], &tmpl_slot,
			       sizeof(*(cluster->key_slot_pending)));
		}
	}

	/*
	 *	Check for holes in the pending_addr key_slot array
	 *
	 *	The cluster specification says that upon
	 *	detecting a 'NULL' key_slot we should
	 *	check again to see if the cluster error has
	 *	been resolved, but seeing as we're in the
	 *	middle of updating the cluster from very
	 *	recent output of 'cluster slots' it's best to
	 *	error out.
	 */
	for (i = 0; i < KEY_SLOTS; i++) {
		if (cluster->key_slot_pending[i].master == 0) {
			fr_strerror_printf("Cluster is misconfigured, no node assigned for key %zu", i);
			rcode = FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
			goto error;
		}
	}

	/*
	 *	We have connections/pools for all the nodes in
	 *	the new map, apply it to the live cluster.
	 *
	 *	Other workers may be using the key slot table,
	 *	but that's ok. Nodes and pools are never freed,
	 *	so the worst that will happen, is they'll hit
	 *	the wrong node for the key, and get redirected.
	 */
	memcpy(&cluster->key_slot, &cluster->key_slot_pending, sizeof(cluster->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 < cluster->conf->max_nodes; i++) {
#ifndef NDEBUG
		fr_redis_cluster_node_t *found;

		if (cluster->node[i].is_active) {
			/* Sanity check for duplicates that are active */
			found = fr_rb_find(cluster->used_nodes, &cluster->node[i]);
			fr_assert(found);
			fr_assert(found->is_active);
			fr_assert(found->id == i);
		}
#endif

		if (!active[i] && cluster->node[i].is_active) {
			SET_INACTIVE(&cluster->node[i]);	/* Sets is_master = false */

		/*
		 *	Only change the masters once we've successfully
		 *	remapped the cluster.
		 */
		} else if (master[i]) {
			cluster->node[i].is_master = true;
		}
	}

	cluster->remapping = false;
	cluster->last_updated = fr_time();

	/*
	 *	Sanity checks
	 */
	fr_assert(((talloc_array_length(cluster->node) - 1) - fr_rb_num_elements(cluster->used_nodes)) ==
		   fr_fifo_num_elements(cluster->free_nodes));

	return FR_REDIS_CLUSTER_RCODE_SUCCESS;
}

/** Validate a cluster map node entry
 *
 * @note Errors may be retrieved with fr_strerror().
 * @note In a separate function, as it's called for both master and slave nodes.
 *
 * @param[in] node we're validating.
 * @param[in] map_idx we're processing.
 * @param[in] node_idx we're processing.
 * @return
 *	- FR_REDIS_CLUSTER_RCODE_SUCCESS on success.
 *	- FR_REDIS_CLUSTER_RCODE_BAD_INPUT on validation failure (bad data returned from Redis).
 */
static int cluster_map_node_validate(redisReply *node, int map_idx, int node_idx)
{
	fr_ipaddr_t ipaddr;

	if (node->type != REDIS_REPLY_ARRAY) {
		fr_strerror_printf("Cluster map %i node %i is wrong type, expected array got %s",
				   map_idx, node_idx,
				   fr_table_str_by_value(redis_reply_types, node->element[1]->type, "<UNKNOWN>"));
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	/*
	 *  As per the redis docs: https://redis.io/commands/cluster-slots
	 *
	 *  Newer versions of Redis Cluster will output, for each Redis instance,
	 *  not just the IP and port, but also the node ID as third element of the
	 *  array. In future versions there could be more elements describing the
	 *  node better. In general a client implementation should just rely on
	 *  the fact that certain parameters are at fixed positions as specified,
	 *  but more parameters may follow and should be ignored.
	 *  Similarly a client library should try if possible to cope with the fact
	 *  that older versions may just have the IP and port parameter.
	 */
	if (node->elements < 2) {
		fr_strerror_printf("Cluster map %i node %i has incorrect number of elements, expected at least "
				   "2 got %zu", map_idx, node_idx, node->elements);
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	if (node->element[0]->type != REDIS_REPLY_STRING) {
		fr_strerror_printf("Cluster map %i node %i ip address is wrong type, expected string got %s",
				   map_idx, node_idx,
				   fr_table_str_by_value(redis_reply_types, node->element[0]->type, "<UNKNOWN>"));
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	if (fr_inet_pton(&ipaddr, node->element[0]->str, node->element[0]->len, AF_UNSPEC, true, true) < 0) {
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	if (node->element[1]->type != REDIS_REPLY_INTEGER) {
		fr_strerror_printf("Cluster map %i node %i port is wrong type, expected integer got %s",
				   map_idx, node_idx,
				   fr_table_str_by_value(redis_reply_types, node->element[1]->type, "<UNKNOWN>"));
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	if (node->element[1]->integer < 0) {
		fr_strerror_printf("Cluster map %i node %i port is too low, expected >= 0 got %lli",
				   map_idx, node_idx, node->element[1]->integer);
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	if (node->element[1]->integer > UINT16_MAX) {
		fr_strerror_printf("Cluster map %i node %i port is too high, expected <= " STRINGIFY(UINT16_MAX)" "
				   "got %lli", map_idx, node_idx, node->element[1]->integer);
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	return FR_REDIS_CLUSTER_RCODE_SUCCESS;
}

/** Learn a new cluster layout by querying the node that issued the -MOVE
 *
 * Also validates the response from the Redis cluster, so we can be sure that
 * it's well formed, before doing more expensive operations.
 *
 * @note Errors may be retrieved with fr_strerror().
 *
 * @param[out] out Where to write cluster map.
 * @param[in] conn to use for learning the new cluster map.
 * @return
 *	- FR_REDIS_CLUSTER_RCODE_IGNORED if 'cluster slots' returned an error (indicating clustering not supported).
 *	- FR_REDIS_CLUSTER_RCODE_SUCCESS on success.
 *	- FR_REDIS_CLUSTER_RCODE_FAILED if issuing the command resulted in an error.
 *	- FR_REDIS_CLUSTER_RCODE_NO_CONNECTION connection failure.
 *	- FR_REDIS_CLUSTER_RCODE_BAD_INPUT on validation failure (bad data returned from Redis).
 */
static fr_redis_cluster_rcode_t cluster_map_get(redisReply **out, fr_redis_conn_t *conn)
{
	redisReply	*reply;
	size_t		i = 0;

	*out = NULL;

	reply = redisCommand(conn->handle, "cluster slots");
	switch (fr_redis_command_status(conn, reply)) {
	case REDIS_RCODE_RECONNECT:
		fr_redis_reply_free(&reply);
		fr_strerror_const("No connections available");
		return FR_REDIS_CLUSTER_RCODE_NO_CONNECTION;

	case REDIS_RCODE_ERROR:
	default:
		if (reply && reply->type == REDIS_REPLY_ERROR) {
			fr_strerror_printf("%.*s", (int)reply->len, reply->str);
			fr_redis_reply_free(&reply);
			return FR_REDIS_CLUSTER_RCODE_IGNORED;
		}
		fr_strerror_const("Unknown client error");
		return FR_REDIS_CLUSTER_RCODE_FAILED;

	case REDIS_RCODE_SUCCESS:
		break;
	}

	if (reply->type != REDIS_REPLY_ARRAY) {
		fr_strerror_printf("Bad response to \"cluster slots\" command, expected array got %s",
				   fr_table_str_by_value(redis_reply_types, reply->type, "<UNKNOWN>"));
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	/*
	 *	Clustering configured but no slots set
	 */
	if (reply->elements == 0) {
		fr_strerror_printf("Empty response to \"cluster slots\" command (zero length array)");
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	/*
	 *	Validate the complete map set before returning.
	 */
	for (i = 0; i < reply->elements; i++) {
		size_t		j;
		redisReply	*map;

		map = reply->element[i];
		if (map->type != REDIS_REPLY_ARRAY) {
			fr_strerror_printf("Cluster map %zu is wrong type, expected array got %s",
				   	   i, fr_table_str_by_value(redis_reply_types, map->type, "<UNKNOWN>"));
		error:
			fr_redis_reply_free(&reply);
			return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
		}

		if (map->elements < 3) {
			fr_strerror_printf("Cluster map %zu has too few elements, expected at least 3, got %zu",
					   i, map->elements);
			goto error;
		}

		/*
		 *	Key slot start
		 */
		if (map->element[0]->type != REDIS_REPLY_INTEGER) {
			fr_strerror_printf("Cluster map %zu key slot start is wrong type, expected integer got %s",
					   i, fr_table_str_by_value(redis_reply_types, map->element[0]->type, "<UNKNOWN>"));
			goto error;
		}

		if (map->element[0]->integer < 0) {
			fr_strerror_printf("Cluster map %zu key slot start is too low, expected >= 0 got %lli",
					   i, map->element[0]->integer);
			goto error;
		}

		if (map->element[0]->integer > KEY_SLOTS) {
			fr_strerror_printf("Cluster map %zu key slot start is too high, expected <= "
					   STRINGIFY(KEY_SLOTS) " got %lli", i, map->element[0]->integer);
			goto error;
		}

		/*
		 *	Key slot end
		 */
		if (map->element[1]->type != REDIS_REPLY_INTEGER) {
			fr_strerror_printf("Cluster map %zu key slot end is wrong type, expected integer got %s",
					   i, fr_table_str_by_value(redis_reply_types, map->element[1]->type, "<UNKNOWN>"));
			goto error;
		}

		if (map->element[1]->integer < 0) {
			fr_strerror_printf("Cluster map %zu key slot end is too low, expected >= 0 got %lli",
					   i, map->element[1]->integer);
			goto error;
		}

		if (map->element[1]->integer > KEY_SLOTS) {
			fr_strerror_printf("Cluster map %zu key slot end is too high, expected <= "
					   STRINGIFY(KEY_SLOTS) " got %lli", i, map->element[1]->integer);
			goto error;
		}

		if (map->element[1]->integer < map->element[0]->integer) {
			fr_strerror_printf("Cluster map %zu key slot start/end out of order.  "
					   "Start was %lli, end was %lli", i, map->element[0]->integer,
					   map->element[1]->integer);
			goto error;
		}

		/*
		 *	Master node
		 */
		if (cluster_map_node_validate(map->element[2], i, 0) < 0) goto error;

		/*
		 *	Slave nodes
		 */
		for (j = 3; j < map->elements; j++) {
			if (cluster_map_node_validate(map->element[j], i, j - 2) < 0) goto error;
		}
	}
	*out = reply;

	return FR_REDIS_CLUSTER_RCODE_SUCCESS;
}

/** Perform a runtime remap of the cluster
 *
 * @note Errors may be retrieved with fr_strerror().
 * @note Must be called with the cluster mutex free.
 *
 * @param[in] request The current request.
 * @param[in,out] cluster to remap.
 * @param[in] conn to use to query the cluster.
 * @return
 *	- FR_REDIS_CLUSTER_RCODE_IGNORED if 'cluster slots' returned an error (indicating clustering not supported).
 *	- FR_REDIS_CLUSTER_RCODE_SUCCESS on success.
 *	- FR_REDIS_CLUSTER_RCODE_FAILED if issuing the 'cluster slots' command resulted in a protocol error.
 *	- FR_REDIS_CLUSTER_RCODE_NO_CONNECTION connection failure.
 *	- FR_REDIS_CLUSTER_RCODE_BAD_INPUT on validation failure (bad data returned from Redis).
 */
fr_redis_cluster_rcode_t fr_redis_cluster_remap(request_t *request, fr_redis_cluster_t *cluster, fr_redis_conn_t *conn)
{
	fr_time_t	now;
	redisReply	*map;
	fr_redis_cluster_rcode_t	ret;
	size_t		i, j;

	/*
	 *	If the cluster was remapped very recently, or is being
	 *	remapped it's unlikely that it needs remapping again.
	 */
	if (cluster->remapping) {
	in_progress:
		ROPTIONAL(RDEBUG2, DEBUG2, "Cluster remapping in progress, ignoring remap request");
		return FR_REDIS_CLUSTER_RCODE_IGNORED;
	}

	/*
	 *	The remap times are _our_ times, not the _request_ time.
	 */
	now = fr_time();
	if (fr_time_to_sec(now) == fr_time_to_sec(cluster->last_updated)) {
	too_soon:
		ROPTIONAL(RWARN, WARN, "Cluster was updated less than a second ago, ignoring remap request");
		return FR_REDIS_CLUSTER_RCODE_IGNORED;
	}

	ROPTIONAL(RINFO, INFO, "Initiating cluster remap");

	/*
	 *	Get new cluster information
	 */
	ret = cluster_map_get(&map, conn);
	switch (ret) {
	case FR_REDIS_CLUSTER_RCODE_BAD_INPUT:		/* Validation error */
	case FR_REDIS_CLUSTER_RCODE_NO_CONNECTION:		/* Connection error */
	case FR_REDIS_CLUSTER_RCODE_FAILED:			/* Error issuing command */
		return ret;

	case FR_REDIS_CLUSTER_RCODE_IGNORED:		/* Clustering not enabled, or not supported */
		cluster->remap_needed = false;
		return FR_REDIS_CLUSTER_RCODE_IGNORED;

	case FR_REDIS_CLUSTER_RCODE_SUCCESS:		/* Success */
		break;
	}

	/*
	 *	Print the mapping we received
	 */
	ROPTIONAL(RINFO, INFO, "Cluster map consists of %zu key ranges", map->elements);
	for (i = 0; i < map->elements; i++) {
		redisReply *map_node = map->element[i];

		ROPTIONAL(RINFO, INFO, "%zu - keys %lli-%lli", i,
			  map_node->element[0]->integer,
			  map_node->element[1]->integer);

		if (request) RINDENT();
		ROPTIONAL(RINFO, INFO, "master: %s:%lli",
			  map_node->element[2]->element[0]->str,
			  map_node->element[2]->element[1]->integer);
		for (j = 3; j < map_node->elements; j++) {
			ROPTIONAL(RINFO, INFO, "slave%zu: %s:%lli", j - 3,
				  map_node->element[j]->element[0]->str,
				  map_node->element[j]->element[1]->integer);
		}
		if (request) REXDENT();
	}

	/*
	 *	Check again that the cluster isn't being
	 *	remapped, or was remapped too recently,
	 *	now we hold the mutex and the state of
	 *	those variables is synchronized.
	 */
	pthread_mutex_lock(&cluster->mutex);
	if (cluster->remapping) {
		pthread_mutex_unlock(&cluster->mutex);
		fr_redis_reply_free(&map);	/* Free the map */
		goto in_progress;
	}
	if (fr_time_to_sec(now) == fr_time_to_sec(cluster->last_updated)) {
		pthread_mutex_unlock(&cluster->mutex);
		fr_redis_reply_free(&map);	/* Free the map */
		goto too_soon;
	}
	ret = cluster_map_apply(cluster, map);
	if (ret == FR_REDIS_CLUSTER_RCODE_SUCCESS) cluster->remap_needed = false;	/* Change on successful remap */
	pthread_mutex_unlock(&cluster->mutex);

	fr_redis_reply_free(&map);	/* Free the map */
	if (ret < 0) return FR_REDIS_CLUSTER_RCODE_FAILED;

	return FR_REDIS_CLUSTER_RCODE_SUCCESS;
}

/** Retrieve or associate a node with the server indicated in the redirect
 *
 * @note Errors may be retrieved with fr_strerror().
 *
 * @param[out] out Where to write the node representing the redirect server.
 * @param[in] cluster to draw node from.
 * @param[in] reply Redis reply containing the redirect information.
 * @return
 *	- FR_REDIS_CLUSTER_RCODE_SUCCESS on success.
 *	- FR_REDIS_CLUSTER_RCODE_FAILED no more nodes available.
 *	- FR_REDIS_CLUSTER_RCODE_NO_CONNECTION connection failure.
 *	- FR_REDIS_CLUSTER_RCODE_BAD_INPUT on validation failure (bad data returned from Redis).
 */
static fr_redis_cluster_rcode_t cluster_redirect(fr_redis_cluster_node_t **out, fr_redis_cluster_t *cluster, redisReply *reply)
{
	fr_redis_cluster_node_t		find, *found, *spare;
	fr_redis_conn_t		*rconn;

	uint16_t		key;

	memset(&find, 0, sizeof(find));

	*out = NULL;

	if (cluster_node_conf_from_redirect(&key, &find.addr, reply) < 0) return FR_REDIS_CLUSTER_RCODE_FAILED;

	pthread_mutex_lock(&cluster->mutex);
	/*
	 *	If we have already have a pool for the
	 *	host we were redirected to, use that.
	 */
	found = fr_rb_find(cluster->used_nodes, &find);
	if (found) {
		/* We have the new pool, don't need to hold the lock */
		pthread_mutex_unlock(&cluster->mutex);
		*out = found;
		return FR_REDIS_CLUSTER_RCODE_SUCCESS;
	}

	/*
	 *	Otherwise grab a free node and try and connect
	 *	it to the server we were redirected to.
	 */
	spare = fr_fifo_peek(cluster->free_nodes);
	if (!spare) {
		fr_strerror_const("Reached maximum connected nodes");
		pthread_mutex_unlock(&cluster->mutex);
		return FR_REDIS_CLUSTER_RCODE_FAILED;
	}
	spare->pending_addr = find.addr;	/* Set the config to be applied */
	if (cluster_node_connect(cluster, spare) < 0) {
		pthread_mutex_unlock(&cluster->mutex);
		return FR_REDIS_CLUSTER_RCODE_NO_CONNECTION;
	}
	fr_rb_insert(cluster->used_nodes, spare);
	fr_fifo_pop(cluster->free_nodes);
	found = spare;

	/* We have the new pool, don't need to hold the lock */
	pthread_mutex_unlock(&cluster->mutex);

	/*
	 *	Determine if we can establish a connection to
	 *	the new pool, to check if it's viable.
	 */
	rconn = fr_pool_connection_get(found->pool, NULL);
	if (!rconn) {
		/*
		 *	To prevent repeated misconfigurations
		 *	using all free nodes, add the node
		 *	back to the spare queue if this
		 *	was the first connection attempt and
		 *	it failed.
		 */
		pthread_mutex_lock(&cluster->mutex);
		fr_fifo_push(cluster->free_nodes, spare);
		pthread_mutex_unlock(&cluster->mutex);

		fr_strerror_const("No connections available");
		return FR_REDIS_CLUSTER_RCODE_NO_CONNECTION;
	}
	fr_pool_connection_release(found->pool, NULL, rconn);
	*out = found;

	return FR_REDIS_CLUSTER_RCODE_SUCCESS;
}


/** Try to determine the health of a cluster node passively by examining its pool state
 *
 * Returns an integer value representing the likelihood that the pool is live.
 * Range is between 1 and 11,000.
 *
 * If a weight of 1 is returned, connections from the pool should be checked
 * (by pinging) before use.
 *
 * @param now The current time.
 * @param state of the connection pool.
 * @return
 *	- 1 the pool is very likely to be bad.
 *	- 2-11000 the pool is likely to be good, with a higher number
 *	  indicating higher probability of liveness.
 */
static int cluster_node_pool_health(fr_time_t now, fr_pool_state_t const *state)
{
	/*
	 *	Failed spawn recently, probably bad
	 */
	if (fr_time_delta_to_msec(fr_time_sub(now, state->last_failed)) < FAILED_PERIOD) return FAILED_WEIGHT;

	/*
	 *	Closed recently, probably bad
	 */
	if (fr_time_delta_to_msec(fr_time_sub(now, state->last_closed)) < CLOSED_PERIOD) return CLOSED_WEIGHT;

	/*
	 *	Released too long ago, don't know
	 */
	if (fr_time_delta_to_msec(fr_time_sub(now, state->last_released)) > RELEASED_PERIOD) return RELEASED_MIN_WEIGHT;

	/*
	 *	Released not long ago, might be ok.
	 */
	return RELEASED_MIN_WEIGHT + (RELEASED_PERIOD - fr_time_delta_to_msec(fr_time_sub(now, state->last_released)));
}

/** Issue a ping request against a cluster node
 *
 * Establishes whether the connection to the node we have is live.
 *
 * @param request The current request.
 * @param node to ping.
 * @param conn the connection to ping on.
 * @return
 *	- FR_REDIS_CLUSTER_RCODE_BAD_INPUT if we got a bad response.
 *	- FR_REDIS_CLUSTER_RCODE_SUCCESS on success.
 *	- FR_REDIS_CLUSTER_RCODE_NO_CONNECTION on connection down.
 */
static fr_redis_cluster_rcode_t cluster_node_ping(request_t *request, fr_redis_cluster_node_t *node, fr_redis_conn_t *conn)
{
	redisReply		*reply;
	fr_redis_rcode_t	rcode;

	ROPTIONAL(RDEBUG2, DEBUG2, "[%i] Executing command: PING", node->id);
	reply = redisCommand(conn->handle, "PING");
	rcode = fr_redis_command_status(conn, reply);
	if (rcode != REDIS_RCODE_SUCCESS) {
		ROPTIONAL(RPERROR, PERROR, "[%i] PING failed to %s:%i", node->id, node->name, node->addr.inet.dst_port);
		fr_redis_reply_free(&reply);
		return FR_REDIS_CLUSTER_RCODE_NO_CONNECTION;
	}

	if (reply->type != REDIS_REPLY_STATUS) {
		ROPTIONAL(RERROR, ERROR, "[%i] Bad PING response from %s:%i, expected status got %s",
			  node->id, node->name, node->addr.inet.dst_port,
			  fr_table_str_by_value(redis_reply_types, reply->type, "<UNKNOWN>"));
		fr_redis_reply_free(&reply);
		return FR_REDIS_CLUSTER_RCODE_BAD_INPUT;
	}

	ROPTIONAL(RDEBUG2, DEBUG2, "[%i] Got response: %s", node->id, reply->str);
	fr_redis_reply_free(&reply);
	return FR_REDIS_CLUSTER_RCODE_SUCCESS;
}

/** Attempt to find a live pool in the cluster
 *
 * The intent here is to find pools/nodes where a connection was released the shortest
 * time ago.  Having a connection be released (vs closed) indicates that the pool is live.
 *
 * We don't want to have all workers try and grab a connection to this node however, as it
 * may still be dead (we don't know).
 *
 * So we use an inverse transform sample, to weight the nodes, based on time between now
 * and when the connection was released.  Connections released closest to the current
 * time are given a higher weighting.
 *
 * Weight range is between 1 - 11,000.
 *
 * - If released > 10.0 seconds ago,information is not valid, weight 500.
 * - If closed < 10.0 seconds ago, it's a bad pool, weight 1.
 * - If spawn failed < 10.0 seconds ago, it's a bad pool, weight 1.
 * - If a connection was released 0.0 seconds ago, weight 11,000.
 * - If a connection was released 10.0 seconds ago, weight 1000.
 *
 * Using the above algorithm we use the experience of other workers using the cluster to
 * inform our alternative node selection.
 *
 * Suggestions on improving live node selection appreciated.
 *
 * Inverse transform sampling based roughly on the solution from this post:
 *    http://stackoverflow.com/questions/17250568/randomly-choosing-from-a-list-with-weighted-probabilities
 *
 * Wikipedia page here:
 *    https://en.wikipedia.org/wiki/Inverse_transform_sampling
 *
 * @note Must be called with the cluster mutex free.
 *
 * @param[out] live_node we found.
 * @param[out] live_conn to that node.
 * @param[in] request The current request (used for logging).
 * @param[in] cluster to search for live pools in.
 * @param[in] skip this node (it's bad).
 * @return 0 (iterates over the whole tree).
 */
static int cluster_node_find_live(fr_redis_cluster_node_t **live_node, fr_redis_conn_t **live_conn,
				  request_t *request, fr_redis_cluster_t *cluster, fr_redis_cluster_node_t *skip)
{
	uint32_t		i;

	cluster_nodes_live_t	*live;
	fr_time_t		now;
	fr_rb_iter_inorder_t	iter;
	fr_redis_cluster_node_t		*node;

	ROPTIONAL(RDEBUG2, DEBUG2, "Searching for live cluster nodes");

	if (fr_rb_num_elements(cluster->used_nodes) == 1) {
	no_alts:
		ROPTIONAL(RERROR, ERROR, "No alternative nodes available");
		return -1;
	}

	live = talloc_zero(NULL, cluster_nodes_live_t);	/* Too big for stack */
	live->skip = skip->id;

	pthread_mutex_lock(&cluster->mutex);
	for (node = fr_rb_iter_init_inorder(&iter, cluster->used_nodes);
	     node;
	     node = fr_rb_iter_next_inorder(&iter)) {
		fr_assert(node->pool);
		if (live->skip == node->id) continue;	/* Skip dead nodes */

		live->node[live->next].pool_state = fr_pool_state(node->pool);
		live->node[live->next++].id = node->id;
	}
	pthread_mutex_unlock(&cluster->mutex);

	fr_assert(live->next);			/* There should be at least one */
	if (live->next == 1) goto no_alts;	/* Weird, but conceivable */

	now = fr_time();

	/*
	 *	Weighted random selection
	 */
	for (i = 0; (i < cluster->conf->max_alt) && live->next; i++) {
		fr_redis_conn_t 	*conn;
		uint8_t			j;
		int			first, last, pivot;	/* Must be signed for BS */
		unsigned int		find, cumulative = 0;

		ROPTIONAL(RDEBUG3, DEBUG2, "(Re)assigning node weights:");
		if (request) RINDENT();
		for (j = 0; j < live->next; j++) {
			int weight;

			weight = cluster_node_pool_health(now, live->node[j].pool_state);
			ROPTIONAL(RDEBUG3, DEBUG3, "Node %i weight: %i", live->node[j].id, weight);
			live->node[j].cumulative = (cumulative += weight);
		}
		if (request) REXDENT();

		/*
		 *	Select a node at random
		 */
		find = (fr_rand() & (cumulative - 1));	/* Between 1 and total */
		first = 0;
		last = live->next - 1;
		pivot = (first + last) / 2;

		while (first <= last) {
			if (live->node[pivot].cumulative < find) {
				first = pivot + 1;
			} else if (live->node[pivot].cumulative == find) {
				break;
			} else {
				last = pivot - 1;
			}
			pivot = (first + last) / 2;
		}
		/*
		 *	Round up...
		 */
		if (first > last) pivot = last + 1;

		/*
		 *	Resolve the index to the actual node.  We use IDs
		 *	to save memory...
		 */
		node = &cluster->node[live->node[pivot].id];
		fr_assert(live->node[pivot].id == node->id);

		ROPTIONAL(RDEBUG2, DEBUG2, "Selected node %i (using random value %i)", node->id, find);
		conn = fr_pool_connection_get(node->pool, request);
		if (!conn) {
			ROPTIONAL(RERROR, ERROR, "No connections available to node %i %s:%i", node->id,
				  node->name, node->addr.inet.dst_port);
		next:
			/*
			 *	Remove the node we just discovered was bad
			 *	out of the set of nodes we're selecting over.
			 */
			if (pivot == live->next) {
				live->next--;
				continue;
			}
			memcpy(&live->node[pivot], &live->node[live->next - 1], sizeof(live->node[pivot]));
			live->next--;
			continue;
		}

		/*
		 *	PING! PONG?
		 */
		switch (cluster_node_ping(request, node, conn)) {
		case FR_REDIS_CLUSTER_RCODE_SUCCESS:
			break;

		case FR_REDIS_CLUSTER_RCODE_NO_CONNECTION:
			fr_pool_connection_close(node->pool, request, conn);
			goto next;

		default:
			fr_pool_connection_release(node->pool, request, conn);
			goto next;
		}

		*live_node = node;
		*live_conn = conn;
		talloc_free(live);

		return 0;
	}

	ROPTIONAL(RERROR, ERROR, "Hit max alt limit %i, and no live connections found", cluster->conf->max_alt);
	talloc_free(live);

	return -1;
}

/** Callback for freeing a Redis connection
 *
 * @param[in] conn to free.
 * @return 0.
 */
static int _cluster_conn_free(fr_redis_conn_t *conn)
{
	redisFree(conn->handle);

	return 0;
}

/** Create a new connection to a Redis node
 *
 * @param[in] ctx to allocate connection structure in. Will be freed at the same time as the pool.
 * @param[in] instance data of type #fr_redis_cluster_node_t. Holds parameters for establishing new connection.
 * @param[in] timeout The maximum time allowed to complete the connection.
 * @return
 *	- New #fr_redis_conn_t on success.
 *	- NULL on failure.
 */
void *fr_redis_cluster_conn_create(TALLOC_CTX *ctx, void *instance, fr_time_delta_t timeout)
{
	fr_redis_cluster_node_t	*node = instance;
	fr_redis_conn_t		*conn = NULL;
	redisContext		*handle;
	redisReply		*reply = NULL;
	char const		*log_prefix = node->cluster->log_prefix;

	DEBUG2("%s - [%i] Connecting to node %s:%i", log_prefix, node->id, node->name, node->addr.inet.dst_port);

	handle = redisConnectWithTimeout(node->name, node->addr.inet.dst_port, fr_time_delta_to_timeval(timeout));
	if ((handle != NULL) && handle->err) {
		ERROR("%s - [%i] Connection failed: %s", log_prefix, node->id, handle->errstr);
		redisFree(handle);
		return NULL;
	} else if (!handle) {
		ERROR("%s - [%i] Connection failed", log_prefix, node->id);
		return NULL;
	}

	conn = talloc_zero(ctx, fr_redis_conn_t);
	conn->handle = handle;
	conn->node = node;
	talloc_set_destructor(conn, _cluster_conn_free);

#ifdef HAVE_REDIS_SSL
	if (node->cluster->ssl_ctx != NULL) {
		fr_tls_session_t *tls_session = fr_tls_session_alloc_client(conn, node->cluster->ssl_ctx);
		if (!tls_session) {
			fr_tls_strerror_printf("%s - [%i]", log_prefix, node->id);
			ERROR("%s - [%i] Failed to allocate TLS session", log_prefix, node->id);
			talloc_free(conn);
			return NULL;
		}

		// redisInitiateSSL() takes ownership of SSL object on success
		SSL_up_ref(tls_session->ssl);
		if (redisInitiateSSL(handle, tls_session->ssl) != REDIS_OK) {
			ERROR("%s - [%i] Failed to initiate SSL: %s", log_prefix, node->id, handle->errstr);
			SSL_free(tls_session->ssl);
			talloc_free(conn);
			return NULL;
		}
	}
#endif

	if (node->cluster->conf->password) {
		if (node->cluster->conf->username) {
			DEBUG3("%s - [%i] Executing: AUTH %s %s", log_prefix, node->id,
			       node->cluster->conf->username,
			       node->cluster->conf->password);
			reply = redisCommand(handle, "AUTH %s %s",
					     node->cluster->conf->username,
					     node->cluster->conf->password);
		} else {
			DEBUG3("%s - [%i] Executing: AUTH %s", log_prefix, node->id, node->cluster->conf->password);
			reply = redisCommand(handle, "AUTH %s", node->cluster->conf->password);
		}
		if (!reply) {
			ERROR("%s - [%i] Failed authenticating: %s", log_prefix, node->id, handle->errstr);
		error:
			if (reply) fr_redis_reply_free(&reply);
			talloc_free(conn);
			return NULL;
		}

		switch (reply->type) {
		case REDIS_REPLY_STATUS:
			if (strcmp(reply->str, "OK") != 0) {
				ERROR("%s - [%i] Failed authenticating: %s", log_prefix,
				      node->id, reply->str);
				goto error;
			}
			fr_redis_reply_free(&reply);
			break;	/* else it's OK */

		case REDIS_REPLY_ERROR:
			ERROR("%s - [%i] Failed authenticating: %s", log_prefix, node->id, reply->str);
			goto error;

		default:
			ERROR("%s - [%i] Unexpected reply of type %s to AUTH", log_prefix, node->id,
			      fr_table_str_by_value(redis_reply_types, reply->type, "<UNKNOWN>"));
			goto error;
		}
	}

	if (node->cluster->conf->database) {
		DEBUG3("%s - [%i] Executing: SELECT %i", log_prefix, node->id, node->cluster->conf->database);
		reply = redisCommand(handle, "SELECT %i", node->cluster->conf->database);
		if (!reply) {
			ERROR("%s - [%i] Failed selecting database %i: %s", log_prefix, node->id,
			      node->cluster->conf->database, handle->errstr);
			goto error;
		}

		switch (reply->type) {
		case REDIS_REPLY_STATUS:
			if (strcmp(reply->str, "OK") != 0) {
				ERROR("%s - [%i] Failed selecting database %i: %s", log_prefix, node->id,
				      node->cluster->conf->database, reply->str);
				goto error;
			}
			fr_redis_reply_free(&reply);
			break;	/* else it's OK */

		case REDIS_REPLY_ERROR:
			ERROR("%s - [%i] Failed selecting database %i: %s", log_prefix, node->id,
			      node->cluster->conf->database, reply->str);
			goto error;

		default:
			ERROR("%s - [%i] Unexpected reply of type %s, to SELECT", log_prefix, node->id,
			      fr_table_str_by_value(redis_reply_types, reply->type, "<UNKNOWN>"));
			goto error;
		}
	}

	return conn;
}

/** Implements the key slot selection scheme used by freeradius
 *
 * Like the scheme in the clustering specification but with some differences
 * if the key is NULL or zero length, then a random keyslot is chosen.
 *
 * If there's only a single node in the cluster, then we avoid the CRC16
 * and just use key slot 0.
 *
 * @param cluster to determine key slot for.
 * @param request The current request.
 * @param key the key to resolve.
 * @param key_len the length of the key.
 * @return pointer to key slot key resolves to.
 */
fr_redis_cluster_key_slot_t const *fr_redis_cluster_slot_by_key(fr_redis_cluster_t *cluster, request_t *request,
								uint8_t const *key, size_t key_len)
{
	fr_redis_cluster_key_slot_t *key_slot;

	if (!key || (key_len == 0)) {
		key_slot = &cluster->key_slot[(uint16_t)(fr_rand() & (KEY_SLOTS - 1))];
		ROPTIONAL(RDEBUG2, DEBUG2, "Key rand() -> slot %zu", key_slot - cluster->key_slot);

		return key_slot;
	}

	/*
	 *	Avoid CRC16 if we're operating with one cluster node or
	 *	without clustering.
	 */
	if (fr_rb_num_elements(cluster->used_nodes) > 1) {
		key_slot = &cluster->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 - cluster->key_slot);

		return key_slot;
	}
	ROPTIONAL(RDEBUG3, DEBUG3, "Single node available, skipping key selection");

	return &cluster->key_slot[0];
}

/** Return the master node that would be used for a particular key
 *
 * @param[in] cluster		To resolve key 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_cluster_node_t const *fr_redis_cluster_master(fr_redis_cluster_t *cluster,
						       fr_redis_cluster_key_slot_t const *key_slot)
{
	return &cluster->node[key_slot->master];
}

/** Return the slave node that would be used for a particular key
 *
 * @param[in] cluster		To resolve key in.
 * @param[in] key_slot		To resolve to node.
 * @param[in] slave_num		0..n.
 * @return
 *	- A slave node.
 *	- NULL if no slave node is assigned, or is at the specific key slot.
 *
 */
fr_redis_cluster_node_t const *fr_redis_cluster_slave(fr_redis_cluster_t *cluster,
						      fr_redis_cluster_key_slot_t const *key_slot,
						      uint8_t slave_num)
{
	if (slave_num >= key_slot->slave_num) return NULL;	/* No slave available */

	return &cluster->node[key_slot->slave[slave_num]];
}

/** Return the ipaddr of a particular node
 *
 * @param[out] out	Ipaddr of the node.
 * @param[in] node	to get ip address from.
 * @return
 *	- 0 on success.
 *	- -1 on failure (node is NULL).
 */
int fr_redis_cluster_ipaddr(fr_ipaddr_t *out, fr_redis_cluster_node_t const *node)
{
	if (!node) return -1;

	memcpy(out, &node->addr.inet.dst_ipaddr, sizeof(*out));

	return 0;
}

/** 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_cluster_port(uint16_t *out, fr_redis_cluster_node_t const *node)
{
	if (!node) return -1;

	*out = node->addr.inet.dst_port;

	return 0;
}

/** Resolve a key to a pool, and reserve a connection in that pool
 *
 * This should be used with #fr_redis_cluster_state_next, and #fr_redis_command_status, to
 * transparently locate the cluster node we need to perform the operation on.
 *
 * Example code below shows how this function is used in conjunction
 * with #fr_redis_cluster_state_next to follow redirects, and reconnect handles.
 *
 @code{.c}
    int                 s_ret;
    redis_conn_state	state;
    fr_redis_conn_t  	*conn;
    redisReply		*reply;
    fr_redis_rcode_t	status;

    for (s_ret = fr_redis_cluster_state_init(&state, &conn, cluster, key, key_len, false);
         s_ret == REDIS_RCODE_TRY_AGAIN,
         s_ret = fr_redis_cluster_state_next(&state, &conn, cluster, request, status, &reply)) {
            reply = redisCommand(conn->handle, "SET foo bar");
            status = fr_redis_command_status(conn, reply);
    }
    // Reply is freed if ret == REDIS_RCODE_TRY_AGAIN, but left in all other cases to allow error
    // processing, or extraction of results.
    fr_redis_reply_free(&reply);
    if (s_ret != REDIS_RCODE_SUCCESS) {
    	// Error
    }
    // Success
 @endcode
 *
 * @param[out] state to track current pool and various counters, will be initialised.
 * @param[out] conn Where to write the reserved connection to.
 * @param[in] cluster of pools.
 * @param[in] request The current request.
 * @param[in] key to resolve to a cluster node/pool. If no key is NULL or key_len is 0 a random
 *	slot will be chosen.
 * @param[in] key_len Length of the key.
 * @param[in] read_only If true, will use random slave pool in preference to the master, falling
 *	back to the master if no slaves are available.
 * @return
 *	- REDIS_RCODE_TRY_AGAIN - try your command with this connection (provided via command).
 *	- REDIS_RCODE_RECONNECT - when no additional connections available.
 */
fr_redis_rcode_t fr_redis_cluster_state_init(fr_redis_cluster_state_t *state, fr_redis_conn_t **conn,
					     fr_redis_cluster_t *cluster, request_t *request,
					     uint8_t const *key, size_t key_len, bool read_only)
{
	fr_redis_cluster_node_t			*node;
	fr_redis_cluster_key_slot_t const	*key_slot;
	uint8_t					first, i;
	uint64_t				used_nodes;

	fr_assert(cluster);
	fr_assert(state);
	fr_assert(conn);

	memset(state, 0, sizeof(*state));
	*conn = NULL;	/* Better safe than exploding */

	used_nodes = fr_rb_num_elements(cluster->used_nodes);
	if (used_nodes == 0) {
		ROPTIONAL(REDEBUG, ERROR, "No nodes in cluster");
		return REDIS_RCODE_RECONNECT;
	}

again:
	key_slot = fr_redis_cluster_slot_by_key(cluster, request, key, key_len);

	/*
	 *	1. Try each of the slaves for the key slot
	 *	2. Fall through to trying the master, and a single alternate node.
	 */
	if (read_only) {
		first = fr_rand() & key_slot->slave_num;
		for (i = 0; i < key_slot->slave_num; i++) {
			uint8_t node_id;

			node_id = key_slot->slave[(first + i) % key_slot->slave_num];
			node = &cluster->node[node_id];
			*conn = fr_pool_connection_get(node->pool, request);
			if (!*conn) {
				ROPTIONAL(RDEBUG2, DEBUG2, "[%i] No connections available (key slot %zu slave %i)",
					  node->id, key_slot - cluster->key_slot, (first + i) % key_slot->slave_num);
				cluster->remap_needed = true;
				continue;	/* Continue until we find a live pool */
			}

			goto finish;
		}
		/* Fall through to using key slot master or alternate */
	}

	/*
	 *	1. Try the master for the key slot
	 *	2. If unavailable search for any pools with handles available
	 *	3. If there are no pools, or we can't reserve a handle,
	 *	   give up.
	 */
	node = &cluster->node[key_slot->master];
	*conn = fr_pool_connection_get(node->pool, request);
	if (!*conn) {
		ROPTIONAL(RDEBUG2, DEBUG2, "[%i] No connections available (key slot %zu master)",
			  node->id, key_slot - cluster->key_slot);
		cluster->remap_needed = true;

		if (cluster_node_find_live(&node, conn, request, cluster, node) < 0) return REDIS_RCODE_RECONNECT;
	}

finish:
	/*
	 *	Something set the remap_needed flag, and we have a live connection
	 */
	if (cluster->remap_needed) {
		if (fr_redis_cluster_remap(request, cluster, *conn) == FR_REDIS_CLUSTER_RCODE_SUCCESS) {
			fr_pool_connection_release(node->pool, request, *conn);
			goto again;	/* New map, try again */
		}
		ROPTIONAL(RPDEBUG2, PDEBUG2, "%s", "");
	}

	state->node = node;
	state->key = key;
	state->key_len = key_len;

	ROPTIONAL(RDEBUG2, DEBUG2, "[%i] >>> Sending command(s) to %s:%i",
		  state->node->id, state->node->name, state->node->addr.inet.dst_port);

	return REDIS_RCODE_TRY_AGAIN;
}

/** Get the next connection to attempt a command against
 *
 * Will process reconnect and redirect states performing the actions necessary.
 *
 * - May trigger a cluster remap on receiving a #REDIS_RCODE_MOVE status.
 * - May perform a temporary redirect on receiving a #REDIS_RCODE_ASK status.
 * - May reserve a new connection on receiving a #REDIS_RCODE_RECONNECT status.
 *
 * If a remap is in progress, has occurred within the last second, has recently failed,
 * or fails, the '-MOVE' will be treated as a temporary redirect (-ASK).
 *
 * This allows the server to be more responsive during remaps, as unless the worker has been
 * redirected to a node we don't currently have a pool for, it can grab a connection for the
 * node it was redirected to, and continue.
 *
 * @note Irrespective of return code, the connection passed via conn will be released,
 *	A new connection to attempt command on will be provided via conn.
 *
 * @note reply will be automatically freed and set to NULL if a new connection is provided
 *	in all other cases, the caller is responsible for freeing the reply.
 *
 * @param[in,out] state containing the current pool, and various counters which control
 *	retries, and limit redirects.
 * @param[in,out] conn we received the '-ASK' or '-MOVE' redirect on. Will be replaced with a
 *	connection in the new pool the key points to.
 * @param[in] request The current request.
 * @param[in] cluster of pools.
 * @param[in] status of the last command, must be #REDIS_RCODE_MOVE or #REDIS_RCODE_ASK.
 * @param[in] reply from last command.  Freed if 0 is returned, else caller must free.
 * @return
 *	- REDIS_RCODE_SUCCESS - on success.
 *	- REDIS_RCODE_TRY_AGAIN - try new connection (provided via conn). Will free reply.
 *	- REDIS_RCODE_ERROR - on failure or command error.
 *	- REDIS_RCODE_RECONNECT - when no additional connections available.
 */
fr_redis_rcode_t fr_redis_cluster_state_next(fr_redis_cluster_state_t *state, fr_redis_conn_t **conn,
					     fr_redis_cluster_t *cluster, request_t *request,
					     fr_redis_rcode_t status, redisReply **reply)
{
	fr_assert(state && state->node && state->node->pool);
	fr_assert(conn && *conn);

	if (*reply) fr_redis_reply_print(L_DBG_LVL_3, *reply, request, 0);

 	ROPTIONAL(RDEBUG2, DEBUG2, "[%i] <<< Returned: %s",
 		  state->node->id, fr_table_str_by_value(redis_rcodes, status, "<UNKNOWN>"));

	/*
	 *	Caller indicated we should close the connection
	 */
	if (state->close_conn) {
		ROPTIONAL(RDEBUG2, DEBUG2, "[%i] Connection no longer viable, closing it", state->node->id);
		fr_pool_connection_close(state->node->pool, request, *conn);
		*conn = NULL;
		state->close_conn = false;
	}

	/*
	 *	If we have a proven live connection, and something
	 *	has set the remap_needed flag, do that now before
	 *	releasing the connection.
	 */
	if (cluster->remap_needed && *conn) switch(status) {
	case REDIS_RCODE_MOVE:		/* We're going to remap anyway */
	case REDIS_RCODE_RECONNECT:	/* The connection's dead */
		break;

	default:
		/*
		 *	Remap the cluster. On success, will clear the
		 *	remap_needed flag.
		 */
		if (fr_redis_cluster_remap(request, cluster, *conn) != FR_REDIS_CLUSTER_RCODE_SUCCESS)
		ROPTIONAL(RPDEBUG2, PDEBUG2, "%s", "");
	}

	/*
	 *	Check the result of the last redis command, and do
	 *	something appropriate.
	 */
	switch (status) {
	case REDIS_RCODE_SUCCESS:
		fr_pool_connection_release(state->node->pool, request, *conn);
		*conn = NULL;
		return REDIS_RCODE_SUCCESS;

	/*
	 *	Command error, not fixable.
	 */
	case REDIS_RCODE_NO_SCRIPT:
	case REDIS_RCODE_ERROR:
		ROPTIONAL(RPEDEBUG, PERROR, "[%i] Command failed", state->node->id);
		fr_pool_connection_release(state->node->pool, request, *conn);
		*conn = NULL;
		return REDIS_RCODE_ERROR;

	/*
	 *	Cluster's unstable, try again.
	 */
	case REDIS_RCODE_TRY_AGAIN:
		if (state->retries++ >= cluster->conf->max_retries) {
			ROPTIONAL(REDEBUG, ERROR, "[%i] Hit maximum retry attempts", state->node->id);
			fr_pool_connection_release(state->node->pool, request, *conn);
			*conn = NULL;
			return REDIS_RCODE_ERROR;
		}

		if (!*conn) *conn = fr_pool_connection_get(state->node->pool, request);

		if (fr_time_delta_ispos(cluster->conf->retry_delay)) nanosleep(&fr_time_delta_to_timespec(cluster->conf->retry_delay), NULL);
		goto try_again;

	/*
	 *	Connection's dead, check to see if we can switch nodes,
	 *	or, failing that, reconnect the connection.
	 */
	case REDIS_RCODE_RECONNECT:
	{
		fr_redis_cluster_key_slot_t const *key_slot;

		ROPTIONAL(RPERROR, PERROR, "[%i] Failed communicating with %s:%i",
			  state->node->id, state->node->name,
			  state->node->addr.inet.dst_port);

		fr_pool_connection_close(state->node->pool, request, *conn);	/* He's dead jim */

		if (state->reconnects++ > state->in_pool) {
			ROPTIONAL(REDEBUG, ERROR, "[%i] Hit maximum reconnect attempts", state->node->id);
			cluster->remap_needed = true;
			return REDIS_RCODE_RECONNECT;
		}

		/*
		 *	Refresh the key slot
		 */
		key_slot = fr_redis_cluster_slot_by_key(cluster, request, state->key, state->key_len);
		state->node = &cluster->node[key_slot->master];

		*conn = fr_pool_connection_get(state->node->pool, request);
		if (!*conn) {
			ROPTIONAL(REDEBUG, ERROR, "[%i] No connections available for %s:%i",
				  state->node->id, state->node->name, state->node->addr.inet.dst_port);
			cluster->remap_needed = true;

			if (cluster_node_find_live(&state->node, conn, request,
						   cluster, state->node) < 0) return REDIS_RCODE_RECONNECT;

			goto try_again;
		}

		state->retries = 0;
	}
		goto try_again;

	/*
	 *	-MOVE is treated identically to -ASK, except it may
	 *	trigger a cluster remap.
	 */
	case REDIS_RCODE_MOVE:
		fr_assert(*reply);

		if (*conn && (fr_redis_cluster_remap(request, cluster, *conn) != FR_REDIS_CLUSTER_RCODE_SUCCESS)) {
			ROPTIONAL(RPDEBUG2, PDEBUG2, "%s", "");
		}
		FALL_THROUGH;

	/*
	 *	-ASK process a redirect.
	 */
	case REDIS_RCODE_ASK:
	{
		fr_redis_cluster_node_t *new;

		fr_pool_connection_release(state->node->pool, request, *conn);	/* Always release the old connection */

		if (!fr_cond_assert(*reply)) return REDIS_RCODE_ERROR;

		ROPTIONAL(RDEBUG2, DEBUG2, "[%i] Processing redirect \"%s\"", state->node->id, (*reply)->str);
		if (state->redirects++ >= cluster->conf->max_redirects) {
			ROPTIONAL(REDEBUG, ERROR, "[%i] Reached max_redirects (%i)", state->node->id, state->redirects);
			return REDIS_RCODE_ERROR;
		}

		switch (cluster_redirect(&new, cluster, *reply)) {
		case FR_REDIS_CLUSTER_RCODE_SUCCESS:
			if (new == state->node) {
				ROPTIONAL(REDEBUG, ERROR, "[%i] %s:%i issued redirect to itself", state->node->id,
					  state->node->name, state->node->addr.inet.dst_port);
				return REDIS_RCODE_ERROR;
			}

			ROPTIONAL(RDEBUG2, DEBUG2, "[%i] Redirected from %s:%i to [%i] %s:%i",
				  state->node->id, state->node->name,
				  state->node->addr.inet.dst_port, new->id, new->name, new->addr.inet.dst_port);
			state->node = new;

			*conn = fr_pool_connection_get(state->node->pool, request);
			if (!*conn) return REDIS_RCODE_RECONNECT;

			/*
			 *	Reset these counters, their scope is
			 *	a single node in the cluster.
			 */
			state->reconnects = 0;
			state->retries = 0;
			state->in_pool = fr_pool_state(state->node->pool)->num;
			goto try_again;

		case FR_REDIS_CLUSTER_RCODE_NO_CONNECTION:
			cluster->remap_needed = true;
			return REDIS_RCODE_RECONNECT;

		default:
			return REDIS_RCODE_ERROR;
		}
	}
	}

try_again:
	ROPTIONAL(RDEBUG2, DEBUG2, "[%i] >>> Sending command(s) to %s:%i",
		  state->node->id, state->node->name, state->node->addr.inet.dst_port);

	fr_redis_reply_free(&*reply);
	*reply = NULL;

	return REDIS_RCODE_TRY_AGAIN;
}

/** Get the pool associated with a node in the cluster
 *
 * @note This is used for testing only.  It's not ifdef'd out because
 *	tests need to run against production builds too.
 *
 * @param[out] pool associated with the node.
 * @param[in] cluster to search for node in.
 * @param[in] node_addr to retrieve pool for.  Specifies IP and port of node.
 * @param[in] create Establish a connection to the specified node if it
 *	was previously unknown to the cluster client.
 * @return
 *	- 0 on success.
 *	- -1 if no such node exists.
 */
int fr_redis_cluster_pool_by_node_addr(fr_pool_t **pool, fr_redis_cluster_t *cluster,
				       fr_socket_t *node_addr, bool create)
{
	fr_redis_cluster_node_t	find, *found;

	find.addr = (fr_socket_t) {
		.inet = {
			.dst_ipaddr = node_addr->inet.dst_ipaddr,
			.dst_port = node_addr->inet.dst_port,
		}
	};

	pthread_mutex_lock(&cluster->mutex);
	found = fr_rb_find(cluster->used_nodes, &find);
	if (!found) {
		fr_redis_cluster_node_t *spare;
		char buffer[INET6_ADDRSTRLEN];
		char const *hostname;

		if (!create) {
			pthread_mutex_unlock(&cluster->mutex);

			hostname = inet_ntop(node_addr->inet.dst_ipaddr.af, &node_addr->inet.dst_ipaddr.addr, buffer, sizeof(buffer));
			fr_assert(hostname);	/* addr.ipaddr is probably corrupt */
			fr_strerror_printf("No existing node found with address %s, port %i",
					   hostname, node_addr->inet.dst_port);
			return -1;
		}

		spare = fr_fifo_peek(cluster->free_nodes);
		if (!spare) {
			fr_strerror_const("Reached maximum connected nodes");
			pthread_mutex_unlock(&cluster->mutex);
			return -1;
		}
		spare->pending_addr = find.addr;	/* Set the config to be applied */
		if (cluster_node_connect(cluster, spare) < 0) {
			pthread_mutex_unlock(&cluster->mutex);
			return -1;
		}
		fr_rb_insert(cluster->used_nodes, spare);
		fr_fifo_pop(cluster->free_nodes);
		found = spare;
	}
	/*
	 *	Sanity checks
	 */
	fr_assert(((talloc_array_length(cluster->node) - 1) - fr_rb_num_elements(cluster->used_nodes)) ==
		   fr_fifo_num_elements(cluster->free_nodes));
	pthread_mutex_unlock(&cluster->mutex);

	*pool = found->pool;

	return 0;
}

/** Return an array of IP addresses belonging to masters or slaves
 *
 * @note We return IP addresses as they're safe to use across cluster remaps.
 * @note Result array must be freed (talloc_free()) after use.
 *
 * @param[in] ctx to allocate array of IP addresses in.
 * @param[out] out		Where to write the addresses of the nodes.
 * @param[in] cluster		to search for nodes in.
 * @param[in] is_master		If true, include the addresses of all the master nodes.
 * @param[in] is_slave		If true, include the addresses of all the slaves nodes.
 * @return the number of ip addresses written to out.
 */
ssize_t fr_redis_cluster_node_addr_by_role(TALLOC_CTX *ctx, fr_socket_t *out[],
					   fr_redis_cluster_t *cluster, bool is_master, bool is_slave)
{
	uint64_t 			in_use = fr_rb_num_elements(cluster->used_nodes);
	fr_rb_iter_inorder_t	iter;
	fr_redis_cluster_node_t		*node;
	uint8_t				count;
	fr_socket_t			*found;

	if (in_use == 0) {
		*out = NULL;
		return 0;
	}

	count = 0;
	found = talloc_zero_array(ctx, fr_socket_t, in_use);
	if (!found) {
		fr_strerror_const("Out of memory");
		return -1;
	}

	pthread_mutex_lock(&cluster->mutex);

	for (node = fr_rb_iter_init_inorder(&iter, cluster->used_nodes);
	     node;
	     node = fr_rb_iter_next_inorder(&iter)) {
		if ((is_master && node->is_master) || (is_slave && !node->is_master)) found[count++] = node->addr;
	}

	pthread_mutex_unlock(&cluster->mutex);

	if (count == 0) {
		*out = NULL;
		talloc_free(found);
		return 0;
	}

	*out = found;

	return count;
}

/** Destroy mutex associated with cluster slots structure
 *
 * @param cluster being freed.
 * @return 0
 */
static int _fr_redis_cluster_free(fr_redis_cluster_t *cluster)
{
	pthread_mutex_destroy(&cluster->mutex);

	return 0;
}

/** Check if members of the cluster are above a certain version
 *
 * @param cluster to perform check on.
 * @param min_version that must be found on each node for the check to succeed.
 *	Must be in the format @verbatim <major>.<minor>.<release> @endverbatim.
 * @return
 *	- true if all contactable members are above min_version.
 *	- false if at least one member if not above minimum version
 *	  (use #fr_strerror to retrieve node information).
 */
bool fr_redis_cluster_min_version(fr_redis_cluster_t *cluster, char const *min_version)
{
	fr_redis_cluster_node_t		*node;
	fr_rb_iter_inorder_t	iter;
	fr_redis_conn_t			*conn;
	int				ret;
	char				buffer[40];
	bool				all_above = true;

	pthread_mutex_lock(&cluster->mutex);

	for (node = fr_rb_iter_init_inorder(&iter, cluster->used_nodes);
	     node;
	     node = fr_rb_iter_next_inorder(&iter)) {
		conn = fr_pool_connection_get(node->pool, NULL);
		if (!conn) continue;

		/*
		 *	We don't care if we can't get the version
		 *	as we don't want to prevent the server from
		 *	starting if start == 0.
		 */
		ret = fr_redis_get_version(buffer, sizeof(buffer), conn);
		fr_pool_connection_release(node->pool, NULL, conn);
		if (ret < 0) continue;

		if (fr_redis_version_num(buffer) < fr_redis_version_num(min_version)) {
			fr_strerror_printf("Redis node %s:%i (currently v%s) needs update to >= v%s",
					   node->name, node->addr.inet.dst_port, buffer, min_version);
			all_above = false;
			break;
		}
	}

	pthread_mutex_unlock(&cluster->mutex);

	return all_above;
}

/** Allocate and initialise a new cluster structure
 *
 * This holds all the data necessary to manage a pool of pools for a specific redis cluster.
 *
 * @note Will not error out unless cs.pool.start > 0.  This is consistent with other pool based
 *	modules/code.
 *
 * @param ctx			to link the lifetime of the cluster structure to.
 * @param module		Configuration section to search for 'server' conf pairs in.
 * @param conf			Base redis server configuration. Cluster nodes share database
 *				number and password.
 * @param triggers_enabled	Whether triggers should be enabled.
 * @param log_prefix		Custom log prefix.  Defaults to @verbatim rlm_<module> (<instance>) @endverbatim.
 * @param trigger_prefix	Custom trigger prefix.  Defaults to @verbatim modules.<module>.pool @endverbatim.
 * @param trigger_args		Argument pairs to pass to the trigger in addition to Connection-Pool-Server,
 *				and Connection-Pool-Port (which are always set by the cluster code).
 * @return
 *	- New #fr_redis_cluster_t on success.
 *	- NULL on error.
 */
fr_redis_cluster_t *fr_redis_cluster_alloc(TALLOC_CTX *ctx,
					   CONF_SECTION *module,
					   fr_redis_conf_t *conf,
					   bool triggers_enabled,
					   char const *log_prefix,
					   char const *trigger_prefix,
					   fr_pair_list_t *trigger_args)
{
	uint8_t			i;
	uint16_t		s;

	char const		*cs_name1, *cs_name2;

	CONF_PAIR		*cp;
	int			af = AF_UNSPEC;		/* AF of first server */

	uint64_t		num_nodes;
	fr_redis_cluster_t	*cluster;

	fr_assert(triggers_enabled || !trigger_prefix);
	fr_assert(triggers_enabled || (!trigger_args || fr_pair_list_empty(trigger_args)));

	MEM(cluster = talloc_zero(NULL, fr_redis_cluster_t));
	fr_pair_list_init(&cluster->trigger_args);

	cs_name1 = cf_section_name1(module);
	cs_name2 = cf_section_name2(module);

	cluster->triggers_enabled = triggers_enabled;
	if (cluster->triggers_enabled) {
		/*
		 *	Setup trigger prefix
		 */
		if (!trigger_prefix) {
			cluster->trigger_prefix = talloc_typed_asprintf(cluster, "modules.%s.pool", cs_name1);
		} else {
			cluster->trigger_prefix = talloc_strdup(cluster, trigger_prefix);
		}

		/*
		 *	Duplicate the trigger arguments.
		 */
		 if (trigger_args) MEM(fr_pair_list_copy(cluster, &cluster->trigger_args, trigger_args) >= 0);
	}

	/*
	 *	Setup log prefix
	 */
	if (!log_prefix) {
		if (!cs_name2) cs_name2 = cs_name1;
		cluster->log_prefix = talloc_typed_asprintf(conf, "rlm_%s (%s)", cs_name1, cs_name2);
	} else {
		cluster->log_prefix = talloc_strdup(cluster, log_prefix);
	}

	/*
	 *	Ensure we always have a pool section (even if it's empty)
	 */
	if (!cf_section_find(module, "pool", NULL)) {
		(void) cf_section_alloc(module, module, "pool", NULL);
	}

	/*
	 *	Parse TLS configuration
	 */
	if (conf->use_tls) {
#ifdef HAVE_REDIS_SSL
		CONF_SECTION *tls_cs;
		fr_tls_conf_t *tls_conf;

		tls_cs = cf_section_find(module, "tls", NULL);
		if (!tls_cs) {
			tls_cs = cf_section_alloc(module, module, "tls", NULL);
		}

		tls_conf = fr_tls_conf_parse_client(tls_cs);
		if (!tls_conf) {
			ERROR("%s - Failed to parse TLS configuration", cluster->log_prefix);
			talloc_free(cluster);
			return NULL;
		}

		cluster->ssl_ctx = fr_tls_ctx_alloc(tls_conf, true);
		if (!cluster->ssl_ctx) {
			ERROR("%s - Failed to allocate SSL context", cluster->log_prefix);
			talloc_free(cluster);
			return NULL;
		}
#else
		WARN("%s - No redis SSL support, ignoring \"use_tls = yes\"", cluster->log_prefix);
#endif
	}

	if (conf->max_nodes == UINT8_MAX) {
		ERROR("%s - Maximum number of connected nodes allowed is %i", cluster->log_prefix, UINT8_MAX - 1);
		talloc_free(cluster);
		return NULL;
	}

	if (conf->max_nodes == 0) {
		ERROR("%s - Minimum number of nodes allowed is 1", cluster->log_prefix);
		talloc_free(cluster);
		return NULL;
	}

	cp = cf_pair_find(module, "server");
	if (!cp) {
		ERROR("%s - No servers configured", cluster->log_prefix);
		talloc_free(cluster);
		return NULL;
	}

	cluster->module = module;

	/*
	 *	Ensure the pool is freed at the same time as its
	 *	parent.
	 *
	 *	We need to break the link between the cluster and
	 *	its parent context, as the two contexts may be
	 *	modified by multiple threads.
	 */
	MEM(talloc_link_ctx(ctx, cluster) >= 0);
	MEM(cluster->node = talloc_zero_array(cluster, fr_redis_cluster_node_t, conf->max_nodes + 1));
	MEM(cluster->used_nodes = fr_rb_inline_alloc(cluster, fr_redis_cluster_node_t, rbnode, _cluster_node_cmp, NULL));
	MEM(cluster->free_nodes = fr_fifo_create(cluster, conf->max_nodes, NULL));

	cluster->conf = conf;

	pthread_mutex_init(&cluster->mutex, NULL);
	talloc_set_destructor(cluster, _fr_redis_cluster_free);

	/*
	 *	Node id 0 is reserved, so we can detect misconfigured
	 *	clusters.
	 */
	for (i = 1; i < (cluster->conf->max_nodes + 1); i++) {
		cluster->node[i].id = i;
		cluster->node[i].cluster = cluster;

		/* Push them all into the queue */
		fr_fifo_push(cluster->free_nodes, &cluster->node[i]);
	}

	/*
	 *	Don't connect to cluster nodes if we're just
	 *	checking the config.
	 */
	if (check_config) return cluster;

	/*
	 *	Populate the cluster with the bootstrap servers.
	 *
	 *	If we fail getting a key_slot map here, then the
	 *	bootstrap servers are distributed evenly through
	 *	the key slots.
	 *
	 *	This allows the server to start, and potentially,
	 *	a valid map to be applied, once the server starts
	 *	processing requests.
	 */
	do {
		char const	*server;
		fr_redis_cluster_node_t	*node;
		fr_redis_conn_t	*conn;
		redisReply	*map;
		size_t		j, k;

		node = fr_fifo_peek(cluster->free_nodes);
		if (!node) {
			ERROR("%s - Number of bootstrap servers exceeds 'max_nodes'", cluster->log_prefix);

		error:
			talloc_free(cluster);
			return NULL;
		}

		server = cf_pair_value(cp);
		if (fr_inet_pton_port(&node->pending_addr.inet.dst_ipaddr, &node->pending_addr.inet.dst_port, server,
				 talloc_array_length(server) - 1, af, true, true) < 0) {
			PERROR("%s - Failed parsing server \"%s\"", cluster->log_prefix, server);
			goto error;
		}
		if (!node->pending_addr.inet.dst_port) node->pending_addr.inet.dst_port = conf->port;

		if (cluster_node_connect(cluster, node) < 0) {
			WARN("%s - Connecting to %s:%i failed", cluster->log_prefix, node->name, node->pending_addr.inet.dst_port);
			continue;
		}

		if (!fr_rb_insert(cluster->used_nodes, node)) {
			WARN("%s - Skipping duplicate bootstrap server \"%s\"", cluster->log_prefix, server);
			continue;
		}
		node->is_active = true;
		fr_fifo_pop(cluster->free_nodes);

		/*
		 *	Prefer the same IPaddr family as the first node
		 */
		if (af == AF_UNSPEC) af = node->addr.inet.dst_ipaddr.af;

		/*
		 * 	Only get cluster map config if required
		 */
		if (fr_pool_start_num(node->pool) > 0) {
			/*
			 *	Fine to leave this node configured, if we do find
			 *	a live node, and it's not in the map, it'll be cleared out.
			 */
			conn = fr_pool_connection_get(node->pool, NULL);
			if (!conn) {
				WARN("%s - Can't contact bootstrap server \"%s\"", cluster->log_prefix, server);
				continue;
			}
		} else {
			break;
		}

		if (!cluster->conf->use_cluster_map) {
			fr_pool_connection_release(node->pool, NULL, conn);
			continue;
		}

		switch (cluster_map_get(&map, conn)) {
		/*
		 *	We got a valid map! See if we can apply it...
		 */
		case FR_REDIS_CLUSTER_RCODE_SUCCESS:
			fr_pool_connection_release(node->pool, NULL, conn);

			DEBUG("%s - Cluster map consists of %zu key ranges", cluster->log_prefix, map->elements);
			for (j = 0; j < map->elements; j++) {
				redisReply *map_node = map->element[j];

				DEBUG("%s - %zu - keys %lli-%lli", cluster->log_prefix, j,
				      map_node->element[0]->integer,
				      map_node->element[1]->integer);
				DEBUG("%s -  master: %s:%lli", cluster->log_prefix,
				      map_node->element[2]->element[0]->str,
				      map_node->element[2]->element[1]->integer);
				for (k = 3; k < map_node->elements; k++) {
					DEBUG("%s -  slave%zu: %s:%lli", cluster->log_prefix, k - 3,
					      map_node->element[k]->element[0]->str,
					      map_node->element[k]->element[1]->integer);
				}
			}

			if (cluster_map_apply(cluster, map) < 0) {
				PWARN("%s: Applying cluster map failed", cluster->log_prefix);
				fr_redis_reply_free(&map);
				continue;
			}
			fr_redis_reply_free(&map);

			return cluster;

		/*
		 *	Unusable bootstrap node
		 */
		case FR_REDIS_CLUSTER_RCODE_BAD_INPUT:
			PWARN("%s - Bootstrap server \"%s\" returned invalid data", cluster->log_prefix, server);
			fr_pool_connection_release(node->pool, NULL, conn);
			continue;

		case FR_REDIS_CLUSTER_RCODE_NO_CONNECTION:
			PWARN("%s - Can't contact bootstrap server \"%s\"", cluster->log_prefix, server);
			fr_pool_connection_close(node->pool, NULL, conn);
			continue;

		/*
		 *	Clustering not enabled, or not supported,
		 *	by this node, skip it and check the others.
		 */
		case FR_REDIS_CLUSTER_RCODE_FAILED:
		case FR_REDIS_CLUSTER_RCODE_IGNORED:
			PDEBUG2("%s - Bootstrap server \"%s\" returned", cluster->log_prefix, server);
			fr_pool_connection_release(node->pool, NULL, conn);
			break;
		}
	} while ((cp = cf_pair_find_next(module, cp, "server")));

	/*
	 *	Catch pool.start != 0
	 */
	num_nodes = fr_rb_num_elements(cluster->used_nodes);
	if (!num_nodes) {
		ERROR("%s - Can't contact any bootstrap servers", cluster->log_prefix);
		goto error;
	}

	/*
	 *	We've failed to apply a valid cluster map.
	 *	Distribute the node(s) throughout the key_slots,
	 *	hopefully we'll get one when we start processing
	 *	requests.
	 */
	for (s = 0; s < KEY_SLOTS; s++) cluster->key_slot[s].master = (s % (uint16_t) num_nodes) + 1;

	return cluster;
}
