support for redis cachedb backend

cachedb/cachedb.c

@@ -43,6 +43,7 @@
 #include "config.h"
 #ifdef USE_CACHEDB
 #include "cachedb/cachedb.h"
+#include "util/alloc.h"
 #include "util/regional.h"
 #include "util/net_help.h"
 #include "util/config_file.h"
@@ -56,6 +57,25 @@
 #include "sldns/wire2str.h"
 #include "sldns/sbuffer.h"
 
+/* header file for htobe64 */
+#ifdef HAVE_ENDIAN_H
+#  include <endian.h>
+#endif
+#ifdef HAVE_SYS_ENDIAN_H
+#  include <sys/endian.h>
+#endif
+#ifdef HAVE_LIBKERN_OSBYTEORDER_H
+/* In practice this is specific to MacOS X.  We assume it doesn't have
+* htobe64/be64toh but has alternatives with a different name. */
+#  include <libkern/OSByteOrder.h>
+#  define htobe64(x) OSSwapHostToBigInt64(x)
+#  define be64toh(x) OSSwapBigToHostInt64(x)
+#endif
+
+#ifdef USE_REDIS
+#include "hiredis/hiredis.h"
+#endif
+
 #define CACHEDB_HASHSIZE 256 /* bit hash */
 
 /** the unit test testframe for cachedb, its module state contains
@@ -167,15 +187,252 @@ testframe_store(struct module_env* env, struct cachedb_env* cachedb_env,
 	/* (key,data) successfully stored */
 }
 
