Unverified Commit efb6495a authored by Salvatore Sanfilippo's avatar Salvatore Sanfilippo Committed by GitHub
Browse files

Merge pull request #6236 from yossigo/poc/conns

Abstract Connections I/O API & TLS Support
parents b8e02f2b 8e29b0b2
......@@ -47,6 +47,10 @@
#include <math.h>
#include <hiredis.h>
#ifdef USE_OPENSSL
#include <openssl/ssl.h>
#include <hiredis_ssl.h>
#endif
#include <sds.h> /* use sds.h from hiredis, so that only one set of sds functions will be present in the binary */
#include "dict.h"
#include "adlist.h"
......@@ -188,6 +192,12 @@ static struct config {
char *hostip;
int hostport;
char *hostsocket;
int tls;
char *sni;
char *cacert;
char *cacertdir;
char *cert;
char *key;
long repeat;
long interval;
int dbnum;
......@@ -758,6 +768,71 @@ static int cliSelect(void) {
return REDIS_ERR;
}
/* Wrapper around redisSecureConnection to avoid hiredis_ssl dependencies if
* not building with TLS support.
*/
static int cliSecureConnection(redisContext *c, const char **err) {
#ifdef USE_OPENSSL
static SSL_CTX *ssl_ctx = NULL;
if (!ssl_ctx) {
ssl_ctx = SSL_CTX_new(SSLv23_client_method());
if (!ssl_ctx) {
*err = "Failed to create SSL_CTX";
goto error;
}
SSL_CTX_set_options(ssl_ctx, SSL_OP_NO_SSLv2 | SSL_OP_NO_SSLv3);
SSL_CTX_set_verify(ssl_ctx, SSL_VERIFY_PEER, NULL);
if (config.cacert || config.cacertdir) {
if (!SSL_CTX_load_verify_locations(ssl_ctx, config.cacert, config.cacertdir)) {
*err = "Invalid CA Certificate File/Directory";
goto error;
}
} else {
if (!SSL_CTX_set_default_verify_paths(ssl_ctx)) {
*err = "Failed to use default CA paths";
goto error;
}
}
if (config.cert && !SSL_CTX_use_certificate_chain_file(ssl_ctx, config.cert)) {
*err = "Invalid client certificate";
goto error;
}
if (config.key && !SSL_CTX_use_PrivateKey_file(ssl_ctx, config.key, SSL_FILETYPE_PEM)) {
*err = "Invalid private key";
goto error;
}
}
SSL *ssl = SSL_new(ssl_ctx);
if (!ssl) {
*err = "Failed to create SSL object";
return REDIS_ERR;
}
if (config.sni && !SSL_set_tlsext_host_name(ssl, config.sni)) {
*err = "Failed to configure SNI";
SSL_free(ssl);
return REDIS_ERR;
}
return redisInitiateSSL(c, ssl);
error:
SSL_CTX_free(ssl_ctx);
ssl_ctx = NULL;
return REDIS_ERR;
#else
(void) c;
(void) err;
return REDIS_OK;
#endif
}
/* Select RESP3 mode if redis-cli was started with the -3 option. */
static int cliSwitchProto(void) {
redisReply *reply;
......@@ -789,6 +864,16 @@ static int cliConnect(int flags) {
context = redisConnectUnix(config.hostsocket);
}
if (!context->err && config.tls) {
const char *err = NULL;
if (cliSecureConnection(context, &err) == REDIS_ERR && err) {
fprintf(stderr, "Could not negotiate a TLS connection: %s\n", err);
context = NULL;
redisFree(context);
return REDIS_ERR;
}
}
if (context->err) {
if (!(flags & CC_QUIET)) {
fprintf(stderr,"Could not connect to Redis at ");
......@@ -804,6 +889,7 @@ static int cliConnect(int flags) {
return REDIS_ERR;
}
/* Set aggressive KEEP_ALIVE socket option in the Redis context socket
* in order to prevent timeouts caused by the execution of long
* commands. At the same time this improves the detection of real
......@@ -1305,6 +1391,13 @@ static redisReply *reconnectingRedisCommand(redisContext *c, const char *fmt, ..
redisFree(c);
c = redisConnect(config.hostip,config.hostport);
if (!c->err && config.tls) {
const char *err = NULL;
if (cliSecureConnection(c, &err) == REDIS_ERR && err) {
fprintf(stderr, "TLS Error: %s\n", err);
exit(1);
}
}
usleep(1000000);
}
......@@ -1498,6 +1591,20 @@ static int parseOptions(int argc, char **argv) {
} else if (!strcmp(argv[i],"--cluster-search-multiple-owners")) {
config.cluster_manager_command.flags |=
CLUSTER_MANAGER_CMD_FLAG_CHECK_OWNERS;
#ifdef USE_OPENSSL
} else if (!strcmp(argv[i],"--tls")) {
config.tls = 1;
} else if (!strcmp(argv[i],"--sni")) {
config.sni = argv[++i];
} else if (!strcmp(argv[i],"--cacertdir")) {
config.cacertdir = argv[++i];
} else if (!strcmp(argv[i],"--cacert")) {
config.cacert = argv[++i];
} else if (!strcmp(argv[i],"--cert")) {
config.cert = argv[++i];
} else if (!strcmp(argv[i],"--key")) {
config.key = argv[++i];
#endif
} else if (!strcmp(argv[i],"-v") || !strcmp(argv[i], "--version")) {
sds version = cliVersion();
printf("redis-cli %s\n", version);
......@@ -1591,6 +1698,15 @@ static void usage(void) {
" -x Read last argument from STDIN.\n"
" -d <delimiter> Multi-bulk delimiter in for raw formatting (default: \\n).\n"
" -c Enable cluster mode (follow -ASK and -MOVED redirections).\n"
#ifdef USE_OPENSSL
" --tls Establish a secure TLS connection.\n"
" --cacert CA Certificate file to verify with.\n"
" --cacertdir Directory where trusted CA certificates are stored.\n"
" If neither cacert nor cacertdir are specified, the default\n"
" system-wide trusted root certs configuration will apply.\n"
" --cert Client certificate to authenticate with.\n"
" --key Private key file to authenticate with.\n"
#endif
" --raw Use raw formatting for replies (default when STDOUT is\n"
" not a tty).\n"
" --no-raw Force formatted output even when STDOUT is not a tty.\n"
......@@ -1615,7 +1731,9 @@ static void usage(void) {
" --pipe Transfer raw Redis protocol from stdin to server.\n"
" --pipe-timeout <n> In --pipe mode, abort with error if after sending all data.\n"
" no reply is received within <n> seconds.\n"
" Default timeout: %d. Use 0 to wait forever.\n"
" Default timeout: %d. Use 0 to wait forever.\n",
REDIS_CLI_DEFAULT_PIPE_TIMEOUT);
fprintf(stderr,
" --bigkeys Sample Redis keys looking for keys with many elements (complexity).\n"
" --memkeys Sample Redis keys looking for keys consuming a lot of memory.\n"
" --memkeys-samples <n> Sample Redis keys looking for keys consuming a lot of memory.\n"
......@@ -1638,8 +1756,7 @@ static void usage(void) {
" line interface.\n"
" --help Output this help and exit.\n"
" --version Output version and exit.\n"
"\n",
REDIS_CLI_DEFAULT_PIPE_TIMEOUT);
"\n");
/* Using another fprintf call to avoid -Woverlength-strings compile warning */
fprintf(stderr,
"Cluster Manager Commands:\n"
......@@ -2407,6 +2524,15 @@ cleanup:
static int clusterManagerNodeConnect(clusterManagerNode *node) {
if (node->context) redisFree(node->context);
node->context = redisConnect(node->ip, node->port);
if (!node->context->err && config.tls) {
const char *err = NULL;
if (cliSecureConnection(node->context, &err) == REDIS_ERR && err) {
fprintf(stderr,"TLS Error: %s\n", err);
redisFree(node->context);
node->context = NULL;
return 0;
}
}
if (node->context->err) {
fprintf(stderr,"Could not connect to Redis at ");
fprintf(stderr,"%s:%d: %s\n", node->ip, node->port,
......
This diff is collapsed.
......@@ -159,13 +159,13 @@ void rioInitWithFile(rio *r, FILE *fp) {
r->io.file.autosync = 0;
}
/* ------------------- File descriptor implementation -------------------
/* ------------------- Connection implementation -------------------
* We use this RIO implemetnation when reading an RDB file directly from
* the socket to the memory via rdbLoadRio(), thus this implementation
* only implements reading from a file descriptor that is, normally,
* the connection to the memory via rdbLoadRio(), thus this implementation
* only implements reading from a connection that is, normally,
* just a socket. */
static size_t rioFdWrite(rio *r, const void *buf, size_t len) {
static size_t rioConnWrite(rio *r, const void *buf, size_t len) {
UNUSED(r);
UNUSED(buf);
UNUSED(len);
......@@ -173,72 +173,72 @@ static size_t rioFdWrite(rio *r, const void *buf, size_t len) {
}
/* Returns 1 or 0 for success/failure. */
static size_t rioFdRead(rio *r, void *buf, size_t len) {
size_t avail = sdslen(r->io.fd.buf)-r->io.fd.pos;
static size_t rioConnRead(rio *r, void *buf, size_t len) {
size_t avail = sdslen(r->io.conn.buf)-r->io.conn.pos;
/* If the buffer is too small for the entire request: realloc. */
if (sdslen(r->io.fd.buf) + sdsavail(r->io.fd.buf) < len)
r->io.fd.buf = sdsMakeRoomFor(r->io.fd.buf, len - sdslen(r->io.fd.buf));
if (sdslen(r->io.conn.buf) + sdsavail(r->io.conn.buf) < len)
r->io.conn.buf = sdsMakeRoomFor(r->io.conn.buf, len - sdslen(r->io.conn.buf));
/* If the remaining unused buffer is not large enough: memmove so that we
* can read the rest. */
if (len > avail && sdsavail(r->io.fd.buf) < len - avail) {
sdsrange(r->io.fd.buf, r->io.fd.pos, -1);
r->io.fd.pos = 0;
if (len > avail && sdsavail(r->io.conn.buf) < len - avail) {
sdsrange(r->io.conn.buf, r->io.conn.pos, -1);
r->io.conn.pos = 0;
}
/* If we don't already have all the data in the sds, read more */
while (len > sdslen(r->io.fd.buf) - r->io.fd.pos) {
size_t buffered = sdslen(r->io.fd.buf) - r->io.fd.pos;
while (len > sdslen(r->io.conn.buf) - r->io.conn.pos) {
size_t buffered = sdslen(r->io.conn.buf) - r->io.conn.pos;
size_t toread = len - buffered;
/* Read either what's missing, or PROTO_IOBUF_LEN, the bigger of
* the two. */
if (toread < PROTO_IOBUF_LEN) toread = PROTO_IOBUF_LEN;
if (toread > sdsavail(r->io.fd.buf)) toread = sdsavail(r->io.fd.buf);
if (r->io.fd.read_limit != 0 &&
r->io.fd.read_so_far + buffered + toread > r->io.fd.read_limit)
if (toread > sdsavail(r->io.conn.buf)) toread = sdsavail(r->io.conn.buf);
if (r->io.conn.read_limit != 0 &&
r->io.conn.read_so_far + buffered + toread > r->io.conn.read_limit)
{
if (r->io.fd.read_limit >= r->io.fd.read_so_far - buffered)
toread = r->io.fd.read_limit - r->io.fd.read_so_far - buffered;
if (r->io.conn.read_limit >= r->io.conn.read_so_far - buffered)
toread = r->io.conn.read_limit - r->io.conn.read_so_far - buffered;
else {
errno = EOVERFLOW;
return 0;
}
}
int retval = read(r->io.fd.fd,
(char*)r->io.fd.buf + sdslen(r->io.fd.buf),
int retval = connRead(r->io.conn.conn,
(char*)r->io.conn.buf + sdslen(r->io.conn.buf),
toread);
if (retval <= 0) {
if (errno == EWOULDBLOCK) errno = ETIMEDOUT;
return 0;
}
sdsIncrLen(r->io.fd.buf, retval);
sdsIncrLen(r->io.conn.buf, retval);
}
memcpy(buf, (char*)r->io.fd.buf + r->io.fd.pos, len);
r->io.fd.read_so_far += len;
r->io.fd.pos += len;
memcpy(buf, (char*)r->io.conn.buf + r->io.conn.pos, len);
r->io.conn.read_so_far += len;
r->io.conn.pos += len;
return len;
}
/* Returns read/write position in file. */
static off_t rioFdTell(rio *r) {
return r->io.fd.read_so_far;
static off_t rioConnTell(rio *r) {
return r->io.conn.read_so_far;
}
/* Flushes any buffer to target device if applicable. Returns 1 on success
* and 0 on failures. */
static int rioFdFlush(rio *r) {
static int rioConnFlush(rio *r) {
/* Our flush is implemented by the write method, that recognizes a
* buffer set to NULL with a count of zero as a flush request. */
return rioFdWrite(r,NULL,0);
return rioConnWrite(r,NULL,0);
}
static const rio rioFdIO = {
rioFdRead,
rioFdWrite,
rioFdTell,
rioFdFlush,
static const rio rioConnIO = {
rioConnRead,
rioConnWrite,
rioConnTell,
rioConnFlush,
NULL, /* update_checksum */
0, /* current checksum */
0, /* flags */
......@@ -249,108 +249,90 @@ static const rio rioFdIO = {
/* Create an RIO that implements a buffered read from an fd
* read_limit argument stops buffering when the reaching the limit. */
void rioInitWithFd(rio *r, int fd, size_t read_limit) {
*r = rioFdIO;
r->io.fd.fd = fd;
r->io.fd.pos = 0;
r->io.fd.read_limit = read_limit;
r->io.fd.read_so_far = 0;
r->io.fd.buf = sdsnewlen(NULL, PROTO_IOBUF_LEN);
sdsclear(r->io.fd.buf);
void rioInitWithConn(rio *r, connection *conn, size_t read_limit) {
*r = rioConnIO;
r->io.conn.conn = conn;
r->io.conn.pos = 0;
r->io.conn.read_limit = read_limit;
r->io.conn.read_so_far = 0;
r->io.conn.buf = sdsnewlen(NULL, PROTO_IOBUF_LEN);
sdsclear(r->io.conn.buf);
}
/* Release the RIO tream. Optionally returns the unread buffered data
* when the SDS pointer 'remaining' is passed. */
void rioFreeFd(rio *r, sds *remaining) {
if (remaining && (size_t)r->io.fd.pos < sdslen(r->io.fd.buf)) {
if (r->io.fd.pos > 0) sdsrange(r->io.fd.buf, r->io.fd.pos, -1);
*remaining = r->io.fd.buf;
void rioFreeConn(rio *r, sds *remaining) {
if (remaining && (size_t)r->io.conn.pos < sdslen(r->io.conn.buf)) {
if (r->io.conn.pos > 0) sdsrange(r->io.conn.buf, r->io.conn.pos, -1);
*remaining = r->io.conn.buf;
} else {
sdsfree(r->io.fd.buf);
sdsfree(r->io.conn.buf);
if (remaining) *remaining = NULL;
}
r->io.fd.buf = NULL;
r->io.conn.buf = NULL;
}
/* ------------------- File descriptors set implementation ------------------
* This target is used to write the RDB file to N different replicas via
* sockets, when the master just streams the data to the replicas without
* creating an RDB on-disk image (diskless replication option).
/* ------------------- File descriptor implementation ------------------
* This target is used to write the RDB file to pipe, when the master just
* streams the data to the replicas without creating an RDB on-disk image
* (diskless replication option).
* It only implements writes. */
/* Returns 1 or 0 for success/failure.
* The function returns success as long as we are able to correctly write
* to at least one file descriptor.
*
* When buf is NULL and len is 0, the function performs a flush operation
* if there is some pending buffer, so this function is also used in order
* to implement rioFdsetFlush(). */
static size_t rioFdsetWrite(rio *r, const void *buf, size_t len) {
* to implement rioFdFlush(). */
static size_t rioFdWrite(rio *r, const void *buf, size_t len) {
ssize_t retval;
int j;
unsigned char *p = (unsigned char*) buf;
int doflush = (buf == NULL && len == 0);
/* To start we always append to our buffer. If it gets larger than
* a given size, we actually write to the sockets. */
if (len) {
r->io.fdset.buf = sdscatlen(r->io.fdset.buf,buf,len);
len = 0; /* Prevent entering the while below if we don't flush. */
if (sdslen(r->io.fdset.buf) > PROTO_IOBUF_LEN) doflush = 1;
/* For small writes, we rather keep the data in user-space buffer, and flush
* it only when it grows. however for larger writes, we prefer to flush
* any pre-existing buffer, and write the new one directly without reallocs
* and memory copying. */
if (len > PROTO_IOBUF_LEN) {
/* First, flush any pre-existing buffered data. */
if (sdslen(r->io.fd.buf)) {
if (rioFdWrite(r, NULL, 0) == 0)
return 0;
}
if (doflush) {
p = (unsigned char*) r->io.fdset.buf;
len = sdslen(r->io.fdset.buf);
/* Write the new data, keeping 'p' and 'len' from the input. */
} else {
if (len) {
r->io.fd.buf = sdscatlen(r->io.fd.buf,buf,len);
if (sdslen(r->io.fd.buf) > PROTO_IOBUF_LEN)
doflush = 1;
if (!doflush)
return 1;
}
/* Write in little chunchs so that when there are big writes we
* parallelize while the kernel is sending data in background to
* the TCP socket. */
while(len) {
size_t count = len < 1024 ? len : 1024;
int broken = 0;
for (j = 0; j < r->io.fdset.numfds; j++) {
if (r->io.fdset.state[j] != 0) {
/* Skip FDs alraedy in error. */
broken++;
continue;
/* Flusing the buffered data. set 'p' and 'len' accordintly. */
p = (unsigned char*) r->io.fd.buf;
len = sdslen(r->io.fd.buf);
}
/* Make sure to write 'count' bytes to the socket regardless
* of short writes. */
size_t nwritten = 0;
while(nwritten != count) {
retval = write(r->io.fdset.fds[j],p+nwritten,count-nwritten);
while(nwritten != len) {
retval = write(r->io.fd.fd,p+nwritten,len-nwritten);
if (retval <= 0) {
/* With blocking sockets, which is the sole user of this
/* With blocking io, which is the sole user of this
* rio target, EWOULDBLOCK is returned only because of
* the SO_SNDTIMEO socket option, so we translate the error
* into one more recognizable by the user. */
if (retval == -1 && errno == EWOULDBLOCK) errno = ETIMEDOUT;
break;
return 0; /* error. */
}
nwritten += retval;
}
if (nwritten != count) {
/* Mark this FD as broken. */
r->io.fdset.state[j] = errno;
if (r->io.fdset.state[j] == 0) r->io.fdset.state[j] = EIO;
}
}
if (broken == r->io.fdset.numfds) return 0; /* All the FDs in error. */
p += count;
len -= count;
r->io.fdset.pos += count;
}
if (doflush) sdsclear(r->io.fdset.buf);
r->io.fd.pos += len;
sdsclear(r->io.fd.buf);
return 1;
}
/* Returns 1 or 0 for success/failure. */
static size_t rioFdsetRead(rio *r, void *buf, size_t len) {
static size_t rioFdRead(rio *r, void *buf, size_t len) {
UNUSED(r);
UNUSED(buf);
UNUSED(len);
......@@ -358,23 +340,23 @@ static size_t rioFdsetRead(rio *r, void *buf, size_t len) {
}
/* Returns read/write position in file. */
static off_t rioFdsetTell(rio *r) {
return r->io.fdset.pos;
static off_t rioFdTell(rio *r) {
return r->io.fd.pos;
}
/* Flushes any buffer to target device if applicable. Returns 1 on success
* and 0 on failures. */
static int rioFdsetFlush(rio *r) {
static int rioFdFlush(rio *r) {
/* Our flush is implemented by the write method, that recognizes a
* buffer set to NULL with a count of zero as a flush request. */
return rioFdsetWrite(r,NULL,0);
return rioFdWrite(r,NULL,0);
}
static const rio rioFdsetIO = {
rioFdsetRead,
rioFdsetWrite,
rioFdsetTell,
rioFdsetFlush,
static const rio rioFdIO = {
rioFdRead,
rioFdWrite,
rioFdTell,
rioFdFlush,
NULL, /* update_checksum */
0, /* current checksum */
0, /* flags */
......@@ -383,24 +365,16 @@ static const rio rioFdsetIO = {
{ { NULL, 0 } } /* union for io-specific vars */
};
void rioInitWithFdset(rio *r, int *fds, int numfds) {
int j;
*r = rioFdsetIO;
r->io.fdset.fds = zmalloc(sizeof(int)*numfds);
r->io.fdset.state = zmalloc(sizeof(int)*numfds);
memcpy(r->io.fdset.fds,fds,sizeof(int)*numfds);
for (j = 0; j < numfds; j++) r->io.fdset.state[j] = 0;
r->io.fdset.numfds = numfds;
r->io.fdset.pos = 0;
r->io.fdset.buf = sdsempty();
void rioInitWithFd(rio *r, int fd) {
*r = rioFdIO;
r->io.fd.fd = fd;
r->io.fd.pos = 0;
r->io.fd.buf = sdsempty();
}
/* release the rio stream. */
void rioFreeFdset(rio *r) {
zfree(r->io.fdset.fds);
zfree(r->io.fdset.state);
sdsfree(r->io.fdset.buf);
void rioFreeFd(rio *r) {
sdsfree(r->io.fd.buf);
}
/* ---------------------------- Generic functions ---------------------------- */
......
......@@ -35,6 +35,7 @@
#include <stdio.h>
#include <stdint.h>
#include "sds.h"
#include "connection.h"
#define RIO_FLAG_READ_ERROR (1<<0)
#define RIO_FLAG_WRITE_ERROR (1<<1)
......@@ -76,22 +77,20 @@ struct _rio {
off_t buffered; /* Bytes written since last fsync. */
off_t autosync; /* fsync after 'autosync' bytes written. */
} file;
/* file descriptor */
/* Connection object (used to read from socket) */
struct {
int fd; /* File descriptor. */
connection *conn; /* Connection */
off_t pos; /* pos in buf that was returned */
sds buf; /* buffered data */
size_t read_limit; /* don't allow to buffer/read more than that */
size_t read_so_far; /* amount of data read from the rio (not buffered) */
} fd;
/* Multiple FDs target (used to write to N sockets). */
} conn;
/* FD target (used to write to pipe). */
struct {
int *fds; /* File descriptors. */
int *state; /* Error state of each fd. 0 (if ok) or errno. */
int numfds;
int fd; /* File descriptor. */
off_t pos;
sds buf;
} fdset;
} fd;
} io;
};
......@@ -159,11 +158,11 @@ static inline void rioClearErrors(rio *r) {
void rioInitWithFile(rio *r, FILE *fp);
void rioInitWithBuffer(rio *r, sds s);
void rioInitWithFd(rio *r, int fd, size_t read_limit);
void rioInitWithFdset(rio *r, int *fds, int numfds);
void rioInitWithConn(rio *r, connection *conn, size_t read_limit);
void rioInitWithFd(rio *r, int fd);
void rioFreeFdset(rio *r);
void rioFreeFd(rio *r, sds* out_remainingBufferedData);
void rioFreeFd(rio *r);
void rioFreeConn(rio *r, sds* out_remainingBufferedData);
size_t rioWriteBulkCount(rio *r, char prefix, long count);
size_t rioWriteBulkString(rio *r, const char *buf, size_t len);
......
......@@ -61,7 +61,7 @@ sds ldbCatStackValue(sds s, lua_State *lua, int idx);
#define LDB_BREAKPOINTS_MAX 64 /* Max number of breakpoints. */
#define LDB_MAX_LEN_DEFAULT 256 /* Default len limit for replies / var dumps. */
struct ldbState {
int fd; /* Socket of the debugging client. */
connection *conn; /* Connection of the debugging client. */
int active; /* Are we debugging EVAL right now? */
int forked; /* Is this a fork()ed debugging session? */
list *logs; /* List of messages to send to the client. */
......@@ -1243,7 +1243,7 @@ void scriptingInit(int setup) {
* Note: there is no need to create it again when this function is called
* by scriptingReset(). */
if (server.lua_client == NULL) {
server.lua_client = createClient(-1);
server.lua_client = createClient(NULL);
server.lua_client->flags |= CLIENT_LUA;
}
......@@ -1734,7 +1734,7 @@ NULL
/* Initialize Lua debugger data structures. */
void ldbInit(void) {
ldb.fd = -1;
ldb.conn = NULL;
ldb.active = 0;
ldb.logs = listCreate();
listSetFreeMethod(ldb.logs,(void (*)(void*))sdsfree);
......@@ -1756,7 +1756,7 @@ void ldbFlushLog(list *log) {
void ldbEnable(client *c) {
c->flags |= CLIENT_LUA_DEBUG;
ldbFlushLog(ldb.logs);
ldb.fd = c->fd;
ldb.conn = c->conn;
ldb.step = 1;
ldb.bpcount = 0;
ldb.luabp = 0;
......@@ -1811,7 +1811,7 @@ void ldbSendLogs(void) {
proto = sdscatlen(proto,"\r\n",2);
listDelNode(ldb.logs,ln);
}
if (write(ldb.fd,proto,sdslen(proto)) == -1) {
if (connWrite(ldb.conn,proto,sdslen(proto)) == -1) {
/* Avoid warning. We don't check the return value of write()
* since the next read() will catch the I/O error and will
* close the debugging session. */
......@@ -1863,8 +1863,8 @@ int ldbStartSession(client *c) {
}
/* Setup our debugging session. */
anetBlock(NULL,ldb.fd);
anetSendTimeout(NULL,ldb.fd,5000);
connBlock(ldb.conn);
connSendTimeout(ldb.conn,5000);
ldb.active = 1;
/* First argument of EVAL is the script itself. We split it into different
......@@ -1891,7 +1891,7 @@ void ldbEndSession(client *c) {
/* If it's a fork()ed session, we just exit. */
if (ldb.forked) {
writeToClient(c->fd, c, 0);
writeToClient(c,0);
serverLog(LL_WARNING,"Lua debugging session child exiting");
exitFromChild(0);
} else {
......@@ -1900,8 +1900,8 @@ void ldbEndSession(client *c) {
}
/* Otherwise let's restore client's state. */
anetNonBlock(NULL,ldb.fd);
anetSendTimeout(NULL,ldb.fd,0);
connNonBlock(ldb.conn);
connSendTimeout(ldb.conn,0);
/* Close the client connectin after sending the final EVAL reply
* in order to signal the end of the debugging session. */
......@@ -2538,7 +2538,7 @@ int ldbRepl(lua_State *lua) {
while(1) {
while((argv = ldbReplParseCommand(&argc)) == NULL) {
char buf[1024];
int nread = read(ldb.fd,buf,sizeof(buf));
int nread = connRead(ldb.conn,buf,sizeof(buf));
if (nread <= 0) {
/* Make sure the script runs without user input since the
* client is no longer connected. */
......
......@@ -30,6 +30,10 @@
#include "server.h"
#include "hiredis.h"
#ifdef USE_OPENSSL
#include "openssl/ssl.h"
#include "hiredis_ssl.h"
#endif
#include "async.h"
#include <ctype.h>
......@@ -40,6 +44,10 @@
extern char **environ;
#ifdef USE_OPENSSL
extern SSL_CTX *redis_tls_ctx;
#endif
#define REDIS_SENTINEL_PORT 26379
/* ======================== Sentinel global state =========================== */
......@@ -1995,6 +2003,19 @@ void sentinelSetClientName(sentinelRedisInstance *ri, redisAsyncContext *c, char
}
}
static int instanceLinkNegotiateTLS(redisAsyncContext *context) {
#ifndef USE_OPENSSL
(void) context;
#else
if (!redis_tls_ctx) return C_ERR;
SSL *ssl = SSL_new(redis_tls_ctx);
if (!ssl) return C_ERR;
if (redisInitiateSSL(&context->c, ssl) == REDIS_ERR) return C_ERR;
#endif
return C_OK;
}
/* Create the async connections for the instance link if the link
* is disconnected. Note that link->disconnected is true even if just
* one of the two links (commands and pub/sub) is missing. */
......@@ -2010,7 +2031,11 @@ void sentinelReconnectInstance(sentinelRedisInstance *ri) {
/* Commands connection. */
if (link->cc == NULL) {
link->cc = redisAsyncConnectBind(ri->addr->ip,ri->addr->port,NET_FIRST_BIND_ADDR);
if (link->cc->err) {
if (!link->cc->err && server.tls_replication &&
(instanceLinkNegotiateTLS(link->cc) == C_ERR)) {
sentinelEvent(LL_DEBUG,"-cmd-link-reconnection",ri,"%@ #Failed to initialize TLS");
instanceLinkCloseConnection(link,link->cc);
} else if (link->cc->err) {
sentinelEvent(LL_DEBUG,"-cmd-link-reconnection",ri,"%@ #%s",
link->cc->errstr);
instanceLinkCloseConnection(link,link->cc);
......@@ -2033,7 +2058,10 @@ void sentinelReconnectInstance(sentinelRedisInstance *ri) {
/* Pub / Sub */
if ((ri->flags & (SRI_MASTER|SRI_SLAVE)) && link->pc == NULL) {
link->pc = redisAsyncConnectBind(ri->addr->ip,ri->addr->port,NET_FIRST_BIND_ADDR);
if (link->pc->err) {
if (!link->pc->err && server.tls_replication &&
(instanceLinkNegotiateTLS(link->pc) == C_ERR)) {
sentinelEvent(LL_DEBUG,"-pubsub-link-reconnection",ri,"%@ #Failed to initialize TLS");
} else if (link->pc->err) {
sentinelEvent(LL_DEBUG,"-pubsub-link-reconnection",ri,"%@ #%s",
link->pc->errstr);
instanceLinkCloseConnection(link,link->pc);
......@@ -2584,8 +2612,9 @@ int sentinelSendHello(sentinelRedisInstance *ri) {
return C_ERR;
announce_ip = ip;
}
announce_port = sentinel.announce_port ?
sentinel.announce_port : server.port;
if (sentinel.announce_port) announce_port = sentinel.announce_port;
else if (server.tls_replication && server.tls_port) announce_port = server.tls_port;
else announce_port = server.port;
/* Format and send the Hello message. */
snprintf(payload,sizeof(payload),
......
......@@ -1752,6 +1752,62 @@ void updateCachedTime(void) {
server.daylight_active = tm.tm_isdst;
}
void checkChildrenDone(void) {
int statloc;
pid_t pid;
/* If we have a diskless rdb child (note that we support only one concurrent
* child), we want to avoid collecting it's exit status and acting on it
* as long as we didn't finish to drain the pipe, since then we're at risk
* of starting a new fork and a new pipe before we're done with the previous
* one. */
if (server.rdb_child_pid != -1 && server.rdb_pipe_conns)
return;
if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
int exitcode = WEXITSTATUS(statloc);
int bysignal = 0;
if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc);
/* sigKillChildHandler catches the signal and calls exit(), but we
* must make sure not to flag lastbgsave_status, etc incorrectly.
* We could directly terminate the child process via SIGUSR1
* without handling it, but in this case Valgrind will log an
* annoying error. */
if (exitcode == SERVER_CHILD_NOERROR_RETVAL) {
bysignal = SIGUSR1;
exitcode = 1;
}
if (pid == -1) {
serverLog(LL_WARNING,"wait3() returned an error: %s. "
"rdb_child_pid = %d, aof_child_pid = %d, module_child_pid = %d",
strerror(errno),
(int) server.rdb_child_pid,
(int) server.aof_child_pid,
(int) server.module_child_pid);
} else if (pid == server.rdb_child_pid) {
backgroundSaveDoneHandler(exitcode,bysignal);
if (!bysignal && exitcode == 0) receiveChildInfo();
} else if (pid == server.aof_child_pid) {
backgroundRewriteDoneHandler(exitcode,bysignal);
if (!bysignal && exitcode == 0) receiveChildInfo();
} else if (pid == server.module_child_pid) {
ModuleForkDoneHandler(exitcode,bysignal);
if (!bysignal && exitcode == 0) receiveChildInfo();
} else {
if (!ldbRemoveChild(pid)) {
serverLog(LL_WARNING,
"Warning, detected child with unmatched pid: %ld",
(long)pid);
}
}
updateDictResizePolicy();
closeChildInfoPipe();
}
}
/* This is our timer interrupt, called server.hz times per second.
* Here is where we do a number of things that need to be done asynchronously.
* For instance:
......@@ -1903,51 +1959,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
/* Check if a background saving or AOF rewrite in progress terminated. */
if (hasActiveChildProcess() || ldbPendingChildren())
{
int statloc;
pid_t pid;
if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
int exitcode = WEXITSTATUS(statloc);
int bysignal = 0;
if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc);
/* sigKillChildHandler catches the signal and calls exit(), but we
* must make sure not to flag lastbgsave_status, etc incorrectly.
* We could directly terminate the child process via SIGUSR1
* without handling it, but in this case Valgrind will log an
* annoying error. */
if (exitcode == SERVER_CHILD_NOERROR_RETVAL) {
bysignal = SIGUSR1;
exitcode = 1;
}
if (pid == -1) {
serverLog(LL_WARNING,"wait3() returned an error: %s. "
"rdb_child_pid = %d, aof_child_pid = %d, module_child_pid = %d",
strerror(errno),
(int) server.rdb_child_pid,
(int) server.aof_child_pid,
(int) server.module_child_pid);
} else if (pid == server.rdb_child_pid) {
backgroundSaveDoneHandler(exitcode,bysignal);
if (!bysignal && exitcode == 0) receiveChildInfo();
} else if (pid == server.aof_child_pid) {
backgroundRewriteDoneHandler(exitcode,bysignal);
if (!bysignal && exitcode == 0) receiveChildInfo();
} else if (pid == server.module_child_pid) {
ModuleForkDoneHandler(exitcode,bysignal);
if (!bysignal && exitcode == 0) receiveChildInfo();
} else {
if (!ldbRemoveChild(pid)) {
serverLog(LL_WARNING,
"Warning, detected child with unmatched pid: %ld",
(long)pid);
}
}
updateDictResizePolicy();
closeChildInfoPipe();
}
checkChildrenDone();
} else {
/* If there is not a background saving/rewrite in progress check if
* we have to save/rewrite now. */
......@@ -2054,6 +2066,11 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
void beforeSleep(struct aeEventLoop *eventLoop) {
UNUSED(eventLoop);
/* Handle TLS pending data. (must be done before flushAppendOnlyFile) */
tlsProcessPendingData();
/* If tls still has pending unread data don't sleep at all. */
aeSetDontWait(server.el, tlsHasPendingData());
/* Call the Redis Cluster before sleep function. Note that this function
* may change the state of Redis Cluster (from ok to fail or vice versa),
* so it's a good idea to call it before serving the unblocked clients
......@@ -2247,11 +2264,13 @@ void initServerConfig(void) {
server.dynamic_hz = CONFIG_DEFAULT_DYNAMIC_HZ;
server.arch_bits = (sizeof(long) == 8) ? 64 : 32;
server.port = CONFIG_DEFAULT_SERVER_PORT;
server.tls_port = CONFIG_DEFAULT_SERVER_TLS_PORT;
server.tcp_backlog = CONFIG_DEFAULT_TCP_BACKLOG;
server.bindaddr_count = 0;
server.unixsocket = NULL;
server.unixsocketperm = CONFIG_DEFAULT_UNIX_SOCKET_PERM;
server.ipfd_count = 0;
server.tlsfd_count = 0;
server.sofd = -1;
server.protected_mode = CONFIG_DEFAULT_PROTECTED_MODE;
server.gopher_enabled = CONFIG_DEFAULT_GOPHER_ENABLED;
......@@ -2286,6 +2305,7 @@ void initServerConfig(void) {
server.aof_rewrite_min_size = AOF_REWRITE_MIN_SIZE;
server.aof_rewrite_base_size = 0;
server.aof_rewrite_scheduled = 0;
server.aof_flush_sleep = 0;
server.aof_last_fsync = time(NULL);
server.aof_rewrite_time_last = -1;
server.aof_rewrite_time_start = -1;
......@@ -2297,6 +2317,7 @@ void initServerConfig(void) {
server.aof_rewrite_incremental_fsync = CONFIG_DEFAULT_AOF_REWRITE_INCREMENTAL_FSYNC;
server.rdb_save_incremental_fsync = CONFIG_DEFAULT_RDB_SAVE_INCREMENTAL_FSYNC;
server.rdb_key_save_delay = CONFIG_DEFAULT_RDB_KEY_SAVE_DELAY;
server.key_load_delay = CONFIG_DEFAULT_KEY_LOAD_DELAY;
server.aof_load_truncated = CONFIG_DEFAULT_AOF_LOAD_TRUNCATED;
server.aof_use_rdb_preamble = CONFIG_DEFAULT_AOF_USE_RDB_PREAMBLE;
server.pidfile = NULL;
......@@ -2368,7 +2389,7 @@ void initServerConfig(void) {
server.repl_state = REPL_STATE_NONE;
server.repl_transfer_tmpfile = NULL;
server.repl_transfer_fd = -1;
server.repl_transfer_s = -1;
server.repl_transfer_s = NULL;
server.repl_syncio_timeout = CONFIG_REPL_SYNCIO_TIMEOUT;
server.repl_serve_stale_data = CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA;
server.repl_slave_ro = CONFIG_DEFAULT_SLAVE_READ_ONLY;
......@@ -2765,6 +2786,11 @@ void initServer(void) {
server.clients_paused = 0;
server.system_memory_size = zmalloc_get_memory_size();
if (server.tls_port && tlsConfigure(&server.tls_ctx_config) == C_ERR) {
serverLog(LL_WARNING, "Failed to configure TLS. Check logs for more info.");
exit(1);
}
createSharedObjects();
adjustOpenFilesLimit();
server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
......@@ -2780,6 +2806,9 @@ void initServer(void) {
if (server.port != 0 &&
listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
exit(1);
if (server.tls_port != 0 &&
listenToPort(server.tls_port,server.tlsfd,&server.tlsfd_count) == C_ERR)
exit(1);
/* Open the listening Unix domain socket. */
if (server.unixsocket != NULL) {
......@@ -2794,7 +2823,7 @@ void initServer(void) {
}
/* Abort if there are no listening sockets at all. */
if (server.ipfd_count == 0 && server.sofd < 0) {
if (server.ipfd_count == 0 && server.tlsfd_count == 0 && server.sofd < 0) {
serverLog(LL_WARNING, "Configured to not listen anywhere, exiting.");
exit(1);
}
......@@ -2820,6 +2849,11 @@ void initServer(void) {
server.aof_child_pid = -1;
server.module_child_pid = -1;
server.rdb_child_type = RDB_CHILD_TYPE_NONE;
server.rdb_pipe_conns = NULL;
server.rdb_pipe_numconns = 0;
server.rdb_pipe_numconns_writing = 0;
server.rdb_pipe_buff = NULL;
server.rdb_pipe_bufflen = 0;
server.rdb_bgsave_scheduled = 0;
server.child_info_pipe[0] = -1;
server.child_info_pipe[1] = -1;
......@@ -2866,6 +2900,14 @@ void initServer(void) {
"Unrecoverable error creating server.ipfd file event.");
}
}
for (j = 0; j < server.tlsfd_count; j++) {
if (aeCreateFileEvent(server.el, server.tlsfd[j], AE_READABLE,
acceptTLSHandler,NULL) == AE_ERR)
{
serverPanic(
"Unrecoverable error creating server.tlsfd file event.");
}
}
if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event.");
......@@ -3570,6 +3612,7 @@ void closeListeningSockets(int unlink_unix_socket) {
int j;
for (j = 0; j < server.ipfd_count; j++) close(server.ipfd[j]);
for (j = 0; j < server.tlsfd_count; j++) close(server.tlsfd[j]);
if (server.sofd != -1) close(server.sofd);
if (server.cluster_enabled)
for (j = 0; j < server.cfd_count; j++) close(server.cfd[j]);
......@@ -3940,7 +3983,7 @@ sds genRedisInfoString(char *section) {
#endif
(int64_t) getpid(),
server.runid,
server.port,
server.port ? server.port : server.tls_port,
(int64_t)uptime,
(int64_t)(uptime/(3600*24)),
server.hz,
......@@ -4324,7 +4367,7 @@ sds genRedisInfoString(char *section) {
long lag = 0;
if (slaveip[0] == '\0') {
if (anetPeerToString(slave->fd,ip,sizeof(ip),&port) == -1)
if (connPeerToString(slave->conn,ip,sizeof(ip),&port) == -1)
continue;
slaveip = ip;
}
......@@ -4578,7 +4621,7 @@ void redisAsciiArt(void) {
if (!show_logo) {
serverLog(LL_NOTICE,
"Running mode=%s, port=%d.",
mode, server.port
mode, server.port ? server.port : server.tls_port
);
} else {
snprintf(buf,1024*16,ascii_logo,
......@@ -4586,7 +4629,7 @@ void redisAsciiArt(void) {
redisGitSHA1(),
strtol(redisGitDirty(),NULL,10) > 0,
(sizeof(long) == 8) ? "64" : "32",
mode, server.port,
mode, server.port ? server.port : server.tls_port,
(long) getpid()
);
serverLogRaw(LL_NOTICE|LL_RAW,buf);
......@@ -4769,7 +4812,7 @@ void redisSetProcTitle(char *title) {
setproctitle("%s %s:%d%s",
title,
server.bindaddr_count ? server.bindaddr[0] : "*",
server.port,
server.port ? server.port : server.tls_port,
server_mode);
#else
UNUSED(title);
......@@ -4920,6 +4963,7 @@ int main(int argc, char **argv) {
ACLInit(); /* The ACL subsystem must be initialized ASAP because the
basic networking code and client creation depends on it. */
moduleInitModulesSystem();
tlsInit();
/* Store the executable path and arguments in a safe place in order
* to be able to restart the server later. */
......@@ -5053,7 +5097,7 @@ int main(int argc, char **argv) {
exit(1);
}
}
if (server.ipfd_count > 0)
if (server.ipfd_count > 0 || server.tlsfd_count > 0)
serverLog(LL_NOTICE,"Ready to accept connections");
if (server.sofd > 0)
serverLog(LL_NOTICE,"The server is now ready to accept connections at %s", server.unixsocket);
......
......@@ -66,6 +66,7 @@ typedef long long mstime_t; /* millisecond time type. */
#include "quicklist.h" /* Lists are encoded as linked lists of
N-elements flat arrays */
#include "rax.h" /* Radix tree */
#include "connection.h" /* Connection abstraction */
/* Following includes allow test functions to be called from Redis main() */
#include "zipmap.h"
......@@ -84,6 +85,7 @@ typedef long long mstime_t; /* millisecond time type. */
#define CONFIG_MAX_HZ 500
#define MAX_CLIENTS_PER_CLOCK_TICK 200 /* HZ is adapted based on that. */
#define CONFIG_DEFAULT_SERVER_PORT 6379 /* TCP port. */
#define CONFIG_DEFAULT_SERVER_TLS_PORT 0 /* TCP port. */
#define CONFIG_DEFAULT_TCP_BACKLOG 511 /* TCP listen backlog. */
#define CONFIG_DEFAULT_CLIENT_TIMEOUT 0 /* Default client timeout: infinite */
#define CONFIG_DEFAULT_DBNUM 16
......@@ -133,6 +135,7 @@ typedef long long mstime_t; /* millisecond time type. */
#define CONFIG_DEFAULT_REPL_DISKLESS_SYNC 0
#define CONFIG_DEFAULT_REPL_DISKLESS_SYNC_DELAY 5
#define CONFIG_DEFAULT_RDB_KEY_SAVE_DELAY 0
#define CONFIG_DEFAULT_KEY_LOAD_DELAY 0
#define CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA 1
#define CONFIG_DEFAULT_SLAVE_READ_ONLY 1
#define CONFIG_DEFAULT_SLAVE_IGNORE_MAXMEMORY 1
......@@ -826,7 +829,7 @@ typedef struct user {
* Clients are taken in a linked list. */
typedef struct client {
uint64_t id; /* Client incremental unique ID. */
int fd; /* Client socket. */
connection *conn;
int resp; /* RESP protocol version. Can be 2 or 3. */
redisDb *db; /* Pointer to currently SELECTed DB. */
robj *name; /* As set by CLIENT SETNAME. */
......@@ -1034,6 +1037,22 @@ struct malloc_stats {
size_t allocator_resident;
};
/*-----------------------------------------------------------------------------
* TLS Context Configuration
*----------------------------------------------------------------------------*/
typedef struct redisTLSContextConfig {
char *cert_file;
char *key_file;
char *dh_params_file;
char *ca_cert_file;
char *ca_cert_dir;
char *protocols;
char *ciphers;
char *ciphersuites;
int prefer_server_ciphers;
} redisTLSContextConfig;
/*-----------------------------------------------------------------------------
* Global server state
*----------------------------------------------------------------------------*/
......@@ -1088,6 +1107,7 @@ struct redisServer {
pid_t module_child_pid; /* PID of module child */
/* Networking */
int port; /* TCP listening port */
int tls_port; /* TLS listening port */
int tcp_backlog; /* TCP listen() backlog */
char *bindaddr[CONFIG_BINDADDR_MAX]; /* Addresses we should bind to */
int bindaddr_count; /* Number of addresses in server.bindaddr[] */
......@@ -1095,6 +1115,8 @@ struct redisServer {
mode_t unixsocketperm; /* UNIX socket permission */
int ipfd[CONFIG_BINDADDR_MAX]; /* TCP socket file descriptors */
int ipfd_count; /* Used slots in ipfd[] */
int tlsfd[CONFIG_BINDADDR_MAX]; /* TLS socket file descriptors */
int tlsfd_count; /* Used slots in tlsfd[] */
int sofd; /* Unix socket file descriptor */
int cfd[CONFIG_BINDADDR_MAX];/* Cluster bus listening socket */
int cfd_count; /* Used slots in cfd[] */
......@@ -1198,6 +1220,7 @@ struct redisServer {
off_t aof_rewrite_base_size; /* AOF size on latest startup or rewrite. */
off_t aof_current_size; /* AOF current size. */
off_t aof_fsync_offset; /* AOF offset which is already synced to disk. */
int aof_flush_sleep; /* Micros to sleep before flush. (used by tests) */
int aof_rewrite_scheduled; /* Rewrite once BGSAVE terminates. */
pid_t aof_child_pid; /* PID if rewriting process */
list *aof_rewrite_buf_blocks; /* Hold changes during an AOF rewrite. */
......@@ -1243,10 +1266,17 @@ struct redisServer {
int rdb_child_type; /* Type of save by active child. */
int lastbgsave_status; /* C_OK or C_ERR */
int stop_writes_on_bgsave_err; /* Don't allow writes if can't BGSAVE */
int rdb_pipe_write_result_to_parent; /* RDB pipes used to return the state */
int rdb_pipe_read_result_from_child; /* of each slave in diskless SYNC. */
int rdb_pipe_write; /* RDB pipes used to transfer the rdb */
int rdb_pipe_read; /* data to the parent process in diskless repl. */
connection **rdb_pipe_conns; /* Connections which are currently the */
int rdb_pipe_numconns; /* target of diskless rdb fork child. */
int rdb_pipe_numconns_writing; /* Number of rdb conns with pending writes. */
char *rdb_pipe_buff; /* In diskless replication, this buffer holds data */
int rdb_pipe_bufflen; /* that was read from the the rdb pipe. */
int rdb_key_save_delay; /* Delay in microseconds between keys while
* writing the RDB. (for testings) */
int key_load_delay; /* Delay in microseconds between keys while
* loading aof or rdb. (for testings) */
/* Pipe and data structures for child -> parent info sharing. */
int child_info_pipe[2]; /* Pipe used to write the child_info_data. */
struct {
......@@ -1299,7 +1329,7 @@ struct redisServer {
off_t repl_transfer_size; /* Size of RDB to read from master during sync. */
off_t repl_transfer_read; /* Amount of RDB read from master during sync. */
off_t repl_transfer_last_fsync_off; /* Offset when we fsync-ed last time. */
int repl_transfer_s; /* Slave -> Master SYNC socket */
connection *repl_transfer_s; /* Slave -> Master SYNC connection */
int repl_transfer_fd; /* Slave -> Master SYNC temp file descriptor */
char *repl_transfer_tmpfile; /* Slave-> master SYNC temp file name */
time_t repl_transfer_lastio; /* Unix time of the latest read, for timeout */
......@@ -1423,6 +1453,11 @@ struct redisServer {
int watchdog_period; /* Software watchdog period in ms. 0 = off */
/* System hardware info */
size_t system_memory_size; /* Total memory in system as reported by OS */
/* TLS Configuration */
int tls_cluster;
int tls_replication;
int tls_auth_clients;
redisTLSContextConfig tls_ctx_config;
};
typedef struct pubsubPattern {
......@@ -1570,12 +1605,12 @@ size_t redisPopcount(void *s, long count);
void redisSetProcTitle(char *title);
/* networking.c -- Networking and Client related operations */
client *createClient(int fd);
client *createClient(connection *conn);
void closeTimedoutClients(void);
void freeClient(client *c);
void freeClientAsync(client *c);
void resetClient(client *c);
void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask);
void sendReplyToClient(connection *conn);
void *addReplyDeferredLen(client *c);
void setDeferredArrayLen(client *c, void *node, long length);
void setDeferredMapLen(client *c, void *node, long length);
......@@ -1587,8 +1622,9 @@ void processInputBufferAndReplicate(client *c);
void processGopherRequest(client *c);
void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void acceptTLSHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask);
void readQueryFromClient(connection *conn);
void addReplyNull(client *c);
void addReplyNullArray(client *c);
void addReplyBool(client *c, int b);
......@@ -1646,7 +1682,7 @@ int handleClientsWithPendingReadsUsingThreads(void);
int stopThreadedIOIfNeeded(void);
int clientHasPendingReplies(client *c);
void unlinkClient(client *c);
int writeToClient(int fd, client *c, int handler_installed);
int writeToClient(client *c, int handler_installed);
void linkClient(client *c);
void protectClient(client *c);
void unprotectClient(client *c);
......@@ -1782,6 +1818,8 @@ void clearReplicationId2(void);
void chopReplicationBacklog(void);
void replicationCacheMasterUsingMyself(void);
void feedReplicationBacklog(void *ptr, size_t len);
void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
void rdbPipeWriteHandlerConnRemoved(struct connection *conn);
/* Generic persistence functions */
void startLoadingFile(FILE* fp, char* filename);
......@@ -1954,6 +1992,7 @@ unsigned int LRU_CLOCK(void);
const char *evictPolicyToString(void);
struct redisMemOverhead *getMemoryOverheadData(void);
void freeMemoryOverheadData(struct redisMemOverhead *mh);
void checkChildrenDone(void);
#define RESTART_SERVER_NONE 0
#define RESTART_SERVER_GRACEFULLY (1<<0) /* Do proper shutdown. */
......@@ -2369,6 +2408,10 @@ void mixDigest(unsigned char *digest, void *ptr, size_t len);
void xorDigest(unsigned char *digest, void *ptr, size_t len);
int populateCommandTableParseFlags(struct redisCommand *c, char *strflags);
/* TLS stuff */
void tlsInit(void);
int tlsConfigure(redisTLSContextConfig *ctx_config);
#define redisDebug(fmt, ...) \
printf("DEBUG %s:%d > " fmt "\n", __FILE__, __LINE__, __VA_ARGS__)
#define redisDebugMark() \
......
This diff is collapsed.
......@@ -8,6 +8,7 @@ source ../instances.tcl
source ../../support/cluster.tcl ; # Redis Cluster client.
set ::instances_count 20 ; # How many instances we use at max.
set ::tlsdir "../../tls"
proc main {} {
parse_options
......
......@@ -4,6 +4,7 @@
# are preseved across iterations.
source "../tests/includes/init-tests.tcl"
source "../../../tests/support/cli.tcl"
test "Create a 5 nodes cluster" {
create_cluster 5 5
......@@ -79,6 +80,7 @@ test "Cluster consistency during live resharding" {
--cluster-to $target \
--cluster-slots 100 \
--cluster-yes \
{*}[rediscli_tls_config "../../../tests"] \
| [info nameofexecutable] \
../tests/helpers/onlydots.tcl \
&] 0]
......
......@@ -5,6 +5,7 @@
# other masters have slaves.
source "../tests/includes/init-tests.tcl"
source "../../../tests/support/cli.tcl"
# Create a cluster with 5 master and 15 slaves, to make sure there are no
# empty masters and make rebalancing simpler to handle during the test.
......@@ -33,7 +34,9 @@ test "Resharding all the master #0 slots away from it" {
set output [exec \
../../../src/redis-cli --cluster rebalance \
127.0.0.1:[get_instance_attrib redis 0 port] \
{*}[rediscli_tls_config "../../../tests"] \
--cluster-weight ${master0_id}=0 >@ stdout ]
}
test "Master #0 should lose its replicas" {
......@@ -51,6 +54,7 @@ test "Resharding back some slot to master #0" {
set output [exec \
../../../src/redis-cli --cluster rebalance \
127.0.0.1:[get_instance_attrib redis 0 port] \
{*}[rediscli_tls_config "../../../tests"] \
--cluster-weight ${master0_id}=.01 \
--cluster-use-empty-masters >@ stdout]
}
......
source tests/support/redis.tcl
source tests/support/util.tcl
set ::tlsdir "tests/tls"
# This function sometimes writes sometimes blocking-reads from lists/sorted
# sets. There are multiple processes like this executing at the same time
# so that we have some chance to trap some corner condition if there is
......@@ -8,8 +10,8 @@ source tests/support/util.tcl
# space to just a few elements, and balance the operations so that it is
# unlikely that lists and zsets just get more data without ever causing
# blocking.
proc bg_block_op {host port db ops} {
set r [redis $host $port]
proc bg_block_op {host port db ops tls} {
set r [redis $host $port 0 $tls]
$r select $db
for {set j 0} {$j < $ops} {incr j} {
......@@ -49,4 +51,4 @@ proc bg_block_op {host port db ops} {
}
}
bg_block_op [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] [lindex $argv 3]
bg_block_op [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] [lindex $argv 3] [lindex $argv 4]
source tests/support/redis.tcl
source tests/support/util.tcl
proc bg_complex_data {host port db ops} {
set r [redis $host $port]
set ::tlsdir "tests/tls"
proc bg_complex_data {host port db ops tls} {
set r [redis $host $port 0 $tls]
$r select $db
createComplexDataset $r $ops
}
bg_complex_data [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] [lindex $argv 3]
bg_complex_data [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] [lindex $argv 3] [lindex $argv 4]
source tests/support/redis.tcl
proc gen_write_load {host port seconds} {
set ::tlsdir "tests/tls"
proc gen_write_load {host port seconds tls} {
set start_time [clock seconds]
set r [redis $host $port 1]
set r [redis $host $port 0 $tls]
$r select 9
while 1 {
$r set [expr rand()] [expr rand()]
......@@ -12,4 +14,4 @@ proc gen_write_load {host port seconds} {
}
}
gen_write_load [lindex $argv 0] [lindex $argv 1] [lindex $argv 2]
gen_write_load [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] [lindex $argv 3]
......@@ -17,6 +17,7 @@ source ../support/test.tcl
set ::verbose 0
set ::valgrind 0
set ::tls 0
set ::pause_on_error 0
set ::simulate_error 0
set ::failed 0
......@@ -69,7 +70,19 @@ proc spawn_instance {type base_port count {conf {}}} {
# Write the instance config file.
set cfgfile [file join $dirname $type.conf]
set cfg [open $cfgfile w]
if {$::tls} {
puts $cfg "tls-port $port"
puts $cfg "tls-replication yes"
puts $cfg "tls-cluster yes"
puts $cfg "port 0"
puts $cfg [format "tls-cert-file %s/../../tls/redis.crt" [pwd]]
puts $cfg [format "tls-key-file %s/../../tls/redis.key" [pwd]]
puts $cfg [format "tls-dh-params-file %s/../../tls/redis.dh" [pwd]]
puts $cfg [format "tls-ca-cert-file %s/../../tls/ca.crt" [pwd]]
puts $cfg "loglevel debug"
} else {
puts $cfg "port $port"
}
puts $cfg "dir ./$dirname"
puts $cfg "logfile log.txt"
# Add additional config files
......@@ -88,7 +101,7 @@ proc spawn_instance {type base_port count {conf {}}} {
}
# Push the instance into the right list
set link [redis 127.0.0.1 $port]
set link [redis 127.0.0.1 $port 0 $::tls]
$link reconnect 1
lappend ::${type}_instances [list \
pid $pid \
......@@ -148,6 +161,13 @@ proc parse_options {} {
set ::simulate_error 1
} elseif {$opt eq {--valgrind}} {
set ::valgrind 1
} elseif {$opt eq {--tls}} {
package require tls 1.6
::tls::init \
-cafile "$::tlsdir/ca.crt" \
-certfile "$::tlsdir/redis.crt" \
-keyfile "$::tlsdir/redis.key"
set ::tls 1
} elseif {$opt eq "--help"} {
puts "Hello, I'm sentinel.tcl and I run Sentinel unit tests."
puts "\nOptions:"
......@@ -492,7 +512,7 @@ proc restart_instance {type id} {
}
# Connect with it with a fresh link
set link [redis 127.0.0.1 $port]
set link [redis 127.0.0.1 $port 0 $::tls]
$link reconnect 1
set_instance_attrib $type $id link $link
......
......@@ -13,8 +13,9 @@ tags {"aof"} {
# cleaned after a child responsible for an AOF rewrite exited. This buffer
# was subsequently appended to the new AOF, resulting in duplicate commands.
start_server_aof [list dir $server_path] {
set client [redis [srv host] [srv port]]
set bench [open "|src/redis-benchmark -q -p [srv port] -c 20 -n 20000 incr foo" "r+"]
set client [redis [srv host] [srv port] 0 $::tls]
set bench [open "|src/redis-benchmark -q -s [srv unixsocket] -c 20 -n 20000 incr foo" "r+"]
after 100
# Benchmark should be running by now: start background rewrite
......@@ -29,7 +30,7 @@ tags {"aof"} {
# Restart server to replay AOF
start_server_aof [list dir $server_path] {
set client [redis [srv host] [srv port]]
set client [redis [srv host] [srv port] 0 $::tls]
assert_equal 20000 [$client get foo]
}
}
......@@ -52,7 +52,7 @@ tags {"aof"} {
assert_equal 1 [is_alive $srv]
}
set client [redis [dict get $srv host] [dict get $srv port]]
set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
test "Truncated AOF loaded: we expect foo to be equal to 5" {
assert {[$client get foo] eq "5"}
......@@ -69,7 +69,7 @@ tags {"aof"} {
assert_equal 1 [is_alive $srv]
}
set client [redis [dict get $srv host] [dict get $srv port]]
set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
test "Truncated AOF loaded: we expect foo to be equal to 6 now" {
assert {[$client get foo] eq "6"}
......@@ -170,7 +170,7 @@ tags {"aof"} {
}
test "Fixed AOF: Keyspace should contain values that were parseable" {
set client [redis [dict get $srv host] [dict get $srv port]]
set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
wait_for_condition 50 100 {
[catch {$client ping} e] == 0
} else {
......@@ -194,7 +194,7 @@ tags {"aof"} {
}
test "AOF+SPOP: Set should have 1 member" {
set client [redis [dict get $srv host] [dict get $srv port]]
set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
wait_for_condition 50 100 {
[catch {$client ping} e] == 0
} else {
......@@ -218,7 +218,7 @@ tags {"aof"} {
}
test "AOF+SPOP: Set should have 1 member" {
set client [redis [dict get $srv host] [dict get $srv port]]
set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
wait_for_condition 50 100 {
[catch {$client ping} e] == 0
} else {
......@@ -241,7 +241,7 @@ tags {"aof"} {
}
test "AOF+EXPIRE: List should be empty" {
set client [redis [dict get $srv host] [dict get $srv port]]
set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
wait_for_condition 50 100 {
[catch {$client ping} e] == 0
} else {
......@@ -257,4 +257,35 @@ tags {"aof"} {
r expire x -1
}
}
start_server {overrides {appendonly {yes} appendfilename {appendonly.aof} appendfsync always}} {
test {AOF fsync always barrier issue} {
set rd [redis_deferring_client]
# Set a sleep when aof is flushed, so that we have a chance to look
# at the aof size and detect if the response of an incr command
# arrives before the data was written (and hopefully fsynced)
# We create a big reply, which will hopefully not have room in the
# socket buffers, and will install a write handler, then we sleep
# a big and issue the incr command, hoping that the last portion of
# the output buffer write, and the processing of the incr will happen
# in the same event loop cycle.
# Since the socket buffers and timing are unpredictable, we fuzz this
# test with slightly different sizes and sleeps a few times.
for {set i 0} {$i < 10} {incr i} {
r debug aof-flush-sleep 0
r del x
r setrange x [expr {int(rand()*5000000)+10000000}] x
r debug aof-flush-sleep 500000
set aof [file join [lindex [r config get dir] 1] appendonly.aof]
set size1 [file size $aof]
$rd get x
after [expr {int(rand()*30)}]
$rd incr new_value
$rd read
$rd read
set size2 [file size $aof]
assert {$size1 != $size2}
}
}
}
}
......@@ -2,9 +2,9 @@
# Unlike stream operations such operations are "pop" style, so they consume
# the list or sorted set, and must be replicated correctly.
proc start_bg_block_op {host port db ops} {
proc start_bg_block_op {host port db ops tls} {
set tclsh [info nameofexecutable]
exec $tclsh tests/helpers/bg_block_op.tcl $host $port $db $ops &
exec $tclsh tests/helpers/bg_block_op.tcl $host $port $db $ops $tls &
}
proc stop_bg_block_op {handle} {
......@@ -18,9 +18,9 @@ start_server {tags {"repl"}} {
set master_port [srv -1 port]
set slave [srv 0 client]
set load_handle0 [start_bg_block_op $master_host $master_port 9 100000]
set load_handle1 [start_bg_block_op $master_host $master_port 9 100000]
set load_handle2 [start_bg_block_op $master_host $master_port 9 100000]
set load_handle0 [start_bg_block_op $master_host $master_port 9 100000 $::tls]
set load_handle1 [start_bg_block_op $master_host $master_port 9 100000 $::tls]
set load_handle2 [start_bg_block_op $master_host $master_port 9 100000 $::tls]
test {First server should have role slave after SLAVEOF} {
$slave slaveof $master_host $master_port
......
......@@ -18,6 +18,7 @@ start_server {} {
set R($j) [srv [expr 0-$j] client]
set R_host($j) [srv [expr 0-$j] host]
set R_port($j) [srv [expr 0-$j] port]
set R_unixsocket($j) [srv [expr 0-$j] unixsocket]
if {$debug_msg} {puts "Log file: [srv [expr 0-$j] stdout]"}
}
......@@ -36,7 +37,7 @@ start_server {} {
}
set cycle_start_time [clock milliseconds]
set bench_pid [exec src/redis-benchmark -p $R_port(0) -n 10000000 -r 1000 incr __rand_int__ > /dev/null &]
set bench_pid [exec src/redis-benchmark -s $R_unixsocket(0) -n 10000000 -r 1000 incr __rand_int__ > /dev/null &]
while 1 {
set elapsed [expr {[clock milliseconds]-$cycle_start_time}]
if {$elapsed > $duration*1000} break
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment