Merge branch 'keydbpro' into PRO_RELEASE_6
Former-commit-id: dd8339da10c17f0e2f238310a10e97395f8f12ac
This commit is contained in:
John Sully
commit
b2a313e0c1
@ -4,6 +4,8 @@
|
||||
|
||||
##### Need Help? Check out our extensive [documentation](https://docs.keydb.dev).
|
||||
|
||||
##### Have feedback? Take our quick survey: https://www.surveymonkey.com/r/Y9XNS93
|
||||
|
||||
What is KeyDB?
|
||||
--------------
|
||||
|
||||
|
||||
13
src/Makefile
13
src/Makefile
@ -108,13 +108,18 @@ ifeq ($(USE_JEMALLOC),no)
|
||||
MALLOC=libc
|
||||
endif
|
||||
|
||||
|
||||
ifeq ($(NO_LICENSE_CHECK),yes)
|
||||
CXXFLAGS+=-DNO_LICENSE_CHECK=1
|
||||
endif
|
||||
|
||||
# Override default settings if possible
|
||||
-include .make-settings
|
||||
|
||||
FINAL_CFLAGS=$(STD) $(WARN) $(OPT) $(DEBUG) $(CFLAGS) $(REDIS_CFLAGS)
|
||||
FINAL_CXXFLAGS=$(CXX_STD) $(WARN) $(OPT) $(DEBUG) $(CXXFLAGS) $(REDIS_CFLAGS)
|
||||
FINAL_LDFLAGS=$(LDFLAGS) $(REDIS_LDFLAGS) $(DEBUG)
|
||||
FINAL_LIBS=-lm -lz -latomic -L$(LICENSE_LIB_DIR) -lkey -lcrypto
|
||||
FINAL_LIBS=-lm -lcurl -lz -latomic -L$(LICENSE_LIB_DIR) -lkey -lcrypto
|
||||
DEBUG=-g -ggdb
|
||||
|
||||
# Linux ARM needs -latomic at linking time
|
||||
@ -265,9 +270,9 @@ endif
|
||||
|
||||
REDIS_SERVER_NAME=keydb-pro-server
|
||||
REDIS_SENTINEL_NAME=keydb-sentinel
|
||||
REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o acl.o storage.o rdb-s3.o fastlock.o new.o tracking.o cron.o connection.o tls.o sha256.o AsyncWorkQueue.o snapshot.o storage/rocksdb.o storage/rocksdbfactory.o storage/teststorageprovider.o keydbutils.o $(ASM_OBJ)
|
||||
REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o acl.o storage.o rdb-s3.o fastlock.o new.o tracking.o cron.o connection.o tls.o sha256.o AsyncWorkQueue.o snapshot.o storage/rocksdb.o storage/rocksdbfactory.o storage/teststorageprovider.o keydbutils.o motd.o $(ASM_OBJ)
|
||||
REDIS_CLI_NAME=keydb-cli
|
||||
REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o redis-cli-cpphelper.o zmalloc.o release.o anet.o ae.o crc64.o siphash.o crc16.o storage-lite.o fastlock.o new.o $(ASM_OBJ)
|
||||
REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o redis-cli-cpphelper.o zmalloc.o release.o anet.o ae.o crc64.o siphash.o crc16.o storage-lite.o fastlock.o new.o motd.o $(ASM_OBJ)
|
||||
REDIS_BENCHMARK_NAME=keydb-benchmark
|
||||
REDIS_BENCHMARK_OBJ=ae.o anet.o redis-benchmark.o adlist.o dict.o zmalloc.o siphash.o redis-benchmark.o storage-lite.o fastlock.o new.o $(ASM_OBJ)
|
||||
REDIS_CHECK_RDB_NAME=keydb-check-rdb
|
||||
@ -337,7 +342,7 @@ $(REDIS_CHECK_AOF_NAME): $(REDIS_SERVER_NAME)
|
||||
|
||||
# keydb-cli
|
||||
$(REDIS_CLI_NAME): $(REDIS_CLI_OBJ)
|
||||
$(REDIS_LD) -o $@ $^ ../deps/hiredis/libhiredis.a ../deps/linenoise/linenoise.o $(FINAL_LIBS) -lcurl
|
||||
$(REDIS_LD) -o $@ $^ ../deps/hiredis/libhiredis.a ../deps/linenoise/linenoise.o $(FINAL_LIBS)
|
||||
|
||||
# keydb-benchmark
|
||||
$(REDIS_BENCHMARK_NAME): $(REDIS_BENCHMARK_OBJ)
|
||||
|
||||
@ -36,7 +36,6 @@ const char *ascii_logo =
|
||||
" Port: %d\n"
|
||||
" PID: %ld\n"
|
||||
" \n"
|
||||
" Like KeyDB? Star us on GitHub! \n"
|
||||
" %s\n"
|
||||
" \n"
|
||||
" https://github.com/JohnSully/KeyDB \n"
|
||||
" \n\n";
|
||||
" \n";
|
||||
|
||||
@ -306,6 +306,7 @@ bool initializeStorageProvider(const char **err)
|
||||
if (!strcasecmp(g_sdsProvider, "flash") && g_sdsArgs != nullptr)
|
||||
{
|
||||
// Create The Storage Factory (if necessary)
|
||||
serverLog(LL_NOTICE, "Initializing FLASH storage provider (this may take a long time)");
|
||||
g_pserver->m_pstorageFactory = CreateRocksDBStorageFactory(g_sdsArgs, cserver.dbnum);
|
||||
}
|
||||
else if (!strcasecmp(g_sdsProvider, "test") && g_sdsArgs == nullptr)
|
||||
@ -990,8 +991,15 @@ void configGetCommand(client *c) {
|
||||
listIter li;
|
||||
listNode *ln;
|
||||
listRewind(g_pserver->masters, &li);
|
||||
bool fFirst = true;
|
||||
while ((ln = listNext(&li)))
|
||||
{
|
||||
if (!fFirst)
|
||||
{
|
||||
addReplyBulkCString(c,optname);
|
||||
matches++;
|
||||
}
|
||||
fFirst = false;
|
||||
struct redisMaster *mi = (struct redisMaster*)listNodeValue(ln);
|
||||
snprintf(buf,sizeof(buf),"%s %d",
|
||||
mi->masterhost, mi->masterport);
|
||||
@ -2365,7 +2373,7 @@ standardConfig configs[] = {
|
||||
createIntConfig("lfu-decay-time", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->lfu_decay_time, 1, INTEGER_CONFIG, NULL, NULL),
|
||||
createIntConfig("replica-priority", "slave-priority", MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->slave_priority, 100, INTEGER_CONFIG, NULL, NULL),
|
||||
createIntConfig("repl-diskless-sync-delay", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->repl_diskless_sync_delay, 5, INTEGER_CONFIG, NULL, NULL),
|
||||
createIntConfig("maxmemory-samples", NULL, MODIFIABLE_CONFIG, 1, INT_MAX, g_pserver->maxmemory_samples, 5, INTEGER_CONFIG, NULL, NULL),
|
||||
createIntConfig("maxmemory-samples", NULL, MODIFIABLE_CONFIG, 1, INT_MAX, g_pserver->maxmemory_samples, 10, INTEGER_CONFIG, NULL, NULL),
|
||||
createIntConfig("timeout", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, cserver.maxidletime, 0, INTEGER_CONFIG, NULL, NULL), /* Default client timeout: infinite */
|
||||
createIntConfig("replica-announce-port", "slave-announce-port", MODIFIABLE_CONFIG, 0, 65535, g_pserver->slave_announce_port, 0, INTEGER_CONFIG, NULL, NULL),
|
||||
createIntConfig("tcp-backlog", NULL, IMMUTABLE_CONFIG, 0, INT_MAX, g_pserver->tcp_backlog, 511, INTEGER_CONFIG, NULL, NULL), /* TCP listen backlog. */
|
||||
@ -2381,6 +2389,7 @@ standardConfig configs[] = {
|
||||
createIntConfig("hz", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->config_hz, CONFIG_DEFAULT_HZ, INTEGER_CONFIG, NULL, updateHZ),
|
||||
createIntConfig("min-replicas-to-write", "min-slaves-to-write", MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->repl_min_slaves_to_write, 0, INTEGER_CONFIG, NULL, updateGoodSlaves),
|
||||
createIntConfig("min-replicas-max-lag", "min-slaves-max-lag", MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->repl_min_slaves_max_lag, 10, INTEGER_CONFIG, NULL, updateGoodSlaves),
|
||||
createIntConfig("min-clients-per-thread", NULL, MODIFIABLE_CONFIG, 0, 400, cserver.thread_min_client_threshold, 50, INTEGER_CONFIG, NULL, NULL),
|
||||
/* Unsigned int configs */
|
||||
createUIntConfig("maxclients", NULL, MODIFIABLE_CONFIG, 1, UINT_MAX, g_pserver->maxclients, 10000, INTEGER_CONFIG, NULL, updateMaxclients),
|
||||
|
||||
|
||||
@ -666,6 +666,21 @@ int freeMemoryIfNeeded(void) {
|
||||
return C_OK;
|
||||
|
||||
cant_free:
|
||||
if (!cserver.delete_on_evict)
|
||||
{
|
||||
for (int idb = 0; idb < cserver.dbnum; ++idb)
|
||||
{
|
||||
redisDb *db = g_pserver->db[idb];
|
||||
if (db->FStorageProvider())
|
||||
{
|
||||
serverLog(LL_WARNING, "Failed to evict keys, falling back to flushing entire cache. Consider increasing maxmemory-samples.");
|
||||
db->removeAllCachedValues();
|
||||
if (((mem_reported - zmalloc_used_memory()) + mem_freed) >= mem_tofree)
|
||||
return C_OK;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* We are here if we are not able to reclaim memory. There is only one
|
||||
* last thing we can try: check if the lazyfree thread has jobs in queue
|
||||
* and wait... */
|
||||
|
||||
132
src/motd.cpp
Normal file
132
src/motd.cpp
Normal file
@ -0,0 +1,132 @@
|
||||
#include "sds.h"
|
||||
#include <cstring>
|
||||
#include <unistd.h>
|
||||
#include <sys/types.h>
|
||||
#include <pwd.h>
|
||||
#include <sys/stat.h>
|
||||
#include "motd.h"
|
||||
|
||||
/*------------------------------------------------------------------------------
|
||||
* Message of the day
|
||||
*--------------------------------------------------------------------------- */
|
||||
#ifdef MOTD
|
||||
#include <curl/curl.h>
|
||||
|
||||
static const char *szMotdCachePath()
|
||||
{
|
||||
static sds sdsMotdCachePath = NULL;
|
||||
if (sdsMotdCachePath != NULL)
|
||||
return sdsMotdCachePath;
|
||||
|
||||
struct passwd *pw = getpwuid(getuid());
|
||||
if (pw == NULL)
|
||||
return "";
|
||||
const char *homedir = pw->pw_dir;
|
||||
sdsMotdCachePath = sdsnew(homedir);
|
||||
sdsMotdCachePath = sdscat(sdsMotdCachePath, motd_cache_file);
|
||||
return sdsMotdCachePath;
|
||||
}
|
||||
static size_t motd_write_callback(void *ptr, size_t size, size_t nmemb, sds *str)
|
||||
{
|
||||
*str = sdscatlen(*str, ptr, size*nmemb);
|
||||
return (size*nmemb);
|
||||
}
|
||||
|
||||
static char *fetchMOTDFromCache()
|
||||
{
|
||||
struct stat attrib;
|
||||
if (stat(szMotdCachePath(), &attrib) != 0)
|
||||
return NULL;
|
||||
time_t t = attrib.st_mtim.tv_sec;
|
||||
time_t now = time(NULL);
|
||||
if ((now - t) < 14400)
|
||||
{
|
||||
// If our cache was updated no more than 4 hours ago use it instead of fetching the MOTD
|
||||
FILE *pf = fopen(szMotdCachePath(), "rb");
|
||||
if (pf == NULL)
|
||||
return NULL;
|
||||
fseek(pf, 0L, SEEK_END);
|
||||
long cb = ftell(pf);
|
||||
fseek(pf, 0L, SEEK_SET); // rewind
|
||||
sds str = sdsnewlen(NULL, cb);
|
||||
size_t cbRead = fread(str, 1, cb, pf);
|
||||
fclose(pf);
|
||||
if ((long)cbRead != cb)
|
||||
{
|
||||
sdsfree(str);
|
||||
return NULL;
|
||||
}
|
||||
return str;
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static void setMOTDCache(const char *sz)
|
||||
{
|
||||
FILE *pf = fopen(szMotdCachePath(), "wb");
|
||||
if (pf == NULL)
|
||||
return;
|
||||
size_t celem = fwrite(sz, strlen(sz), 1, pf);
|
||||
(void)celem; // best effort
|
||||
fclose(pf);
|
||||
}
|
||||
|
||||
extern "C" char *fetchMOTD(int cache)
|
||||
{
|
||||
sds str;
|
||||
CURL *curl;
|
||||
CURLcode res;
|
||||
|
||||
/* First try and get the string from the cache */
|
||||
if (cache) {
|
||||
str = fetchMOTDFromCache();
|
||||
if (str != NULL)
|
||||
return str;
|
||||
}
|
||||
|
||||
str = sdsnew("");
|
||||
curl = curl_easy_init();
|
||||
if(curl) {
|
||||
curl_easy_setopt(curl, CURLOPT_URL, motd_url);
|
||||
curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); // follow redirects
|
||||
curl_easy_setopt(curl, CURLOPT_TIMEOUT, 2); // take no more than two seconds
|
||||
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, motd_write_callback);
|
||||
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &str);
|
||||
|
||||
/* Perform the request, res will get the return code */
|
||||
res = curl_easy_perform(curl);
|
||||
/* Check for errors */
|
||||
if(res != CURLE_OK)
|
||||
{
|
||||
sdsfree(str);
|
||||
str = NULL;
|
||||
}
|
||||
else
|
||||
{
|
||||
long response_code;
|
||||
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &response_code);
|
||||
if ((response_code / 100) != 2)
|
||||
{
|
||||
// An error code not in the 200s implies an error
|
||||
sdsfree(str);
|
||||
str = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
/* always cleanup */
|
||||
curl_easy_cleanup(curl);
|
||||
|
||||
if (str != NULL && cache)
|
||||
setMOTDCache(str);
|
||||
}
|
||||
return str;
|
||||
}
|
||||
|
||||
#else
|
||||
|
||||
extern "C" char *fetchMOTD(int /* cache */)
|
||||
{
|
||||
return NULL;
|
||||
}
|
||||
|
||||
#endif
|
||||
14
src/motd.h
Normal file
14
src/motd.h
Normal file
@ -0,0 +1,14 @@
|
||||
#pragma once
|
||||
|
||||
extern const char *motd_url;
|
||||
extern const char *motd_cache_file;
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
char *fetchMOTD(int fCache);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
@ -174,6 +174,7 @@ client *createClient(connection *conn, int iel) {
|
||||
c->client_tracking_redirection = 0;
|
||||
c->casyncOpsPending = 0;
|
||||
c->mvccCheckpoint = 0;
|
||||
c->master_error = 0;
|
||||
memset(c->uuid, 0, UUID_BINARY_LEN);
|
||||
|
||||
c->auth_callback = NULL;
|
||||
@ -432,6 +433,34 @@ void addReplyProtoAsync(client *c, const char *s, size_t len) {
|
||||
addReplyProtoCore(c, s, len, true);
|
||||
}
|
||||
|
||||
std::string escapeString(sds str)
|
||||
{
|
||||
std::string newstr;
|
||||
size_t len = sdslen(str);
|
||||
for (size_t ich = 0; ich < len; ++ich)
|
||||
{
|
||||
char ch = str[ich];
|
||||
switch (ch)
|
||||
{
|
||||
case '\n':
|
||||
newstr += "\\n";
|
||||
break;
|
||||
|
||||
case '\t':
|
||||
newstr += "\\t";
|
||||
break;
|
||||
|
||||
case '\r':
|
||||
newstr += "\\r";
|
||||
break;
|
||||
|
||||
default:
|
||||
newstr += ch;
|
||||
}
|
||||
}
|
||||
return newstr;
|
||||
}
|
||||
|
||||
/* Low level function called by the addReplyError...() functions.
|
||||
* It emits the protocol for a Redis error, in the form:
|
||||
*
|
||||
@ -464,6 +493,12 @@ void addReplyErrorLengthCore(client *c, const char *s, size_t len, bool fAsync)
|
||||
serverLog(LL_WARNING,"== CRITICAL == This %s is sending an error "
|
||||
"to its %s: '%s' after processing the command "
|
||||
"'%s'", from, to, s, cmdname);
|
||||
|
||||
if (c->querybuf && sdslen(c->querybuf)) {
|
||||
std::string str = escapeString(c->querybuf);
|
||||
serverLog(LL_WARNING, "\tquerybuf: %s", str.c_str());
|
||||
}
|
||||
c->master_error = 1;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
118
src/redis-cli.c
118
src/redis-cli.c
@ -61,6 +61,7 @@
|
||||
#include "anet.h"
|
||||
#include "ae.h"
|
||||
#include "storage.h"
|
||||
#include "motd.h"
|
||||
|
||||
#include "redis-cli.h"
|
||||
|
||||
@ -95,6 +96,9 @@ int spectrum_palette_size;
|
||||
|
||||
int g_fInCrash = 0;
|
||||
|
||||
const char *motd_url = "http://api.keydb.dev/motd/motd_cli.txt";
|
||||
const char *motd_cache_file = "/.keydb-cli-motd";
|
||||
|
||||
/*------------------------------------------------------------------------------
|
||||
* Utility functions
|
||||
*--------------------------------------------------------------------------- */
|
||||
@ -6742,118 +6746,6 @@ static void intrinsicLatencyMode(void) {
|
||||
}
|
||||
}
|
||||
|
||||
/*------------------------------------------------------------------------------
|
||||
* Message of the day
|
||||
*--------------------------------------------------------------------------- */
|
||||
#ifdef MOTD
|
||||
#include <curl/curl.h>
|
||||
|
||||
static const char *szMotdCachePath()
|
||||
{
|
||||
static sds sdsMotdCachePath = NULL;
|
||||
if (sdsMotdCachePath != NULL)
|
||||
return sdsMotdCachePath;
|
||||
|
||||
struct passwd *pw = getpwuid(getuid());
|
||||
if (pw == NULL)
|
||||
return "";
|
||||
const char *homedir = pw->pw_dir;
|
||||
sdsMotdCachePath = sdsnew(homedir);
|
||||
sdsMotdCachePath = sdscat(sdsMotdCachePath, "/.keydb-cli-motd");
|
||||
return sdsMotdCachePath;
|
||||
}
|
||||
static size_t motd_write_callback(void *ptr, size_t size, size_t nmemb, sds *str)
|
||||
{
|
||||
*str = sdscatlen(*str, ptr, size*nmemb);
|
||||
return (size*nmemb);
|
||||
}
|
||||
|
||||
static char *fetchMOTDFromCache()
|
||||
{
|
||||
struct stat attrib;
|
||||
if (stat(szMotdCachePath(), &attrib) != 0)
|
||||
return NULL;
|
||||
time_t t = attrib.st_mtim.tv_sec;
|
||||
time_t now = time(NULL);
|
||||
if ((now - t) < 14400)
|
||||
{
|
||||
// If our cache was updated no more than 4 hours ago use it instead of fetching the MOTD
|
||||
FILE *pf = fopen(szMotdCachePath(), "rb");
|
||||
if (pf == NULL)
|
||||
return NULL;
|
||||
fseek(pf, 0L, SEEK_END);
|
||||
long cb = ftell(pf);
|
||||
fseek(pf, 0L, SEEK_SET); // rewind
|
||||
sds str = sdsnewlen(NULL, cb);
|
||||
size_t cbRead = fread(str, 1, cb, pf);
|
||||
fclose(pf);
|
||||
if ((long)cbRead != cb)
|
||||
{
|
||||
sdsfree(str);
|
||||
return NULL;
|
||||
}
|
||||
return str;
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static void setMOTDCache(const char *sz)
|
||||
{
|
||||
FILE *pf = fopen(szMotdCachePath(), "wb");
|
||||
if (pf == NULL)
|
||||
return;
|
||||
size_t celem = fwrite(sz, strlen(sz), 1, pf);
|
||||
(void)celem; // best effort
|
||||
fclose(pf);
|
||||
}
|
||||
|
||||
static char *fetchMOTD()
|
||||
{
|
||||
sds str;
|
||||
CURL *curl;
|
||||
CURLcode res;
|
||||
|
||||
/* First try and get the string from the cache */
|
||||
str = fetchMOTDFromCache();
|
||||
if (str != NULL)
|
||||
return str;
|
||||
|
||||
str = sdsnew("");
|
||||
curl = curl_easy_init();
|
||||
if(curl) {
|
||||
curl_easy_setopt(curl, CURLOPT_URL, "http://api.keydb.dev/motd/motd.txt");
|
||||
curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); // follow redirects
|
||||
curl_easy_setopt(curl, CURLOPT_TIMEOUT, 2); // take no more than two seconds
|
||||
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, motd_write_callback);
|
||||
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &str);
|
||||
|
||||
/* Perform the request, res will get the return code */
|
||||
res = curl_easy_perform(curl);
|
||||
/* Check for errors */
|
||||
if(res != CURLE_OK)
|
||||
{
|
||||
sdsfree(str);
|
||||
str = NULL;
|
||||
}
|
||||
|
||||
/* always cleanup */
|
||||
curl_easy_cleanup(curl);
|
||||
|
||||
if (str != NULL)
|
||||
setMOTDCache(str);
|
||||
}
|
||||
return str;
|
||||
}
|
||||
|
||||
#else
|
||||
|
||||
static char *fetchMOTD()
|
||||
{
|
||||
return NULL;
|
||||
}
|
||||
|
||||
#endif
|
||||
|
||||
/*------------------------------------------------------------------------------
|
||||
* Program main()
|
||||
*--------------------------------------------------------------------------- */
|
||||
@ -7025,7 +6917,7 @@ int main(int argc, char **argv) {
|
||||
if (argc == 0 && !config.eval) {
|
||||
/* Show the message of the day if we are interactive */
|
||||
if (config.output == OUTPUT_STANDARD) {
|
||||
char *szMotd = fetchMOTD();
|
||||
char *szMotd = fetchMOTD(1 /* cache */);
|
||||
if (szMotd != NULL) {
|
||||
printf("Message of the day:\n %s\n", szMotd);
|
||||
sdsfree(szMotd);
|
||||
|
||||
@ -3960,6 +3960,8 @@ void replicaReplayCommand(client *c)
|
||||
cFake->flags &= ~(CLIENT_MASTER | CLIENT_PREVENT_REPL_PROP);
|
||||
bool fExec = ccmdPrev != serverTL->commandsExecuted;
|
||||
cFake->lock.unlock();
|
||||
if (cFake->master_error)
|
||||
addReplyError(c, "Error in rreplay command, please check logs");
|
||||
if (fExec || cFake->flags & CLIENT_MULTI)
|
||||
{
|
||||
addReply(c, shared.ok);
|
||||
|
||||
@ -63,8 +63,11 @@
|
||||
#include <mutex>
|
||||
#include "aelocker.h"
|
||||
#include "keycheck.h"
|
||||
#include "motd.h"
|
||||
|
||||
int g_fTestMode = false;
|
||||
const char *motd_url = "http://api.keydb.dev/motd/motd_server_pro.txt";
|
||||
const char *motd_cache_file = "/.keydb-pro-server-motd";
|
||||
|
||||
/* Our shared "common" objects */
|
||||
|
||||
@ -1103,6 +1106,7 @@ void serverLog(int level, const char *fmt, ...) {
|
||||
|
||||
static void checkTrialTimeout()
|
||||
{
|
||||
#ifndef NO_LICENSE_CHECK
|
||||
if (cserver.license_key != nullptr && FValidKey(cserver.license_key, strlen(cserver.license_key)))
|
||||
return;
|
||||
time_t curtime = time(NULL);
|
||||
@ -1118,6 +1122,7 @@ static void checkTrialTimeout()
|
||||
{
|
||||
serverLog(LL_WARNING, "Trial timeout in %ld:%02ld minutes", remaining/60, remaining % 60);
|
||||
}
|
||||
#endif
|
||||
}
|
||||
|
||||
/* Log a fixed message without printf-alike capabilities, in a way that is
|
||||
@ -1641,6 +1646,62 @@ int clientsCronResizeQueryBuffer(client *c) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
SymVer parseVersion(const char *version)
|
||||
{
|
||||
SymVer ver = {-1,-1,-1};
|
||||
long versions[3] = {-1,-1,-1};
|
||||
const char *start = version;
|
||||
const char *end = nullptr;
|
||||
|
||||
for (int iver = 0; iver < 3; ++iver)
|
||||
{
|
||||
end = start;
|
||||
while (*end != '\0' && *end != '.')
|
||||
++end;
|
||||
|
||||
if (start >= end)
|
||||
return ver;
|
||||
|
||||
if (!string2l(start, end - start, versions + iver))
|
||||
return ver;
|
||||
if (*end != '\0')
|
||||
start = end+1;
|
||||
else
|
||||
break;
|
||||
}
|
||||
ver.major = versions[0];
|
||||
ver.minor = versions[1];
|
||||
ver.build = versions[2];
|
||||
|
||||
return ver;
|
||||
}
|
||||
|
||||
VersionCompareResult compareVersion(SymVer *pver)
|
||||
{
|
||||
SymVer symVerThis = parseVersion(KEYDB_REAL_VERSION);
|
||||
for (int iver = 0; iver < 3; ++iver)
|
||||
{
|
||||
long verThis, verOther;
|
||||
switch (iver)
|
||||
{
|
||||
case 0:
|
||||
verThis = symVerThis.major; verOther = pver->major;
|
||||
break;
|
||||
case 1:
|
||||
verThis = symVerThis.minor; verOther = pver->minor;
|
||||
break;
|
||||
case 2:
|
||||
verThis = symVerThis.build; verOther = pver->build;
|
||||
}
|
||||
|
||||
if (verThis < verOther)
|
||||
return VersionCompareResult::NewerVersion;
|
||||
if (verThis > verOther)
|
||||
return VersionCompareResult::OlderVersion;
|
||||
}
|
||||
return VersionCompareResult::EqualVerison;
|
||||
}
|
||||
|
||||
/* This function is used in order to track clients using the biggest amount
|
||||
* of memory in the latest few seconds. This way we can provide such information
|
||||
* in the INFO output (clients section), without having to do an O(N) scan for
|
||||
@ -5002,14 +5063,18 @@ void redisAsciiArt(void) {
|
||||
mode, g_pserver->port ? g_pserver->port : g_pserver->tls_port
|
||||
);
|
||||
} else {
|
||||
sds motd = fetchMOTD(true);
|
||||
snprintf(buf,1024*16,ascii_logo,
|
||||
KEYDB_REAL_VERSION,
|
||||
redisGitSHA1(),
|
||||
strtol(redisGitDirty(),NULL,10) > 0,
|
||||
(sizeof(long) == 8) ? "64" : "32",
|
||||
mode, g_pserver->port ? g_pserver->port : g_pserver->tls_port,
|
||||
(long) getpid()
|
||||
(long) getpid(),
|
||||
motd ? motd : ""
|
||||
);
|
||||
if (motd)
|
||||
sdsfree(motd);
|
||||
serverLogRaw(LL_NOTICE|LL_RAW,buf);
|
||||
}
|
||||
|
||||
@ -5388,6 +5453,13 @@ int main(int argc, char **argv) {
|
||||
|
||||
std::set_terminate(OnTerminate);
|
||||
|
||||
{
|
||||
SymVer version;
|
||||
version = parseVersion(KEYDB_REAL_VERSION);
|
||||
serverAssert(version.major >= 0 && version.minor >= 0 && version.build >= 0);
|
||||
serverAssert(compareVersion(&version) == VersionCompareResult::EqualVerison);
|
||||
}
|
||||
|
||||
#ifdef USE_MEMKIND
|
||||
storage_init(NULL, 0);
|
||||
#endif
|
||||
|
||||
@ -57,6 +57,7 @@
|
||||
#include <memory>
|
||||
#include <set>
|
||||
#include <map>
|
||||
#include <string>
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#include <lua.h>
|
||||
@ -1686,6 +1687,7 @@ typedef struct client {
|
||||
|
||||
int iel; /* the event loop index we're registered with */
|
||||
struct fastlock lock;
|
||||
int master_error;
|
||||
} client;
|
||||
|
||||
struct saveparam {
|
||||
|
||||
@ -99,6 +99,15 @@ const IStorage *RocksDBStorageProvider::clone() const
|
||||
|
||||
RocksDBStorageProvider::~RocksDBStorageProvider()
|
||||
{
|
||||
if (m_spbatch != nullptr)
|
||||
endWriteBatch();
|
||||
|
||||
if (m_spdb != nullptr && m_psnapshot == nullptr)
|
||||
{
|
||||
insert(count_key, sizeof(count_key), &m_count, sizeof(m_count), false);
|
||||
flush();
|
||||
}
|
||||
|
||||
if (m_spdb != nullptr)
|
||||
{
|
||||
if (m_psnapshot != nullptr)
|
||||
|
||||
@ -5,6 +5,9 @@
|
||||
#include <rocksdb/db.h>
|
||||
#include "../fastlock.h"
|
||||
|
||||
static const char count_key[] = "\0__keydb__count\1";
|
||||
static const char version_key[] = "\0__keydb__version\1";
|
||||
|
||||
class RocksDBStorageProvider : public IStorage
|
||||
{
|
||||
std::shared_ptr<rocksdb::DB> m_spdb; // Note: This must be first so it is deleted last
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
#include "rocksdb.h"
|
||||
#include "../version.h"
|
||||
#include <rocksdb/filter_policy.h>
|
||||
#include <rocksdb/table.h>
|
||||
#include <rocksdb/utilities/options_util.h>
|
||||
@ -14,6 +15,9 @@ public:
|
||||
|
||||
virtual IStorage *create(int db) override;
|
||||
virtual const char *name() const override;
|
||||
|
||||
private:
|
||||
void setVersion(rocksdb::ColumnFamilyHandle*);
|
||||
};
|
||||
|
||||
IStorageFactory *CreateRocksDBStorageFactory(const char *path, int dbnum)
|
||||
@ -64,9 +68,26 @@ RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum)
|
||||
if (!status.ok())
|
||||
throw status.ToString();
|
||||
|
||||
for (auto handle : handles)
|
||||
m_vecspcols.emplace_back(handle);
|
||||
m_spdb = std::shared_ptr<rocksdb::DB>(db);
|
||||
for (auto handle : handles)
|
||||
{
|
||||
std::string strVersion;
|
||||
auto status = m_spdb->Get(rocksdb::ReadOptions(), handle, rocksdb::Slice(version_key, sizeof(version_key)), &strVersion);
|
||||
if (!status.ok())
|
||||
{
|
||||
setVersion(handle);
|
||||
}
|
||||
else
|
||||
{
|
||||
SymVer ver = parseVersion(strVersion.c_str());
|
||||
auto cmp = compareVersion(&ver);
|
||||
if (cmp == NewerVersion)
|
||||
throw "Cannot load FLASH database created by newer version of KeyDB";
|
||||
if (cmp == OlderVersion)
|
||||
setVersion(handle);
|
||||
}
|
||||
m_vecspcols.emplace_back(handle);
|
||||
}
|
||||
}
|
||||
|
||||
RocksDBStorageFactory::~RocksDBStorageFactory()
|
||||
@ -74,14 +95,33 @@ RocksDBStorageFactory::~RocksDBStorageFactory()
|
||||
m_spdb->SyncWAL();
|
||||
}
|
||||
|
||||
void RocksDBStorageFactory::setVersion(rocksdb::ColumnFamilyHandle *handle)
|
||||
{
|
||||
auto status = m_spdb->Put(rocksdb::WriteOptions(), handle, rocksdb::Slice(version_key, sizeof(version_key)), rocksdb::Slice(KEYDB_REAL_VERSION, strlen(KEYDB_REAL_VERSION)+1));
|
||||
if (!status.ok())
|
||||
throw status.ToString();
|
||||
}
|
||||
|
||||
IStorage *RocksDBStorageFactory::create(int db)
|
||||
{
|
||||
++db; // skip default col family
|
||||
std::shared_ptr<rocksdb::ColumnFamilyHandle> spcolfamily(m_vecspcols[db].release());
|
||||
size_t count = 0;
|
||||
std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(rocksdb::ReadOptions(), spcolfamily.get()));
|
||||
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
||||
++count;
|
||||
|
||||
std::string value;
|
||||
auto status = m_spdb->Get(rocksdb::ReadOptions(), spcolfamily.get(), rocksdb::Slice(count_key, sizeof(count_key)), &value);
|
||||
if (status.ok() && value.size() == sizeof(size_t))
|
||||
{
|
||||
count = *reinterpret_cast<const size_t*>(value.data());
|
||||
m_spdb->Delete(rocksdb::WriteOptions(), spcolfamily.get(), rocksdb::Slice(count_key, sizeof(count_key)));
|
||||
}
|
||||
else
|
||||
{
|
||||
printf("\tDatabase was not shutdown cleanly, recomputing metrics\n");
|
||||
std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(rocksdb::ReadOptions(), spcolfamily.get()));
|
||||
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
||||
++count;
|
||||
}
|
||||
}
|
||||
return new RocksDBStorageProvider(m_spdb, spcolfamily, nullptr, count);
|
||||
}
|
||||
|
||||
@ -1,3 +1,28 @@
|
||||
#define KEYDB_REAL_VERSION "6.0.2"
|
||||
extern const char *KEYDB_SET_VERSION; // Unlike real version, this can be overriden by the config
|
||||
|
||||
enum VersionCompareResult
|
||||
{
|
||||
EqualVerison,
|
||||
OlderVersion,
|
||||
NewerVersion,
|
||||
};
|
||||
|
||||
struct SymVer
|
||||
{
|
||||
long major;
|
||||
long minor;
|
||||
long build;
|
||||
};
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C"
|
||||
{
|
||||
#endif
|
||||
|
||||
struct SymVer parseVersion(const char *version);
|
||||
enum VersionCompareResult compareVersion(struct SymVer *pver);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
@ -1,47 +1,47 @@
|
||||
start_server {tags {"CRON"} overrides {hz 100} } {
|
||||
test {cron singleshot past tense} {
|
||||
test {keydb.cron singleshot past tense} {
|
||||
r flushall
|
||||
r cron testjob single 0 1 {redis.call("incr", "testkey")} 1 testkey
|
||||
r keydb.cron testjob single 0 1 {redis.call("incr", "testkey")} 1 testkey
|
||||
after 300
|
||||
assert_equal 1 [r get testkey]
|
||||
assert_equal 0 [r exists testjob]
|
||||
}
|
||||
|
||||
test {cron repeat past tense next exec is in the future} {
|
||||
test {keydb.cron repeat past tense next exec is in the future} {
|
||||
r flushall
|
||||
r cron testjob repeat 0 1000000 {redis.call("incr", "testkey")} 1 testkey
|
||||
r keydb.cron testjob repeat 0 1000000 {redis.call("incr", "testkey")} 1 testkey
|
||||
after 300
|
||||
assert_equal 1 [r get testkey]
|
||||
assert_equal 1 [r exists testjob]
|
||||
r del testjob
|
||||
}
|
||||
|
||||
test {cron repeat works} {
|
||||
test {keydb.cron repeat works} {
|
||||
r flushall
|
||||
r cron testjob repeat 0 600 {redis.call("incr","testkey")}
|
||||
r keydb.cron testjob repeat 0 600 {redis.call("incr","testkey")}
|
||||
after 1000
|
||||
assert_equal 2 [r get testkey]
|
||||
}
|
||||
|
||||
test {cron overwrite works} {
|
||||
test {keydb.cron overwrite works} {
|
||||
r flushall
|
||||
r cron testjob single 500 {redis.call("set","testkey","a")} 1 testkey
|
||||
r cron testjob single 500 {redis.call("set","anotherkey","b")} 1 anotherkey
|
||||
r keydb.cron testjob single 500 {redis.call("set","testkey","a")} 1 testkey
|
||||
r keydb.cron testjob single 500 {redis.call("set","anotherkey","b")} 1 anotherkey
|
||||
after 1000
|
||||
assert_equal 0 [r exists testkey]
|
||||
assert_equal b [r get anotherkey]
|
||||
}
|
||||
|
||||
test {cron delete key stops job} {
|
||||
test {keydb.cron delete key stops job} {
|
||||
r flushall
|
||||
r cron testjob single 500 {redis.call("set","testkey","a")}
|
||||
r keydb.cron testjob single 500 {redis.call("set","testkey","a")}
|
||||
r del testjob
|
||||
after 1000
|
||||
assert_equal 0 [r exists testkey]
|
||||
}
|
||||
|
||||
test {cron zero interval rejected} {
|
||||
catch {r cron testjob single 0 0 {redis.call("incr","testkey")} 1 testkey} e
|
||||
test {keydb.cron zero interval rejected} {
|
||||
catch {r keydb.cron testjob single 0 0 {redis.call("incr","testkey")} 1 testkey} e
|
||||
assert_match {ERR*} $e
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user