rlm_redis.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: a393f98a94b4a32ca24ecc1c59adc2e0c5b668df $
 * @file rlm_redis.c
 * @brief Driver for the Redis noSQL key value store.
 *
 * @author Gabriel Blanchard
 *
 * @copyright 2015 Arran Cudbard-Bell (a.cudbardb@freeradius.org)
 * @copyright 2011 TekSavvy Solutions (gabe@teksavvy.com)
 * @copyright 2000,2006,2015 The FreeRADIUS server project
 */
 
RCSID("$Id: a393f98a94b4a32ca24ecc1c59adc2e0c5b668df $")
 
#include <assert.h>
#include <stdint.h>
 
#include <freeradius-devel/redis/base.h>
#include <freeradius-devel/redis/cluster.h>
 
#include <freeradius-devel/server/base.h>
#include <freeradius-devel/server/cf_util.h>
#include <freeradius-devel/server/modpriv.h>
#include <freeradius-devel/server/module_rlm.h>
#include <freeradius-devel/server/pool.h>
 
#include <freeradius-devel/unlang/xlat.h>
#include <freeradius-devel/unlang/xlat_func.h>
 
#include <freeradius-devel/util/base16.h>
#include <freeradius-devel/util/debug.h>
#include <freeradius-devel/util/log.h>
#include <freeradius-devel/util/talloc.h>
#include <freeradius-devel/util/types.h>
#include <freeradius-devel/util/value.h>
 
/** A lua function or stored procedure we make available as an xlat
 *
 */
typedef struct {
        char const              *name;                                  //!< Friendly name for the function.  Used to register the equivalent xlat.
        char                    digest[(SHA1_DIGEST_LENGTH * 2) + 1];   //!< pre-computed hash of lua code.
        char const              *body;                                  //!< the actual lua code.
        bool                    read_only;                              //!< Function has no side effects
} redis_lua_func_t;
 
/** Instance of a redis lua func xlat
 *
 */
typedef struct {
        redis_lua_func_t        *func;                                  //!< Function configuration.
} redis_lua_func_inst_t;
 
 
typedef struct {
        redis_lua_func_t        **funcs;                                //!< Array of functions to register.
 
} rlm_redis_lua_t;
 
/** rlm_redis module instance
 *
 */
typedef struct {
        fr_redis_conf_t         conf;                                   //!< Connection parameters for the Redis server.
                                                                        //!< Must be first field in this struct.
 
        rlm_redis_lua_t         lua;                                    //!< Array of functions to register.
 
        fr_redis_cluster_t      *cluster;                               //!< Redis cluster.
} rlm_redis_t;
 
static int lua_func_body_parse(TALLOC_CTX *ctx, void *out, void *parent, CONF_ITEM *ci, conf_parser_t const *rule);
 
static conf_parser_t module_lua_func[] = {
        { FR_CONF_OFFSET("body", redis_lua_func_t, body), .func = lua_func_body_parse },
        { FR_CONF_OFFSET("read_only", redis_lua_func_t, read_only) },
        CONF_PARSER_TERMINATOR
};
 
static conf_parser_t module_lua[] = {
        { FR_CONF_SUBSECTION_ALLOC("function", 0, CONF_FLAG_SUBSECTION | CONF_FLAG_OK_MISSING | CONF_FLAG_MULTI,
                                   rlm_redis_lua_t, funcs, module_lua_func),
                                   .subcs_type = "redis_lua_func_t", .name2 = CF_IDENT_ANY },
        CONF_PARSER_TERMINATOR
};
 
static conf_parser_t module_config[] = {
        { FR_CONF_OFFSET_SUBSECTION("lua", 0, rlm_redis_t, lua, module_lua) },
        REDIS_COMMON_CONFIG,
        CONF_PARSER_TERMINATOR
};
 
/** Do basic processing for a lua function body and compute its sha1 hash
 *
 */