+#ifdef USE_REDIS
+struct redis_moddata {
+	redisContext** ctxs;	/* thread-specific redis contexts */
+	int numctxs;		/* number of ctx entries */
+	const char* server_host; /* server's IP address or host name */
+	int server_port;	 /* server's TCP port */
+	struct timeval timeout;	 /* timeout for connection setup and commands */
+};
+
+static redisContext*
+redis_connect(const struct redis_moddata* moddata)
+{
+	redisContext* ctx;
+
+	ctx = redisConnectWithTimeout(moddata->server_host,
+		moddata->server_port, moddata->timeout);
+	if(!ctx || ctx->err) {
+		const char *errstr = "out of memory";
+		if(ctx)
+			errstr = ctx->errstr;
+		log_err("failed to connect to redis server: %s", errstr);
+		goto fail;
+	}
+	if(redisSetTimeout(ctx, moddata->timeout) != REDIS_OK) {
+		log_err("failed to set redis timeout");
+		goto fail;
+	}
+	return ctx;
+
+  fail:
+	if(ctx)
+		redisFree(ctx);
+	return NULL;
+}
+
+static int
+redis_init(struct module_env* env, struct cachedb_env* cachedb_env)
+{
+	int i;
+	struct redis_moddata* moddata = NULL;
+
+	verbose(VERB_ALGO, "redis_init");
+
+	moddata = calloc(1, sizeof(struct redis_moddata));
+	if(!moddata) {
+		log_err("out of memory");
+		return 0;
+	}
+	moddata->numctxs = env->cfg->num_threads;
+	moddata->ctxs = calloc(env->cfg->num_threads, sizeof(redisContext*));
+	if(!moddata->ctxs) {
+		log_err("out of memory");
+		free(moddata);
+		return 0;
+	}
+	/* note: server_host is a shallow reference to configured string.
+	 * we don't have to free it in this module. */
+	moddata->server_host = env->cfg->redis_server_host;
+	moddata->server_port = env->cfg->redis_server_port;
+	moddata->timeout.tv_sec = env->cfg->redis_timeout / 1000;
+	moddata->timeout.tv_usec = (env->cfg->redis_timeout % 1000) * 1000;
+	for(i = 0; i < moddata->numctxs; i++)
+		moddata->ctxs[i] = redis_connect(moddata);
+	cachedb_env->backend_data = moddata;
+	return 1;
+}
+
+static void
+redis_deinit(struct module_env* env, struct cachedb_env* cachedb_env)
+{
+	struct redis_moddata* moddata = (struct redis_moddata*)
+		cachedb_env->backend_data;
+	(void)env;
+
+	verbose(VERB_ALGO, "redis_deinit");
+
+	if(!moddata)
+		return;
+	if(moddata->ctxs) {
+		int i;
+		for(i = 0; i < moddata->numctxs; i++) {
+			if(moddata->ctxs[i])
+				redisFree(moddata->ctxs[i]);
+		}
+		free(moddata->ctxs);
+	}
+	free(moddata);
+}
+
+/*
+ * Send a redis command and get a reply.  Unified so that it can be used for
+ * both SET and GET.  If 'data' is non-NULL the command is supposed to be
+ * SET and GET otherwise, but the implementation of this function is agnostic
+ * about the semantics (except for logging): 'command', 'data', and 'data_len'
+ * are opaquely passed to redisCommand().
+ * This function first checks whether a connection with a redis server has
+ * been established; if not it tries to set up a new one.
+ * It returns redisReply returned from redisCommand() or NULL if some low
+ * level error happens.  The caller is responsible to check the return value,
+ * if it's non-NULL, it has to free it with freeReplyObject().
+ */
+static redisReply*
+redis_command(struct module_env* env, struct cachedb_env* cachedb_env,
+	const char* command, const uint8_t* data, size_t data_len)
+{
+	redisContext* ctx;
+	redisReply* rep;
+	struct redis_moddata* d = (struct redis_moddata*)
+		cachedb_env->backend_data;
+
+	/* We assume env->alloc->thread_num is a unique ID for each thread
+	 * in [0, num-of-threads).  We could treat it as an error condition
+	 * if the assumption didn't hold, but it seems to be a fundamental
+	 * assumption throughout the unbound architecture, so we simply assert
+	 * it. */
+	log_assert(env->alloc->thread_num < d->numctxs);
+	ctx = d->ctxs[env->alloc->thread_num];
+
+	/* If we've not established a connection to the server or we've closed
+	 * it on a failure, try to re-establish a new one.   Failures will be
+	 * logged in redis_connect(). */
+	if(!ctx) {
+		ctx = redis_connect(d);
+		d->ctxs[env->alloc->thread_num] = ctx;
+	}
+	if(!ctx)
+		return NULL;
+
+	/* Send the command and get a reply, synchronously. */
+	rep = (redisReply*)redisCommand(ctx, command, data, data_len);
+	if(!rep) {
+		/* Once an error as a NULL-reply is returned the context cannot
+		 * be reused and we'll need to set up a new connection. */
+		log_err("redis_command: failed to receive a reply, "
+			"closing connection: %s", ctx->errstr);
+		redisFree(ctx);
+		d->ctxs[env->alloc->thread_num] = NULL;
+		return NULL;
+	}
+
+	/* Check error in reply to unify logging in that case.
+	 * The caller may perform context-dependent checks and logging. */
+	if(rep->type == REDIS_REPLY_ERROR)
+		log_err("redis: %s resulted in an error: %s",
+			data ? "set" : "get", rep->str);
+
+	return rep;
+}
+
+static int
+redis_lookup(struct module_env* env, struct cachedb_env* cachedb_env,
+	char* key, struct sldns_buffer* result_buffer)
+{
+	redisReply* rep;
+	char cmdbuf[4+(CACHEDB_HASHSIZE/8)*2+1]; /* "GET " + key */
+	int n;
+	int ret = 0;
+
+	verbose(VERB_ALGO, "redis_lookup of %s", key);
+
+	n = snprintf(cmdbuf, sizeof(cmdbuf), "GET %s", key);
+	if(n < 0 || n >= (int)sizeof(cmdbuf)) {
+		log_err("redis_lookup: unexpected failure to build command");
+		return 0;
+	}
+
+	rep = redis_command(env, cachedb_env, cmdbuf, NULL, 0);
+	if(!rep)
+		return 0;
+	switch (rep->type) {
+	case REDIS_REPLY_NIL:
+		verbose(VERB_ALGO, "redis_lookup: no data cached");
+		break;
+	case REDIS_REPLY_STRING:
+		verbose(VERB_ALGO, "redis_lookup found %d bytes",
+			(int)rep->len);
+		if((size_t)rep->len > sldns_buffer_capacity(result_buffer)) {
+			log_err("redis_lookup: replied data too long: %lu",
+				(size_t)rep->len);
+			break;
+		}
+		sldns_buffer_clear(result_buffer);
+		sldns_buffer_write(result_buffer, rep->str, rep->len);
+		sldns_buffer_flip(result_buffer);
+		ret = 1;
+		break;
+	case REDIS_REPLY_ERROR:
+		break;		/* already logged */
+	default:
+		log_err("redis_lookup: unexpected type of reply for (%d)",
+			rep->type);
+		break;
+	}
+	freeReplyObject(rep);
+	return ret;
+}
+
+static void
+redis_store(struct module_env* env, struct cachedb_env* cachedb_env,
+	char* key, uint8_t* data, size_t data_len)
+{
+	redisReply* rep;
+	char cmdbuf[4+(CACHEDB_HASHSIZE/8)*2+3+1]; /* "SET " + key + " %b" */
+	int n;
+
+	verbose(VERB_ALGO, "redis_store %s (%d bytes)", key, (int)data_len);
+
+	/* build command to set to a binary safe string */
+	n = snprintf(cmdbuf, sizeof(cmdbuf), "SET %s %%b", key);
+	if(n < 0 || n >= (int)sizeof(cmdbuf)) {
+		log_err("redis_store: unexpected failure to build command");
+		return;
+	}
+
+	rep = redis_command(env, cachedb_env, cmdbuf, data, data_len);
+	if(rep) {
+		verbose(VERB_ALGO, "redis_store set completed");
+		if(rep->type != REDIS_REPLY_STATUS &&
+			rep->type != REDIS_REPLY_ERROR) {
+			log_err("redis_store: unexpected type of reply (%d)",
+				rep->type);
+		}
+		freeReplyObject(rep);
+	}
+}
+#endif	/* USE_REDIS */
+
 /** The testframe backend is for unit tests */
 static struct cachedb_backend testframe_backend = { "testframe",
 	testframe_init, testframe_deinit, testframe_lookup, testframe_store
 };
 
