mirror of
https://github.com/hanzoai/redis.git
synced 2026-10-30 18:25:12 +00:00
ca6e3dafb2
Add new module API `RM_GetClusterNodeSlotRanges` that allows modules to query slot ranges for any cluster node by its node ID, not just the local node: ```c RedisModuleSlotRangeArray *RM_GetClusterNodeSlotRanges(RedisModuleCtx *ctx, const char *nodeid) ``` --------- Co-authored-by: Ozan Tezcan <[email protected]> Co-authored-by: Yuan Wang <[email protected]>
628 lines
25 KiB
C
628 lines
25 KiB
C
#include "redismodule.h"
|
|
|
|
#include <stdlib.h>
|
|
#include <memory.h>
|
|
#include <errno.h>
|
|
|
|
#define MAX_EVENTS 1024
|
|
|
|
/* Log of cluster events. */
|
|
const char *clusterEventLog[MAX_EVENTS];
|
|
int numClusterEvents = 0;
|
|
|
|
/* Log of cluster trim events. */
|
|
const char *clusterTrimEventLog[MAX_EVENTS];
|
|
int numClusterTrimEvents = 0;
|
|
|
|
/* Log of last deleted key event. */
|
|
const char *lastDeletedKeyLog = NULL;
|
|
|
|
/* Flag to disable trim. */
|
|
int disableTrimFlag = 0;
|
|
|
|
int replicateModuleCommand = 0; /* Enable or disable module command replication. */
|
|
RedisModuleString *moduleCommandKeyName = NULL; /* Key name to replicate. */
|
|
RedisModuleString *moduleCommandKeyVal = NULL; /* Key value to replicate. */
|
|
|
|
/* Enable or disable module command replication. */
|
|
int replicate_module_command(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
if (argc != 4) {
|
|
RedisModule_ReplyWithError(ctx, "ERR wrong number of arguments");
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
long long enable = 0;
|
|
if (RedisModule_StringToLongLong(argv[1], &enable) != REDISMODULE_OK) {
|
|
RedisModule_ReplyWithError(ctx, "ERR enable value");
|
|
return REDISMODULE_OK;
|
|
}
|
|
replicateModuleCommand = (enable != 0);
|
|
|
|
/* Set the key name and value to replicate. */
|
|
if (moduleCommandKeyName) RedisModule_FreeString(ctx, moduleCommandKeyName);
|
|
if (moduleCommandKeyVal) RedisModule_FreeString(ctx, moduleCommandKeyVal);
|
|
moduleCommandKeyName = RedisModule_CreateStringFromString(ctx, argv[2]);
|
|
moduleCommandKeyVal = RedisModule_CreateStringFromString(ctx, argv[3]);
|
|
|
|
RedisModule_ReplyWithSimpleString(ctx, "OK");
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
int lpush_and_replicate_crossslot_command(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
if (argc != 3) return RedisModule_WrongArity(ctx);
|
|
|
|
/* LPUSH */
|
|
RedisModuleCallReply *rep = RedisModule_Call(ctx, "LPUSH", "!ss", argv[1], argv[2]);
|
|
RedisModule_Assert(RedisModule_CallReplyType(rep) != REDISMODULE_REPLY_ERROR);
|
|
RedisModule_FreeCallReply(rep);
|
|
|
|
/* Replicate cross slot command */
|
|
int ret = RedisModule_Replicate(ctx, "MSET", "cccccc", "key1", "val1", "key2", "val2", "key3", "val3");
|
|
RedisModule_Assert(ret == REDISMODULE_OK);
|
|
|
|
RedisModule_ReplyWithSimpleString(ctx, "OK");
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
int testClusterGetLocalSlotRanges(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
|
|
static int use_auto_memory = 0;
|
|
use_auto_memory = !use_auto_memory;
|
|
|
|
RedisModuleSlotRangeArray *slots;
|
|
if (use_auto_memory) {
|
|
RedisModule_AutoMemory(ctx);
|
|
slots = RedisModule_ClusterGetLocalSlotRanges(ctx);
|
|
} else {
|
|
slots = RedisModule_ClusterGetLocalSlotRanges(NULL);
|
|
}
|
|
|
|
RedisModule_ReplyWithArray(ctx, slots->num_ranges);
|
|
for (int i = 0; i < slots->num_ranges; i++) {
|
|
RedisModule_ReplyWithArray(ctx, 2);
|
|
RedisModule_ReplyWithLongLong(ctx, slots->ranges[i].start);
|
|
RedisModule_ReplyWithLongLong(ctx, slots->ranges[i].end);
|
|
}
|
|
if (!use_auto_memory)
|
|
RedisModule_ClusterFreeSlotRanges(NULL, slots);
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
/* Test command for RedisModule_GetClusterNodeSlotRanges */
|
|
int testGetClusterNodeSlotRanges(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
if (argc != 2) {
|
|
return RedisModule_WrongArity(ctx);
|
|
}
|
|
|
|
const char *nodeid = RedisModule_StringPtrLen(argv[1], NULL);
|
|
|
|
static int use_auto_memory = 0;
|
|
use_auto_memory = !use_auto_memory;
|
|
|
|
RedisModuleSlotRangeArray *slots;
|
|
if (use_auto_memory) {
|
|
RedisModule_AutoMemory(ctx);
|
|
slots = RedisModule_GetClusterNodeSlotRanges(ctx, nodeid);
|
|
} else {
|
|
slots = RedisModule_GetClusterNodeSlotRanges(NULL, nodeid);
|
|
}
|
|
|
|
RedisModule_ReplyWithArray(ctx, slots->num_ranges);
|
|
for (int i = 0; i < slots->num_ranges; i++) {
|
|
RedisModule_ReplyWithArray(ctx, 2);
|
|
RedisModule_ReplyWithLongLong(ctx, slots->ranges[i].start);
|
|
RedisModule_ReplyWithLongLong(ctx, slots->ranges[i].end);
|
|
}
|
|
if (!use_auto_memory)
|
|
RedisModule_ClusterFreeSlotRanges(NULL, slots);
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
/* Helper function to check if a slot range array contains a given slot. */
|
|
int slotRangeArrayContains(RedisModuleSlotRangeArray *sra, unsigned int slot) {
|
|
for (int i = 0; i < sra->num_ranges; i++)
|
|
if (sra->ranges[i].start <= slot && sra->ranges[i].end >= slot)
|
|
return 1;
|
|
return 0;
|
|
}
|
|
|
|
/* Sanity check. */
|
|
int sanity(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
|
|
RedisModule_Assert(RedisModule_ClusterCanAccessKeysInSlot(-1) == 0);
|
|
RedisModule_Assert(RedisModule_ClusterCanAccessKeysInSlot(16384) == 0);
|
|
RedisModule_Assert(RedisModule_ClusterCanAccessKeysInSlot(100000) == 0);
|
|
|
|
/* Call with invalid args. */
|
|
errno = 0;
|
|
RedisModule_Assert(RedisModule_ClusterPropagateForSlotMigration(NULL, NULL, NULL) == REDISMODULE_ERR);
|
|
RedisModule_Assert(errno == EINVAL);
|
|
|
|
/* Call with invalid args. */
|
|
errno = 0;
|
|
RedisModule_Assert(RedisModule_ClusterPropagateForSlotMigration(ctx, NULL, NULL) == REDISMODULE_ERR);
|
|
RedisModule_Assert(errno == EINVAL);
|
|
|
|
/* Call with invalid args. */
|
|
errno = 0;
|
|
RedisModule_Assert(RedisModule_ClusterPropagateForSlotMigration(NULL, "asm.keyless_cmd", "") == REDISMODULE_ERR);
|
|
RedisModule_Assert(errno == EINVAL);
|
|
|
|
/* Call outside of slot migration. */
|
|
errno = 0;
|
|
RedisModule_Assert(RedisModule_ClusterPropagateForSlotMigration(ctx, "asm.keyless_cmd", "") == REDISMODULE_ERR);
|
|
RedisModule_Assert(errno == EBADF);
|
|
|
|
RedisModule_ReplyWithSimpleString(ctx, "OK");
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
/* Command to test RM_ClusterCanAccessKeysInSlot(). */
|
|
int testClusterCanAccessKeysInSlot(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(argc);
|
|
long long slot = 0;
|
|
|
|
if (RedisModule_StringToLongLong(argv[1],&slot) != REDISMODULE_OK) {
|
|
return RedisModule_ReplyWithError(ctx,"ERR invalid slot");
|
|
}
|
|
RedisModule_ReplyWithLongLong(ctx, RedisModule_ClusterCanAccessKeysInSlot(slot));
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
/* Generate a string representation of the info struct and subevent.
|
|
e.g. 'sub: cluster-slot-migration-import-started, task_id: aeBd..., slots: 0-100,200-300' */
|
|
const char *clusterAsmInfoToString(RedisModuleClusterSlotMigrationInfo *info, uint64_t sub) {
|
|
char buf[1024] = {0};
|
|
|
|
if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_IMPORT_STARTED)
|
|
snprintf(buf, sizeof(buf), "sub: cluster-slot-migration-import-started, ");
|
|
else if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_IMPORT_FAILED)
|
|
snprintf(buf, sizeof(buf), "sub: cluster-slot-migration-import-failed, ");
|
|
else if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_IMPORT_COMPLETED)
|
|
snprintf(buf, sizeof(buf), "sub: cluster-slot-migration-import-completed, ");
|
|
else if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_MIGRATE_STARTED)
|
|
snprintf(buf, sizeof(buf), "sub: cluster-slot-migration-migrate-started, ");
|
|
else if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_MIGRATE_FAILED)
|
|
snprintf(buf, sizeof(buf), "sub: cluster-slot-migration-migrate-failed, ");
|
|
else if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_MIGRATE_COMPLETED)
|
|
snprintf(buf, sizeof(buf), "sub: cluster-slot-migration-migrate-completed, ");
|
|
else {
|
|
RedisModule_Assert(0);
|
|
}
|
|
snprintf(buf + strlen(buf), sizeof(buf) - strlen(buf), "source_node_id:%.40s, destination_node_id:%.40s, ",
|
|
info->source_node_id, info->destination_node_id);
|
|
snprintf(buf + strlen(buf), sizeof(buf) - strlen(buf), "task_id:%s, slots:", info->task_id);
|
|
for (int i = 0; i < info->slots->num_ranges; i++) {
|
|
RedisModuleSlotRange *sr = &info->slots->ranges[i];
|
|
snprintf(buf + strlen(buf), sizeof(buf) - strlen(buf), "%d-%d", sr->start, sr->end);
|
|
if (i != info->slots->num_ranges - 1)
|
|
snprintf(buf + strlen(buf), sizeof(buf) - strlen(buf), ",");
|
|
}
|
|
return RedisModule_Strdup(buf);
|
|
}
|
|
|
|
/* Generate a string representation of the info struct and subevent.
|
|
e.g. 'sub: cluster-slot-migration-trim-started, task_id: aeBd..., slots:0-100,200-300' */
|
|
const char *clusterTrimInfoToString(RedisModuleClusterSlotMigrationTrimInfo *info, uint64_t sub) {
|
|
RedisModule_Assert(info);
|
|
char buf[1024] = {0};
|
|
|
|
if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_TRIM_BACKGROUND)
|
|
snprintf(buf, sizeof(buf), "sub: cluster-slot-migration-trim-background, ");
|
|
else if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_TRIM_STARTED)
|
|
snprintf(buf, sizeof(buf), "sub: cluster-slot-migration-trim-started, ");
|
|
else if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_TRIM_COMPLETED)
|
|
snprintf(buf, sizeof(buf), "sub: cluster-slot-migration-trim-completed, ");
|
|
else {
|
|
RedisModule_Assert(0);
|
|
}
|
|
snprintf(buf + strlen(buf), sizeof(buf) - strlen(buf), "slots:");
|
|
for (int i = 0; i < info->slots->num_ranges; i++) {
|
|
RedisModuleSlotRange *sr = &info->slots->ranges[i];
|
|
snprintf(buf + strlen(buf), sizeof(buf) - strlen(buf), "%d-%d", sr->start, sr->end);
|
|
if (i != info->slots->num_ranges - 1)
|
|
snprintf(buf + strlen(buf), sizeof(buf) - strlen(buf), ",");
|
|
}
|
|
return RedisModule_Strdup(buf);
|
|
}
|
|
|
|
static void testReplicatingOutsideSlotRange(RedisModuleCtx *ctx, RedisModuleClusterSlotMigrationInfo *info) {
|
|
int slot = 0;
|
|
while (slot >= 0 && slot <= 16383) {
|
|
if (!slotRangeArrayContains(info->slots, slot)) {
|
|
break;
|
|
}
|
|
slot++;
|
|
}
|
|
char buf[128] = {0};
|
|
const char *prefix = RedisModule_ClusterCanonicalKeyNameInSlot(slot);
|
|
snprintf(buf, sizeof(buf), "{%s}%s", prefix, "modulekey");
|
|
errno = 0;
|
|
int ret = RedisModule_ClusterPropagateForSlotMigration(ctx, "SET", "cc", buf, "value");
|
|
RedisModule_Assert(ret == REDISMODULE_ERR);
|
|
RedisModule_Assert(errno == ERANGE);
|
|
}
|
|
|
|
static void testReplicatingCrossslotCommand(RedisModuleCtx *ctx) {
|
|
errno = 0;
|
|
int ret = RedisModule_ClusterPropagateForSlotMigration(ctx, "MSET", "cccccc", "key1", "val1", "key2", "val2", "key3", "val3");
|
|
RedisModule_Assert(ret == REDISMODULE_ERR);
|
|
RedisModule_Assert(errno == ENOTSUP);
|
|
}
|
|
|
|
static void testReplicatingUnknownCommand(RedisModuleCtx *ctx) {
|
|
errno = 0;
|
|
int ret = RedisModule_ClusterPropagateForSlotMigration(ctx, "unknowncommand", "");
|
|
RedisModule_Assert(ret == REDISMODULE_ERR);
|
|
RedisModule_Assert(errno == ENOENT);
|
|
}
|
|
|
|
static void testNonFatalScenarios(RedisModuleCtx *ctx, RedisModuleClusterSlotMigrationInfo *info) {
|
|
testReplicatingOutsideSlotRange(ctx, info);
|
|
testReplicatingCrossslotCommand(ctx);
|
|
testReplicatingUnknownCommand(ctx);
|
|
}
|
|
|
|
int disableTrimCmd(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
disableTrimFlag = 1;
|
|
/* Only disable when MIGRATE_COMPLETED for simulating recommended usage. */
|
|
// RedisModule_ClusterDisableTrim(ctx)
|
|
RedisModule_ReplyWithSimpleString(ctx, "OK");
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
int enableTrimCmd(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
disableTrimFlag = 0;
|
|
RedisModule_Assert(RedisModule_ClusterEnableTrim(ctx) == REDISMODULE_OK);
|
|
RedisModule_ReplyWithSimpleString(ctx, "OK");
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
int trimInProgressCmd(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
uint64_t flags = RedisModule_GetContextFlags(ctx);
|
|
RedisModule_ReplyWithLongLong(ctx, !!(flags & REDISMODULE_CTX_FLAGS_TRIM_IN_PROGRESS));
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
void clusterEventCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
int ret;
|
|
|
|
RedisModule_Assert(RedisModule_IsSubEventSupported(e, sub));
|
|
|
|
if (e.id == REDISMODULE_EVENT_CLUSTER_SLOT_MIGRATION) {
|
|
RedisModuleClusterSlotMigrationInfo *info = data;
|
|
|
|
if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_MIGRATE_MODULE_PROPAGATE) {
|
|
/* Test some non-fatal scenarios. */
|
|
testNonFatalScenarios(ctx, info);
|
|
|
|
if (replicateModuleCommand == 0) return;
|
|
|
|
/* Replicate a keyless command. */
|
|
ret = RedisModule_ClusterPropagateForSlotMigration(ctx, "asm.keyless_cmd", "");
|
|
RedisModule_Assert(ret == REDISMODULE_OK);
|
|
|
|
/* Propagate configured key and value. */
|
|
ret = RedisModule_ClusterPropagateForSlotMigration(ctx, "SET", "ss", moduleCommandKeyName, moduleCommandKeyVal);
|
|
RedisModule_Assert(ret == REDISMODULE_OK);
|
|
} else {
|
|
/* Log the event. */
|
|
if (numClusterEvents >= MAX_EVENTS) return;
|
|
clusterEventLog[numClusterEvents++] = clusterAsmInfoToString(info, sub);
|
|
|
|
if (sub == REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_MIGRATE_COMPLETED) {
|
|
/* If users ask to disable trim, we disable trim. */
|
|
if (disableTrimFlag) {
|
|
RedisModule_Assert(RedisModule_ClusterDisableTrim(ctx) == REDISMODULE_OK);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
int getPendingTrimKeyCmd(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
if (argc != 2) {
|
|
RedisModule_ReplyWithError(ctx, "ERR wrong number of arguments");
|
|
return REDISMODULE_ERR;
|
|
}
|
|
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1],
|
|
REDISMODULE_READ | REDISMODULE_OPEN_KEY_ACCESS_TRIMMED);
|
|
if (!key) {
|
|
RedisModule_ReplyWithNull(ctx);
|
|
return REDISMODULE_OK;
|
|
}
|
|
if (RedisModule_KeyType(key) != REDISMODULE_KEYTYPE_STRING) {
|
|
RedisModule_ReplyWithError(ctx, "key is not a string");
|
|
return REDISMODULE_ERR;
|
|
}
|
|
size_t len;
|
|
const char *value = RedisModule_StringDMA(key, &len, 0);
|
|
RedisModule_ReplyWithStringBuffer(ctx, value, len);
|
|
RedisModule_CloseKey(key);
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
void clusterTrimEventCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
|
|
RedisModule_Assert(RedisModule_IsSubEventSupported(e, sub));
|
|
|
|
if (e.id == REDISMODULE_EVENT_CLUSTER_SLOT_MIGRATION_TRIM) {
|
|
/* Log the event. */
|
|
if (numClusterTrimEvents >= MAX_EVENTS) return;
|
|
RedisModuleClusterSlotMigrationTrimInfo *info = data;
|
|
clusterTrimEventLog[numClusterTrimEvents++] = clusterTrimInfoToString(info, sub);
|
|
}
|
|
}
|
|
|
|
static int keyspaceNotificationTrimmedCallback(RedisModuleCtx *ctx, int type, const char *event, RedisModuleString *key) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
|
|
RedisModule_Assert(type == REDISMODULE_NOTIFY_KEY_TRIMMED);
|
|
RedisModule_Assert(strcmp(event, "key_trimmed") == 0);
|
|
|
|
if (numClusterTrimEvents >= MAX_EVENTS) return REDISMODULE_OK;
|
|
|
|
/* Log the trimmed key event. */
|
|
size_t len;
|
|
const char *key_str = RedisModule_StringPtrLen(key, &len);
|
|
|
|
char buf[1024] = {0};
|
|
snprintf(buf, sizeof(buf), "keyspace: key_trimmed, key: %s", key_str);
|
|
|
|
clusterTrimEventLog[numClusterTrimEvents++] = RedisModule_Strdup(buf);
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
/* ASM.PARENT SET key value (just proxy to Redis SET) */
|
|
static int asmParentSet(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
if (argc != 4) return RedisModule_WrongArity(ctx);
|
|
RedisModuleCallReply *reply = RedisModule_Call(ctx, "SET", "ss", argv[2], argv[3]);
|
|
if (!reply) return RedisModule_ReplyWithError(ctx, "ERR internal");
|
|
RedisModule_ReplyWithCallReply(ctx, reply);
|
|
RedisModule_FreeCallReply(reply);
|
|
RedisModule_ReplicateVerbatim(ctx);
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
/* Clear both the cluster and trim event logs. */
|
|
int clearEventLog(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
|
|
for (int i = 0; i < numClusterEvents; i++)
|
|
RedisModule_Free((void *)clusterEventLog[i]);
|
|
numClusterEvents = 0;
|
|
|
|
for (int i = 0; i < numClusterTrimEvents; i++)
|
|
RedisModule_Free((void *)clusterTrimEventLog[i]);
|
|
numClusterTrimEvents = 0;
|
|
|
|
RedisModule_ReplyWithSimpleString(ctx, "OK");
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
/* Reply with the cluster event log. */
|
|
int getClusterEventLog(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
|
|
RedisModule_ReplyWithArray(ctx, numClusterEvents);
|
|
for (int i = 0; i < numClusterEvents; i++)
|
|
RedisModule_ReplyWithStringBuffer(ctx, clusterEventLog[i], strlen(clusterEventLog[i]));
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
/* Reply with the cluster trim event log. */
|
|
int getClusterTrimEventLog(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
|
|
RedisModule_ReplyWithArray(ctx, numClusterTrimEvents);
|
|
for (int i = 0; i < numClusterTrimEvents; i++)
|
|
RedisModule_ReplyWithStringBuffer(ctx, clusterTrimEventLog[i], strlen(clusterTrimEventLog[i]));
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
/* A keyless command to test module command replication. */
|
|
int moduledata = 0;
|
|
int keylessCmd(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
moduledata++;
|
|
RedisModule_ReplyWithLongLong(ctx, moduledata);
|
|
return REDISMODULE_OK;
|
|
}
|
|
int readkeylessCmdVal(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
RedisModule_ReplyWithLongLong(ctx, moduledata);
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
int subscribeTrimmedEvent(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
if (argc != 2)
|
|
return RedisModule_WrongArity(ctx);
|
|
|
|
long long subscribe = 0;
|
|
if (RedisModule_StringToLongLong(argv[1], &subscribe) != REDISMODULE_OK) {
|
|
RedisModule_ReplyWithError(ctx, "ERR subscribe value");
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
if (subscribe) {
|
|
/* Unsubscribe first to avoid duplicate subscription. */
|
|
RedisModule_UnsubscribeFromKeyspaceEvents(ctx, REDISMODULE_NOTIFY_KEY_TRIMMED, keyspaceNotificationTrimmedCallback);
|
|
int ret = RedisModule_SubscribeToKeyspaceEvents(ctx, REDISMODULE_NOTIFY_KEY_TRIMMED, keyspaceNotificationTrimmedCallback);
|
|
RedisModule_Assert(ret == REDISMODULE_OK);
|
|
} else {
|
|
int ret = RedisModule_UnsubscribeFromKeyspaceEvents(ctx, REDISMODULE_NOTIFY_KEY_TRIMMED, keyspaceNotificationTrimmedCallback);
|
|
RedisModule_Assert(ret == REDISMODULE_OK);
|
|
}
|
|
RedisModule_ReplyWithSimpleString(ctx, "OK");
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
void keyEventCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
REDISMODULE_NOT_USED(e);
|
|
|
|
if (sub == REDISMODULE_SUBEVENT_KEY_DELETED) {
|
|
RedisModuleKeyInfoV1 *ei = data;
|
|
RedisModuleKey *kp = ei->key;
|
|
RedisModuleString *key = (RedisModuleString *) RedisModule_GetKeyNameFromModuleKey(kp);
|
|
size_t keylen;
|
|
const char *keyname = RedisModule_StringPtrLen(key, &keylen);
|
|
|
|
/* Verify value can be read. It will be used to verify key's value can
|
|
* be read in a trim callback. */
|
|
size_t valuelen = 0;
|
|
const char *value = "";
|
|
RedisModuleKey *mk = RedisModule_OpenKey(ctx, key, REDISMODULE_READ);
|
|
if (RedisModule_KeyType(mk) == REDISMODULE_KEYTYPE_STRING) {
|
|
value = RedisModule_StringDMA(mk, &valuelen, 0);
|
|
}
|
|
RedisModule_CloseKey(mk);
|
|
|
|
char buf[1024] = {0};
|
|
snprintf(buf, sizeof(buf), "keyevent: key: %.*s, value: %.*s", (int) keylen, keyname, (int)valuelen, value);
|
|
|
|
if (lastDeletedKeyLog) RedisModule_Free((void *)lastDeletedKeyLog);
|
|
lastDeletedKeyLog = RedisModule_Strdup(buf);
|
|
}
|
|
}
|
|
|
|
int getLastDeletedKey(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
|
|
if (lastDeletedKeyLog) {
|
|
RedisModule_ReplyWithStringBuffer(ctx, lastDeletedKeyLog, strlen(lastDeletedKeyLog));
|
|
} else {
|
|
RedisModule_ReplyWithNull(ctx);
|
|
}
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
int asmGetCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(ctx);
|
|
|
|
if (argc != 2) return RedisModule_WrongArity(ctx);
|
|
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_READ);
|
|
if (key == NULL) {
|
|
RedisModule_ReplyWithNull(ctx);
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
RedisModule_Assert(RedisModule_KeyType(key) == REDISMODULE_KEYTYPE_STRING);
|
|
size_t len;
|
|
const char *value = RedisModule_StringDMA(key, &len, 0);
|
|
RedisModule_ReplyWithStringBuffer(ctx, value, len);
|
|
RedisModule_CloseKey(key);
|
|
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
REDISMODULE_NOT_USED(argv);
|
|
REDISMODULE_NOT_USED(argc);
|
|
|
|
if (RedisModule_Init(ctx, "asm", 1, REDISMODULE_APIVER_1) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.cluster_can_access_keys_in_slot", testClusterCanAccessKeysInSlot, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.clear_event_log", clearEventLog, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.get_cluster_event_log", getClusterEventLog, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.get_cluster_trim_event_log", getClusterTrimEventLog, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.keyless_cmd", keylessCmd, "write", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.disable_trim", disableTrimCmd, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.enable_trim", enableTrimCmd, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.read_pending_trim_key", getPendingTrimKeyCmd, "readonly", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.trim_in_progress", trimInProgressCmd, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.read_keyless_cmd_val", readkeylessCmdVal, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.sanity", sanity, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.subscribe_trimmed_event", subscribeTrimmedEvent, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.replicate_module_command", replicate_module_command, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.lpush_replicate_crossslot_command", lpush_and_replicate_crossslot_command, "write", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.cluster_get_local_slot_ranges", testClusterGetLocalSlotRanges, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.get_cluster_node_slot_ranges", testGetClusterNodeSlotRanges, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.get_last_deleted_key", getLastDeletedKey, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.get", asmGetCommand, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx, "asm.parent", NULL, "", 0, 0, 0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
RedisModuleCommand *parent = RedisModule_GetCommand(ctx, "asm.parent");
|
|
if (!parent) return REDISMODULE_ERR;
|
|
|
|
/* Subcommand: ASM.PARENT SET (write) */
|
|
if (RedisModule_CreateSubcommand(parent, "set", asmParentSet, "write fast", 2, 2, 1) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_SubscribeToServerEvent(ctx, RedisModuleEvent_ClusterSlotMigration, clusterEventCallback) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_SubscribeToServerEvent(ctx, RedisModuleEvent_ClusterSlotMigrationTrim, clusterTrimEventCallback) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_SubscribeToKeyspaceEvents(ctx, REDISMODULE_NOTIFY_KEY_TRIMMED, keyspaceNotificationTrimmedCallback) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_SubscribeToServerEvent(ctx, RedisModuleEvent_Key, keyEventCallback) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
return REDISMODULE_OK;
|
|
}
|