static int lua_func_body_parse(TALLOC_CTX *ctx, void *out, void *parent, CONF_ITEM *ci, conf_parser_t const *rule)
{
        int                     ret;
        redis_lua_func_t        *func = talloc_get_type_abort(parent, redis_lua_func_t);
        char const              *body;
        fr_sha1_ctx             sha1_ctx;
        uint8_t                 digest[SHA1_DIGEST_LENGTH];
 
        /*
         *      Get the function name from name2
         *      of the enclosing function section.
         */
        func->name = cf_section_name2(cf_item_to_section(cf_parent(ci)));
        if (unlikely(!func->name)) {
                cf_log_err(cf_parent(ci), "functions must be declared as \"function <name> {\"");
                return -1;
        }
 
        /*
         *      Perform normal string parsing first
         */
        if ((ret = cf_pair_parse_value(ctx, out, parent, ci, rule)) < 0) return ret;
        body = *((char **)out);
 
        fr_sha1_init(&sha1_ctx);
        fr_sha1_update(&sha1_ctx, (uint8_t const *)body, talloc_array_length(body) - 1);
        fr_sha1_final(digest, &sha1_ctx);
        fr_base16_encode(&FR_SBUFF_OUT(func->digest, sizeof(func->digest)), &FR_DBUFF_TMP(digest, sizeof(digest)));
 
        if (DEBUG_ENABLED3) cf_log_debug(ci, "sha1 hash of function is %pV", fr_box_strvalue_len(func->digest, sizeof(func->digest) - 1));
 
        return 0;
}
 
/** Issue a command against redis and get a response
 *
 * This is a convenience function for dealing with commands which made need to execute against an
 * ldap replica.  It temporarily places the connection in readonly mode to allow commands to be
 * run against ldap replicas, then reverts back to readwrite mode.
 *
 * @param[out] status_out       Where to write the status from the command.
 * @param[out] reply_out        Where to write the reply associated with the highest priority status.
 * @param[in] request           The current request.
 * @param[in] conn              to issue commands with.
 * @param[in] read_only         wrap command in READONLY/READWRITE.
 * @param[in] argc              Redis command argument count.
 * @param[in] argv              Redis command arguments.
 * @param[in] arg_len           Optional array of redis command argument length.
 * @return
 *      - 0 success.
 *      - -1 normal failure.
 *      - -2 failure that may leave the connection in a READONLY state.
 */
static int redis_command(fr_redis_rcode_t *status_out, redisReply **reply_out,
                         request_t *request, fr_redis_conn_t *conn,
                         bool read_only,
                         int argc, char const **argv, size_t arg_len[])
{
        bool                    maybe_more = false;
        redisReply              *reply;
        fr_redis_rcode_t        status;
 
        *reply_out = NULL;
 
        if (read_only) redisAppendCommand(conn->handle, "READONLY");
        redisAppendCommandArgv(conn->handle, argc, argv, arg_len);
        if (read_only) {
                redisAppendCommand(conn->handle, "READWRITE");
        } else goto parse_reply;
 
 
        /*
         *      Process the response for READONLY
         */
        reply = NULL;   /* Doesn't set reply to NULL on error *sigh* */
        if (redisGetReply(conn->handle, (void **)&reply) == REDIS_OK) maybe_more = true;
        status = fr_redis_command_status(conn, reply);
        if (status != REDIS_RCODE_SUCCESS) {
                ROPTIONAL(REDEBUG, ERROR, "Setting READONLY failed");
 
                *reply_out = reply;
                *status_out = status;
 
                if (maybe_more) {
                        if (redisGetReply(conn->handle, (void **)&reply) != REDIS_OK) return -1;
                        fr_redis_reply_free(&reply);
                        if (redisGetReply(conn->handle, (void **)&reply) != REDIS_OK) return -1;
                        fr_redis_reply_free(&reply);
                }
                return -1;
        }
 
        fr_redis_reply_free(&reply);
 
parse_reply:
        /*
         *      Process the response for the command
         */
        reply = NULL;   /* Doesn't set reply to NULL on error *sigh* */
        if ((redisGetReply(conn->handle, (void **)&reply) == REDIS_OK) && read_only) maybe_more = true;
        status = fr_redis_command_status(conn, reply);
        if (status != REDIS_RCODE_SUCCESS) {
                *reply_out = reply;
                *status_out = status;
 
                if (maybe_more) {
                        if (redisGetReply(conn->handle, (void **)&reply) != REDIS_OK) return -1;
                        fr_redis_reply_free(&reply);
                }
                return -1;
        }
 
        *reply_out = reply;
        reply = NULL;
        *status_out = status;
 
        if (!read_only) return 0;       /* No more responses to deal with */
 
        /*
         *      Process the response for READWRITE
         */
        reply = NULL;   /* Doesn't set reply to NULL on error *sigh* */
        if ((redisGetReply(conn->handle, (void **)&reply) != REDIS_OK) ||
            (fr_redis_command_status(conn, reply) != REDIS_RCODE_SUCCESS)) {
                ROPTIONAL(REDEBUG, ERROR, "Setting READWRITE failed");
 
                fr_redis_reply_free(&reply);    /* There could be a response we need to free */
                fr_redis_reply_free(reply_out);
                *reply_out = reply;
                *status_out = status;
 
                return -2;
        }
        fr_redis_reply_free(&reply);    /* Free READWRITE response */
 
        return 0;
}
 