+#ifdef USE_REDIS
+static struct cachedb_backend redis_backend = { "redis",
+	redis_init, redis_deinit, redis_lookup, redis_store
+};
+#endif
+
 /** find a particular backend from possible backends */
 static struct cachedb_backend*
 cachedb_find_backend(const char* str)
 {
+#ifdef USE_REDIS
+	if(strcmp(str, redis_backend.name) == 0)
+		return &redis_backend;
+#endif
 	if(strcmp(str, testframe_backend.name) == 0)
 		return &testframe_backend;
 	/* TODO add more backends here */

config.h.in

@@ -95,6 +95,10 @@
    don't. */
 #undef HAVE_DECL_REALLOCARRAY
 
+/* Define to 1 if you have the declaration of `redisConnect', and to 0 if you
+   don't. */
+#undef HAVE_DECL_REDISCONNECT
+
 /* Define to 1 if you have the declaration of `sk_SSL_COMP_pop_free', and to 0
    if you don't. */
 #undef HAVE_DECL_SK_SSL_COMP_POP_FREE
@@ -128,6 +132,13 @@
 /* Define to 1 if you have the <endian.h> header file. */
 #undef HAVE_ENDIAN_H
 
+/* Define to 1 if you have the <sys/endian.h> header file. */
+#undef HAVE_SYS_ENDIAN_H
+
+/* Define to 1 if you have the <libkern/OSByteOrder.h> header file, which is
+ * in effect MacOS X specific. */
+#undef HAVE_LIBKERN_OSBYTEORDER_H
+
 /* Define to 1 if you have the `endprotoent' function. */
 #undef HAVE_ENDPROTOENT
 
@@ -718,6 +729,9 @@
 /* Define this to enable client TCP Fast Open. */
 #undef USE_OSX_MSG_FASTOPEN
 
+/* Define this to use hiredis client. */
+#undef USE_REDIS
+
 /* Define this to enable SHA1 support. */
 #undef USE_SHA1