static xlat_arg_parser_t const redis_remap_xlat_args[] = {
        { .required = true, .concat = true, .type = FR_TYPE_STRING },
        XLAT_ARG_PARSER_TERMINATOR
};
 
/** Force a redis cluster remap
 *
@verbatim
%redis.remap(<redis server ip>:<redis server port>)
@endverbatim
 *
 * @ingroup xlat_functions
 */
static xlat_action_t redis_remap_xlat(TALLOC_CTX *ctx, fr_dcursor_t *out,
                                      xlat_ctx_t const *xctx,
                                      request_t *request, fr_value_box_list_t *in)
{
        rlm_redis_t const               *inst = talloc_get_type_abort_const(xctx->mctx->mi->data, rlm_redis_t);
 
        fr_socket_t                     node_addr;
        fr_pool_t                       *pool;
        fr_redis_conn_t                 *conn;
        fr_redis_cluster_rcode_t        rcode;
        fr_value_box_t                  *vb;
        fr_value_box_t                  *in_head = fr_value_box_list_head(in);
 
        if (fr_inet_pton_port(&node_addr.inet.dst_ipaddr, &node_addr.inet.dst_port, in_head->vb_strvalue, in_head->vb_length,
                              AF_UNSPEC, true, true) < 0) {
                RPEDEBUG("Failed parsing node address");
                return XLAT_ACTION_FAIL;
        }
 
        if (fr_redis_cluster_pool_by_node_addr(&pool, inst->cluster, &node_addr, true) < 0) {
                RPEDEBUG("Failed locating cluster node");
                return XLAT_ACTION_FAIL;
        }
 
        conn = fr_pool_connection_get(pool, request);
        if (!conn) {
                REDEBUG("No connections available for cluster node");
                return XLAT_ACTION_FAIL;
        }
 
        rcode = fr_redis_cluster_remap(request, inst->cluster, conn);
        switch (rcode) {
        case FR_REDIS_CLUSTER_RCODE_NO_CONNECTION:
                fr_pool_connection_close(pool, request, conn);
                break;
 
        default:
                fr_pool_connection_release(pool, request, conn);
                break;
        }
 
        MEM(vb = fr_value_box_alloc_null(ctx));
        fr_value_box_strdup(vb, vb, NULL, fr_table_str_by_value(fr_redis_cluster_rcodes_table, rcode, "<INVALID>"), false);
        fr_dcursor_append(out, vb);
 
        return XLAT_ACTION_DONE;
}
 
static xlat_arg_parser_t const redis_node_xlat_args[] = {
        { .required = true, .single = true, .type = FR_TYPE_STRING },
        { .single = true, .type = FR_TYPE_UINT32 },
        XLAT_ARG_PARSER_TERMINATOR
};
 
/** Return the node that is currently servicing a particular key
 *
@verbatim
%redis.node(<key>[, <index>])
@endverbatim
 *
 * @ingroup xlat_functions
 */
static xlat_action_t redis_node_xlat(TALLOC_CTX *ctx, fr_dcursor_t *out,
                                     xlat_ctx_t const *xctx,
                                     request_t *request, fr_value_box_list_t *in)
{
        rlm_redis_t const                       *inst = talloc_get_type_abort_const(xctx->mctx->mi->data, rlm_redis_t);
 
        fr_redis_cluster_key_slot_t const       *key_slot;
        fr_redis_cluster_node_t const           *node;
        fr_ipaddr_t                             ipaddr;
        uint16_t                                port;
 
        unsigned long                           idx = 0;
        fr_value_box_t                          *vb;
        fr_value_box_t                          *key = fr_value_box_list_head(in);
        fr_value_box_t                          *idx_vb = fr_value_box_list_next(in, key);
 
        if (idx_vb) idx = idx_vb->vb_uint32;
 
        key_slot = fr_redis_cluster_slot_by_key(inst->cluster, request, (uint8_t const *)key->vb_strvalue,
                                                key->vb_length);
        if (idx == 0) {
                node = fr_redis_cluster_master(inst->cluster, key_slot);
        } else {
                node = fr_redis_cluster_slave(inst->cluster, key_slot, idx - 1);
        }
 
        if (!node) {
                RDEBUG2("No node available for this key slot");
                return XLAT_ACTION_DONE;
        }
 
        if ((fr_redis_cluster_ipaddr(&ipaddr, node) < 0) || (fr_redis_cluster_port(&port, node) < 0)) {
                REDEBUG("Failed retrieving node information");
                return XLAT_ACTION_FAIL;
        }
 
        MEM(vb = fr_value_box_alloc_null(ctx));
        fr_value_box_asprintf(vb, vb, NULL, false, "%pV:%u", fr_box_ipaddr(ipaddr), port);
        fr_dcursor_append(out, vb);
 
        return XLAT_ACTION_DONE;
}
 
static xlat_arg_parser_t const redis_lua_func_args[] = {
        { .required = true, .single = true, .type = FR_TYPE_UINT64 }, /* key count */
        { .variadic = XLAT_ARG_VARIADIC_EMPTY_KEEP, .concat = true, .type = FR_TYPE_STRING }, /* keys and args */
        XLAT_ARG_PARSER_TERMINATOR
};
 
/** Call a lua function on the redis server
 *
 * Lua functions either get uploaded when the module is instantiated or the first
 * time they get executed.
 */
static xlat_action_t redis_lua_func_xlat(TALLOC_CTX *ctx, fr_dcursor_t *out,
                                         xlat_ctx_t const *xctx,
                                         request_t *request, fr_value_box_list_t *in)
{
        rlm_redis_t                     *inst = talloc_get_type_abort(xctx->mctx->mi->data, rlm_redis_t);
        redis_lua_func_inst_t const     *xlat_inst = talloc_get_type_abort_const(xctx->inst, redis_lua_func_inst_t);
        redis_lua_func_t                *func = xlat_inst->func;
 
        fr_redis_conn_t                 *conn;
        fr_redis_cluster_state_t        state;
        fr_redis_rcode_t                status;
 
        redisReply                      *reply = NULL;
        int                             s_ret;
 
        char const                      *argv[MAX_REDIS_ARGS];
        size_t                          arg_len[MAX_REDIS_ARGS];
        int                             argc;
        char                            key_count[sizeof("184467440737095551615")];
        uint8_t const                   *key = NULL;
        size_t                          key_len = 0;
 
        xlat_action_t                   action = XLAT_ACTION_DONE;
        fr_value_box_t                  *vb_out;
 
        /*
         *      First argument is always the key count
         */
        if (unlikely(fr_value_box_print(&FR_SBUFF_OUT(key_count, sizeof(key_count)), fr_value_box_list_head(in), NULL) < 0)) {
                RPERROR("Failed converting key count to string");
                return XLAT_ACTION_FAIL;
        }
        fr_value_box_list_talloc_free_head(in);
 
        /*
         *      Try EVALSHA first, and if that fails fall back to SCRIPT LOAD
         */
        argv[0] = "EVALSHA";
        arg_len[0] = sizeof("EVALSHA") - 1;
        argv[1] = func->digest;
        arg_len[1] = sizeof(func->digest) - 1;
        argv[2] = key_count;
        arg_len[2] = strlen(key_count);
        argc = 3;
 
        fr_value_box_list_foreach(in, vb) {
                if (argc == NUM_ELEMENTS(argv)) {
                        REDEBUG("Too many arguments (%i)", argc);
                        REXDENT();
                        return XLAT_ACTION_FAIL;
                }
 
                /*
                 *      Fixup null or empty arguments to be
                 *      zero length strings so that the position
                 *      of subsequent arguments are maintained.
                 */
                if (!fr_type_is_string(vb->type)) {
                        argv[argc] = "";
                        arg_len[argc++] = 0;
                        continue;
                }
 
                argv[argc] = vb->vb_strvalue;
                arg_len[argc++] = vb->vb_length;
        }
 
        /*
         *      For eval commands all keys should hash to the same redis instance
         *      so we just use the first key (the arg after the key count).
         */
        if (argc > 3) {
                key = (uint8_t const *)argv[3];
                key_len = arg_len[3];
        }
 
        for (s_ret = fr_redis_cluster_state_init(&state, &conn, inst->cluster, request, key, key_len, func->read_only);
             s_ret == REDIS_RCODE_TRY_AGAIN;    /* Continue */
             s_ret = fr_redis_cluster_state_next(&state, &conn, inst->cluster, request, status, &reply)) {
                bool script_load_done = false;
 
        again:
                RDEBUG3("Calling script 0x%s", func->digest);
                if (argc > 2) {
                        RDEBUG3("With arguments");
                        RINDENT();
                        for (int i = 2; i < argc; i++) RDEBUG3("[%i] %s", i, argv[i]);
                        REXDENT();
                }
                if (redis_command(&status, &reply, request, conn,
                                  func->read_only, argc, argv, arg_len) == -2) {
                        state.close_conn = true;
                }
 
                if (status != REDIS_RCODE_NO_SCRIPT) continue;
 
                /*
                 *      Discard the error we received, and attempt load the function.
                 */
                fr_redis_reply_free(&reply);
 
                RDEBUG3("Loading lua function \"%s\" (0x%s)", func->name, func->digest);
                {
                        char const      *script_load_argv[] = {
                                                "SCRIPT",
                                                "LOAD",
                                                func->body
                                        };
 
                        size_t          script_load_arg_len[] = {
                                                (sizeof("SCRIPT") - 1),
                                                (sizeof("LOAD") - 1),
                                                (talloc_array_length(func->body) - 1)
                                        };
 
                        /*
                         *      Loading the script failed... fail the call.
                         */
                        if (script_load_done) {
                        script_load_failed:
                                status = REDIS_RCODE_ERROR;
                                fr_redis_reply_free(&reply);
                                continue;
                        }
 
                        /*
                         *      Fixme: Really the script load and the eval call should be
                         *      handled in a single MULTI/EXEC block, but the complexity
                         *      in handling this properly is great, and most of this
                         *      synchronous code will need to be rewritten, so for now
                         *      we just load the script and try again.
                         */
                        if (redis_command(&status, &reply, request, conn, func->read_only,
                                          NUM_ELEMENTS(script_load_argv),
                                          script_load_argv, script_load_arg_len) == -2) {
                                state.close_conn = true;
                        }
 
                        if (status == REDIS_RCODE_SUCCESS) {
                                script_load_done = true;
 
                                /*
                                 *      Verify we got a sane response
                                 */
                                if (reply->type != REDIS_REPLY_STRING) {
                                        REDEBUG("Unexpected reply type after loading function");
                                        fr_redis_reply_print(L_DBG_LVL_OFF, reply, request, 0);
                                        goto script_load_failed;
                                }
 
                                if (strcmp(reply->str, func->digest) != 0) {
                                        REDEBUG("Function digest %s, does not match calculated digest %s", reply->str, func->digest);
                                        goto script_load_failed;
                                }
                                fr_redis_reply_free(&reply);
                                goto again;
                        }
                }
        }
 
        if (s_ret != REDIS_RCODE_SUCCESS) {
                action = XLAT_ACTION_FAIL;
                goto finish;
        }
 
        if (!fr_cond_assert(reply)) {
                action = XLAT_ACTION_FAIL;
                goto finish;
        }
 
        MEM(vb_out = fr_value_box_alloc_null(ctx));
        if (fr_redis_reply_to_value_box(ctx, vb_out, reply, FR_TYPE_VOID, NULL, false, false) < 0) {
                RPERROR("Failed processing reply");
                action = XLAT_ACTION_FAIL;
                goto finish;
        }
        fr_dcursor_append(out, vb_out);
 
finish:
        fr_redis_reply_free(&reply);
 
        return action;
}
 
/** Copies the function configuration into xlat function instance data
 *
 */
static int redis_lua_func_instantiate(xlat_inst_ctx_t const *xctx)
{
        redis_lua_func_inst_t *inst = talloc_get_type_abort(xctx->inst, redis_lua_func_inst_t);
 
        inst->func = talloc_get_type_abort(xctx->uctx, redis_lua_func_t);
 
        return 0;
}
 
static xlat_arg_parser_t const redis_args[] = {
        { .required = true, .concat = true, .type = FR_TYPE_STRING },
        { .variadic = XLAT_ARG_VARIADIC_EMPTY_KEEP, .concat = true, .type = FR_TYPE_STRING },
        XLAT_ARG_PARSER_TERMINATOR
};
 
/** Xlat to make calls to redis
 *
@verbatim
%redis(<redis command>)
@endverbatim
 *
 * @ingroup xlat_functions
 */
static xlat_action_t redis_xlat(TALLOC_CTX *ctx, fr_dcursor_t *out,
                                xlat_ctx_t const *xctx,
                                request_t *request, fr_value_box_list_t *in)
{
        rlm_redis_t const       *inst = talloc_get_type_abort_const(xctx->mctx->mi->data, rlm_redis_t);
        xlat_action_t           action = XLAT_ACTION_DONE;
        fr_redis_conn_t         *conn;
 
        bool                    read_only = false;
        uint8_t const           *key = NULL;
        size_t                  key_len = 0;
 
        fr_redis_cluster_state_t        state;
        fr_redis_rcode_t                status;
 
        redisReply              *reply = NULL;
        int                     s_ret;
 
        fr_value_box_t          *first = fr_value_box_list_head(in);
        fr_sbuff_t              sbuff = FR_SBUFF_IN(first->vb_strvalue, first->vb_length);
 
        int                     argc = 0;
        char const              *argv[MAX_REDIS_ARGS];
        size_t                  arg_len[MAX_REDIS_ARGS];
 
        fr_value_box_t          *vb_out;
 
        if (fr_sbuff_next_if_char(&sbuff, '-')) read_only = true;
 
        /*
         *      Hack to allow querying against a specific node for testing
         */
        if (fr_sbuff_next_if_char(&sbuff, '@')) {
                fr_socket_t     node_addr;
                fr_pool_t       *pool;
 
                RDEBUG3("Overriding node selection");
 
                if (fr_inet_pton_port(&node_addr.inet.dst_ipaddr, &node_addr.inet.dst_port,
                                      fr_sbuff_current(&sbuff), fr_sbuff_remaining(&sbuff),
                                      AF_UNSPEC, true, true) < 0) {
                        RPEDEBUG("Failed parsing node address");
                        return XLAT_ACTION_FAIL;
                }
 
                if (fr_redis_cluster_pool_by_node_addr(&pool, inst->cluster, &node_addr, true) < 0) {
                        RPEDEBUG("Failed locating cluster node");
                        return XLAT_ACTION_FAIL;
                }
 
                conn = fr_pool_connection_get(pool, request);
                if (!conn) {
                        REDEBUG("No connections available for cluster node");
                        return XLAT_ACTION_FAIL;
                }
 
                fr_value_box_list_talloc_free_head(in); /* Remove and free server arg */
 
                fr_value_box_list_foreach(in, vb) {
                        if (argc == NUM_ELEMENTS(argv)) {
                                REDEBUG("Too many arguments (%i)", argc);
                                REXDENT();
                                goto fail;
                        }
 
                        /*
                         *      Fixup null or empty arguments to be
                         *      zero length strings so that the position
                         *      of subsequent arguments are maintained.
                         */
                        if (!fr_type_is_string(vb->type)) {
                                argv[argc] = "";
                                arg_len[argc++] = 0;
                                continue;
                        }
 
                        argv[argc] = vb->vb_strvalue;
                        arg_len[argc++] = vb->vb_length;
                }
 
                RDEBUG2("Executing command: %pV", fr_value_box_list_head(in));
                if (argc > 1) {
                        RDEBUG2("With arguments");
                        RINDENT();
                        for (int i = 1; i < argc; i++) RDEBUG2("[%i] %s", i, argv[i]);
                        REXDENT();
                }
 
                if (redis_command(&status, &reply, request, conn, read_only, argc, argv, arg_len) == -2) {
                        goto close_conn;
                }
 
                if (!reply) goto fail;
 
                switch (status) {
                case REDIS_RCODE_MOVE:
                {
                        fr_value_box_t vb;
 
                        if (fr_redis_reply_to_value_box(NULL, &vb, reply, FR_TYPE_STRING, NULL, false, true) == 0) {
                                REDEBUG("Key served by a different node: %pV", &vb);
                        }
                        goto fail;
                }
 
                case REDIS_RCODE_SUCCESS:
                        goto reply_parse;
 
                case REDIS_RCODE_RECONNECT:
                close_conn:
                        fr_pool_connection_close(pool, request, conn);
                        action = XLAT_ACTION_FAIL;
                        goto finish;
 
                default:
                fail:
                        fr_pool_connection_release(pool, request, conn);
                        action = XLAT_ACTION_FAIL;
                        goto finish;
                }
        }
 
        RDEBUG2("REDIS command arguments");
        RINDENT();
        fr_value_box_list_foreach(in, vb) {
                if (argc == NUM_ELEMENTS(argv)) {
                        REDEBUG("Too many arguments (%i)", argc);
                        REXDENT();
                        goto finish;
                }
 
                argv[argc] = vb->vb_strvalue;
                arg_len[argc] = vb->vb_length;
                argc++;
        }
        REXDENT();
 
        /*
         *      If we've got multiple arguments, the second one is usually the key.
         *      The Redis docs say commands should be analysed first to get key
         *      positions, but this involves sending them to the server, which is
         *      just as expensive as sending them to the wrong server and receiving
         *      a redirect.
         */
        if (argc > 1) {
                key = (uint8_t const *)argv[1];
                key_len = arg_len[1];
        }
 
        for (s_ret = fr_redis_cluster_state_init(&state, &conn, inst->cluster, request, key, key_len, read_only);
             s_ret == REDIS_RCODE_TRY_AGAIN;    /* Continue */
             s_ret = fr_redis_cluster_state_next(&state, &conn, inst->cluster, request, status, &reply)) {
                RDEBUG2("Executing command: %pV", fr_value_box_list_head(in));
                if (argc > 1) {
                        RDEBUG2("With arguments");
                        RINDENT();
                        for (int i = 1; i < argc; i++) RDEBUG2("[%i] %s", i, argv[i]);
                        REXDENT();
                }
 
                if (redis_command(&status, &reply, request, conn, read_only, argc, argv, arg_len) == -2) {
                        state.close_conn = true;
                }
        }
        if (s_ret != REDIS_RCODE_SUCCESS) {
                action = XLAT_ACTION_FAIL;
                goto finish;
        }
 
        if (!fr_cond_assert(reply)) {
                action = XLAT_ACTION_FAIL;
                goto finish;
        }
 
reply_parse:
        MEM(vb_out = fr_value_box_alloc_null(ctx));
        if (fr_redis_reply_to_value_box(ctx, vb_out, reply, FR_TYPE_VOID, NULL, false, false) < 0) {
                RPERROR("Failed processing reply");
                action = XLAT_ACTION_FAIL;
                goto finish;
        }
        fr_dcursor_append(out, vb_out);
 
finish:
        fr_redis_reply_free(&reply);
 
        return action;
}
 
static int mod_instantiate(module_inst_ctx_t const *mctx)
{
        rlm_redis_t *inst = talloc_get_type_abort(mctx->mi->data, rlm_redis_t);
        fr_socket_t *nodes;
        int ret, i;
 
        inst->cluster = fr_redis_cluster_alloc(inst, mctx->mi->conf, &inst->conf, true, NULL, NULL, NULL);
        if (!inst->cluster) return -1;
 
        /*
         *      Best effort - Try and load in scripts on startup
         */
        if (talloc_array_length(inst->lua.funcs) == 0) return 0;
 
        ret = fr_redis_cluster_node_addr_by_role(NULL, &nodes, inst->cluster, true, true);
        if (ret <= 0) return 0;
 
        for (i = 0; i < ret; i++) {
                fr_pool_t               *pool;
                fr_redis_conn_t         *conn;
 
                if (fr_redis_cluster_pool_by_node_addr(&pool, inst->cluster, &nodes[i], true) < 0) {
                        talloc_free(nodes);
                        return 0;
                }
 
                conn = fr_pool_connection_get(pool, 0);
                if (!conn) continue;
 
                talloc_foreach(inst->lua.funcs, func) {
                        char const      *script_load_argv[] = {
                                                "SCRIPT",
                                                "LOAD",
                                                func->body
                                        };
 
                        size_t          script_load_arg_len[] = {
                                                (sizeof("SCRIPT") - 1),
                                                (sizeof("LOAD") - 1),
                                                (talloc_array_length(func->body) - 1)
                                        };
 
                        fr_redis_rcode_t status;
                        redisReply *reply;
 
                        /*
                         *      preload onto every node, even replicas.
                         */
                        if (redis_command(&status, &reply, NULL, conn, false,
                                          NUM_ELEMENTS(script_load_argv), script_load_argv, script_load_arg_len) == -2) {
                        error:
                                fr_pool_connection_release(pool, NULL, conn);
                                talloc_free(nodes);
                                return -1;
                        }
 
                        fr_redis_reply_free(&reply);
 
                        /*
                         *      Only error on explicit errors, not on connectivity issues
                         */
                        switch (status) {
                        case REDIS_RCODE_ERROR:
                                PERROR("Loading lua function \"%s\" onto node failed", func->name);
                                goto error;
 
                        case REDIS_RCODE_SUCCESS:
                                DEBUG2("Loaded lua function \"%s\" onto node", func->name);
                                break;
 
                        default:
                                PWARN("Loading lua function \"%s\" onto node failed", func->name);
                                continue;
                        }
                }
 
                fr_pool_connection_release(pool, NULL, conn);
        }
        talloc_free(nodes);
 
        return 0;
}
 
static int mod_bootstrap(module_inst_ctx_t const *mctx)
{
        rlm_redis_t const       *inst = talloc_get_type_abort(mctx->mi->data, rlm_redis_t);
        xlat_t                  *xlat;
 
        xlat = module_rlm_xlat_register(mctx->mi->boot, mctx, NULL, redis_xlat, FR_TYPE_VOID);
        xlat_func_args_set(xlat, redis_args);
 
        /*
         *      %redis.node(<key>[, idx])
         */
        if (unlikely((xlat = module_rlm_xlat_register(mctx->mi->boot, mctx, "node", redis_node_xlat, FR_TYPE_STRING)) == NULL)) return -1;
        xlat_func_args_set(xlat, redis_node_xlat_args);
 
        if (unlikely((xlat = module_rlm_xlat_register(mctx->mi->boot, mctx, "remap", redis_remap_xlat, FR_TYPE_STRING)) == NULL)) return -1;
        xlat_func_args_set(xlat, redis_remap_xlat_args);
 
        /*
         *      Loop over the lua functions, registering an xlat
         *      that'll call that function specifically.
         */
        talloc_foreach(inst->lua.funcs, func) {
                if (unlikely((xlat = module_rlm_xlat_register(mctx->mi->boot, mctx, func->name, redis_lua_func_xlat, FR_TYPE_VOID)) == NULL)) return -1;
                xlat_func_args_set(xlat, redis_lua_func_args);
                xlat_func_instantiate_set(xlat, redis_lua_func_instantiate, redis_lua_func_inst_t, NULL, func);
        }
 
        return 0;
}
 
static int mod_load(void)
{
        fr_redis_version_print();
 
        return 0;
}
 
extern module_rlm_t rlm_redis;
module_rlm_t rlm_redis = {
        .common = {
                .magic          = MODULE_MAGIC_INIT,
                .name           = "redis",
                .inst_size      = sizeof(rlm_redis_t),
                .config         = module_config,
                .onload         = mod_load,
                .bootstrap      = mod_bootstrap,
                .instantiate    = mod_instantiate
        }
};