Skip to content

feat(commands): Introduce subcommand as a first-class command abstraction #3558

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
geetanshjuneja wants to merge 1 commit into
base: unstable
Choose a base branch
from
Open

feat(commands): Introduce subcommand as a first-class command abstraction #3558

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions kvrocks.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -676,7 +676,7 @@ profiling-sample-record-threshold-ms 100
# Time expression format is the same as crontab (supported cron syntax: *, n, */n, `1,3-6,9,11`)
# e.g. compaction-checker-cron * 0-7 * * * means compaction checker would be worker between
# 0-7am every day.
compaction-checker-cron * 0-7 * * *
compaction-checker-cron "* 0-7 * * *"

# When the compaction checker is triggered, the db will periodically pick the SST file
# with the highest "deleted percentage" (i.e. the percentage of deleted keys in the SST
Expand Down Expand Up @@ -1202,7 +1202,9 @@ rocksdb.sst_file_delete_rate_bytes_per_sec 0
# - Align resource-heavy operations with maintenance windows
#
# Reference: https://github.com/facebook/rocksdb/wiki/Daily-Off%E2%80%90peak-Time-Option
rocksdb.daily_offpeak_time_utc ""

################################ NAMESPACE #####################################
# namespace.test change.me
log-dir /tmp/kvrocks,stdout
namespace.xyz xyz
requirepass pass
166 changes: 116 additions & 50 deletions src/commands/cmd_server.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,17 @@

namespace redis {

namespace {

bool IsNamespaceCommandDisabled(Server *srv) { return srv->GetConfig()->redis_databases > 0; }

bool IsNamespaceReadOnlyOnSlave(Server *srv) {
Config *config = srv->GetConfig();
return config->repl_namespace_enabled && config->IsSlave();
}

} // namespace

class CommandAuth : public Commander {
public:
Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
Expand All @@ -65,63 +76,113 @@ class CommandAuth : public Commander {
};

class CommandNamespace : public Commander {
public:
Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, [[maybe_unused]] Connection *conn,
[[maybe_unused]] std::string *output) override {
if (IsNamespaceCommandDisabled(srv)) {
return {Status::RedisExecErr, "namespace command is not allowed when redis-databases > 0"};
}

return {Status::RedisExecErr, "NAMESPACE subcommand must be one of GET, SET, DEL, ADD and CURRENT"};
}
};

class CommandNamespaceGet : public Commander {
public:
Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
Config *config = srv->GetConfig();
std::string sub_command = util::ToLower(args_[1]);
if (config->repl_namespace_enabled && config->IsSlave() && sub_command != "get") {
return {Status::RedisExecErr, "namespace is read-only for slave"};
}
if (config->redis_databases > 0) {
if (IsNamespaceCommandDisabled(srv)) {
return {Status::RedisExecErr, "namespace command is not allowed when redis-databases > 0"};
}
if (args_.size() == 3 && sub_command == "get") {
if (args_[2] == "*") {
std::vector<std::string> namespaces;
auto tokens = srv->GetNamespace()->List();
for (auto &token : tokens) {
namespaces.emplace_back(token.second); // namespace
namespaces.emplace_back(token.first); // token
}
namespaces.emplace_back(kDefaultNamespace);
namespaces.emplace_back(config->requirepass);
*output = ArrayOfBulkStrings(namespaces);
} else {
auto token = srv->GetNamespace()->Get(args_[2]);
if (token.Is<Status::NotFound>()) {
*output = conn->NilString();
} else {
*output = redis::BulkString(token.GetValue());
}

if (args_[2] == "*") {
std::vector<std::string> namespaces;
auto tokens = srv->GetNamespace()->List();
for (auto &token : tokens) {
namespaces.emplace_back(token.second);
namespaces.emplace_back(token.first);
}
} else if (args_.size() == 4 && sub_command == "set") {
Status s = srv->GetNamespace()->Set(args_[2], args_[3]);
*output = s.IsOK() ? redis::RESP_OK : redis::Error(s);
WARN("Updated namespace: {} with token: {}, addr: {}, result: {}", args_[2], args_[3], conn->GetAddr(), s.Msg());
} else if (args_.size() == 4 && sub_command == "add") {
Status s = srv->GetNamespace()->Add(args_[2], args_[3]);
*output = s.IsOK() ? redis::RESP_OK : redis::Error(s);
WARN("New namespace: {} with token: {}, addr: {}, result: {}", args_[2], args_[3], conn->GetAddr(), s.Msg());
} else if (args_.size() == 3 && sub_command == "del") {
Status s = srv->GetNamespace()->Del(args_[2]);
*output = s.IsOK() ? redis::RESP_OK : redis::Error(s);
WARN("Deleted namespace: {}, addr: {}, result: {}", args_[2], conn->GetAddr(), s.Msg());
} else if (args_.size() == 2 && sub_command == "current") {
*output = redis::BulkString(conn->GetNamespace());
namespaces.emplace_back(kDefaultNamespace);
namespaces.emplace_back(config->requirepass);
*output = ArrayOfBulkStrings(namespaces);
return Status::OK();
}

auto token = srv->GetNamespace()->Get(args_[2]);
if (token.Is<Status::NotFound>()) {
*output = conn->NilString();
} else {
return {Status::RedisExecErr, "NAMESPACE subcommand must be one of GET, SET, DEL, ADD and CURRENT"};
*output = redis::BulkString(token.GetValue());
}
return Status::OK();
}
};

class CommandNamespaceSet : public Commander {
public:
Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
if (IsNamespaceReadOnlyOnSlave(srv)) {
return {Status::RedisExecErr, "namespace is read-only for slave"};
}
if (IsNamespaceCommandDisabled(srv)) {
return {Status::RedisExecErr, "namespace command is not allowed when redis-databases > 0"};
}

auto s = srv->GetNamespace()->Set(args_[2], args_[3]);
*output = s.IsOK() ? redis::RESP_OK : redis::Error(s);
WARN("Updated namespace: {} with token: {}, addr: {}, result: {}", args_[2], args_[3], conn->GetAddr(), s.Msg());
return Status::OK();
}
};

static uint64_t GenerateNamespaceFlag(uint64_t flags, const std::vector<std::string> &args) {
if (args.size() >= 2 && util::EqualICase(args[1], "current")) {
return flags & ~kCmdAdmin;
class CommandNamespaceAdd : public Commander {
public:
Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
if (IsNamespaceReadOnlyOnSlave(srv)) {
return {Status::RedisExecErr, "namespace is read-only for slave"};
}
if (IsNamespaceCommandDisabled(srv)) {
return {Status::RedisExecErr, "namespace command is not allowed when redis-databases > 0"};
}

auto s = srv->GetNamespace()->Add(args_[2], args_[3]);
*output = s.IsOK() ? redis::RESP_OK : redis::Error(s);
WARN("New namespace: {} with token: {}, addr: {}, result: {}", args_[2], args_[3], conn->GetAddr(), s.Msg());
return Status::OK();
}
};

return flags;
}
class CommandNamespaceDel : public Commander {
public:
Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
if (IsNamespaceReadOnlyOnSlave(srv)) {
return {Status::RedisExecErr, "namespace is read-only for slave"};
}
if (IsNamespaceCommandDisabled(srv)) {
return {Status::RedisExecErr, "namespace command is not allowed when redis-databases > 0"};
}

auto s = srv->GetNamespace()->Del(args_[2]);
*output = s.IsOK() ? redis::RESP_OK : redis::Error(s);
WARN("Deleted namespace: {}, addr: {}, result: {}", args_[2], conn->GetAddr(), s.Msg());
return Status::OK();
}
};

class CommandNamespaceCurrent : public Commander {
public:
Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
if (IsNamespaceReadOnlyOnSlave(srv)) {
return {Status::RedisExecErr, "namespace is read-only for slave"};
}
if (IsNamespaceCommandDisabled(srv)) {
return {Status::RedisExecErr, "namespace command is not allowed when redis-databases > 0"};
}

*output = redis::BulkString(conn->GetNamespace());
return Status::OK();
}
};

class CommandKeys : public Commander {
public:
Expand Down Expand Up @@ -865,13 +926,13 @@ class CommandCommand : public Commander {
} else if (sub_command == "info") {
CommandTable::GetCommandsInfo(output, std::vector<std::string>(args_.begin() + 2, args_.end()));
} else if (sub_command == "getkeys") {
auto cmd_iter = CommandTable::GetOriginal()->find(util::ToLower(args_[2]));
if (cmd_iter == CommandTable::GetOriginal()->end()) {
std::vector<std::string> cmd_tokens(args_.begin() + 2, args_.end());
auto resolved = CommandTable::Resolve(cmd_tokens);
if (!resolved) {
return {Status::RedisUnknownCmd, "Invalid command specified"};
}

auto key_indexes = GET_OR_RET(CommandTable::GetKeysFromCommand(
cmd_iter->second, std::vector<std::string>(args_.begin() + 2, args_.end())));
auto key_indexes = GET_OR_RET(CommandTable::GetKeysFromCommand(resolved->attributes, cmd_tokens));

if (key_indexes.size() == 0) {
return {Status::RedisExecErr, "Invalid arguments specified for command"};
Expand All @@ -880,7 +941,7 @@ class CommandCommand : public Commander {
std::vector<std::string> keys;
keys.reserve(key_indexes.size());
for (const auto &key_index : key_indexes) {
keys.emplace_back(args_[key_index + 2]);
keys.emplace_back(cmd_tokens[key_index]);
}
*output = conn->MultiBulkString(keys);
} else {
Expand Down Expand Up @@ -1766,7 +1827,7 @@ REDIS_REGISTER_COMMANDS(
MakeCmdAttr<CommandInfo>("info", -1, "read-only ok-loading", NO_KEY),
MakeCmdAttr<CommandRole>("role", 1, "read-only ok-loading", NO_KEY),
MakeCmdAttr<CommandConfig>("config", -2, "read-only admin skip-monitor", NO_KEY, GenerateConfigFlag),
MakeCmdAttr<CommandNamespace>("namespace", -2, "read-only admin skip-monitor", NO_KEY, GenerateNamespaceFlag),
MakeCmdAttr<CommandNamespace>("namespace", -2, "read-only admin skip-monitor", NO_KEY),
MakeCmdAttr<CommandKeys>("keys", 2, "read-only slow", NO_KEY),
MakeCmdAttr<CommandFlushDB>("flushdb", 1, "write no-dbsize-check exclusive", NO_KEY),
MakeCmdAttr<CommandFlushAll>("flushall", 1, "write no-dbsize-check exclusive admin", NO_KEY),
Expand Down Expand Up @@ -1803,5 +1864,10 @@ REDIS_REGISTER_COMMANDS(
MakeCmdAttr<CommandSST>("sst", -3, "write exclusive admin", 1, 1, 1),
MakeCmdAttr<CommandFlushMemTable>("flushmemtable", -1, "exclusive write", NO_KEY),
MakeCmdAttr<CommandFlushBlockCache>("flushblockcache", 1, "exclusive write", NO_KEY),
MakeCmdAttr<CommandLatency>("latency", -2, "read-only admin", NO_KEY), )
MakeCmdAttr<CommandLatency>("latency", -2, "read-only admin", NO_KEY),
MakeSubCmdAttr<CommandNamespaceGet>("namespace", "get", 3, "read-only admin skip-monitor", NO_KEY),
MakeSubCmdAttr<CommandNamespaceSet>("namespace", "set", 4, "read-only admin skip-monitor", NO_KEY),
MakeSubCmdAttr<CommandNamespaceAdd>("namespace", "add", 4, "read-only admin skip-monitor", NO_KEY),
MakeSubCmdAttr<CommandNamespaceDel>("namespace", "del", 3, "read-only admin skip-monitor", NO_KEY),
MakeSubCmdAttr<CommandNamespaceCurrent>("namespace", "current", 2, "read-only skip-monitor", NO_KEY))
} // namespace redis
99 changes: 94 additions & 5 deletions src/commands/commander.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,22 +20,83 @@

#include "commander.h"

#include <cstdlib>

#include "cluster/cluster_defs.h"
#include "server/redis_reply.h"

namespace redis {

bool CommandTable::isSubcommandName(const std::string &name) { return name.find('|') != std::string::npos; }

std::pair<std::string, std::string> CommandTable::parseSubcommandName(const std::string &name) {
auto delimiter = name.find('|');
if (delimiter == std::string::npos || delimiter == 0 || delimiter + 1 >= name.size()) {
std::cout << fmt::format("Encountered invalid subcommand name '{}'", name) << std::endl;
std::abort();
}

auto normalized_parent = util::ToLower(name.substr(0, delimiter));
auto normalized_sub = util::ToLower(name.substr(delimiter + 1));
return {normalized_parent, normalized_sub};
}

const CommandAttributes *CommandTable::registerCommand(CommandAttributes attr, CommandCategory category) {
if (original_commands.contains(attr.name) || commands.contains(attr.name)) {
std::cout << fmt::format("Duplicate command registration for '{}'", attr.name) << std::endl;
std::abort();
}

attr.category = category;
redis_command_table.emplace_back(std::move(attr));
auto *registered_attr = &redis_command_table.back();
original_commands[registered_attr->name] = registered_attr;
commands[registered_attr->name] = registered_attr;
return registered_attr;
}

const CommandAttributes *CommandTable::registerSubCommand(CommandAttributes attr, CommandCategory category) {
auto [parent, sub] = parseSubcommandName(attr.name);
auto &subcommand_family = sub_commands[parent];
if (subcommand_family.contains(sub)) {
std::cout << fmt::format("Duplicate subcommand registration for '{}|{}'", parent, sub) << std::endl;
std::abort();
}

attr.category = category;
attr.name = fmt::format("{}|{}", parent, sub);
redis_subcommand_table.emplace_back(std::move(attr));
auto *registered_attr = &redis_subcommand_table.back();
subcommand_family[sub] = registered_attr;
return registered_attr;
}

const CommandAttributes *CommandTable::findSubCommand(const std::string &parent, const std::string &sub) {
auto family_iter = sub_commands.find(util::ToLower(parent));
if (family_iter == sub_commands.end()) {
return nullptr;
}

auto subcommand_iter = family_iter->second.find(util::ToLower(sub));
if (subcommand_iter == family_iter->second.end()) {
return nullptr;
}

return subcommand_iter->second;
}

RegisterToCommandTable::RegisterToCommandTable(CommandCategory category,
std::initializer_list<CommandAttributes> list) {
if (category == CommandCategory::Disabled) {
return;
}

for (auto attr : list) {
attr.category = category;
CommandTable::redis_command_table.emplace_back(attr);
CommandTable::original_commands[attr.name] = &CommandTable::redis_command_table.back();
CommandTable::commands[attr.name] = &CommandTable::redis_command_table.back();
if (CommandTable::isSubcommandName(attr.name)) {
CommandTable::registerSubCommand(std::move(attr), category);
continue;
}
CommandTable::registerCommand(std::move(attr), category);
}
}

Expand Down Expand Up @@ -83,6 +144,32 @@ void CommandTable::GetCommandsInfo(std::string *info, const std::vector<std::str
}
}

StatusOr<ResolvedCommand> CommandTable::Resolve(const std::vector<std::string> &cmd_tokens) {
if (cmd_tokens.empty()) {
return {Status::RedisUnknownCmd, "No command specified"};
}

auto cmd_iter = commands.find(util::ToLower(cmd_tokens.front()));
if (cmd_iter == commands.end()) {
return {Status::RedisUnknownCmd, "Invalid command specified"};
}

const auto *root_attributes = cmd_iter->second;
ResolvedCommand resolved{root_attributes->name, root_attributes};

if (cmd_tokens.size() <= 1) {
return resolved;
}

auto subcommand_attributes = findSubCommand(root_attributes->name, cmd_tokens[1]);
if (subcommand_attributes == nullptr) {
return resolved;
}

resolved.attributes = subcommand_attributes;
return resolved;
}

StatusOr<std::vector<int>> CommandTable::GetKeysFromCommand(const CommandAttributes *attributes,
const std::vector<std::string> &cmd_tokens) {
int argc = static_cast<int>(cmd_tokens.size());
Expand All @@ -92,7 +179,9 @@ StatusOr<std::vector<int>> CommandTable::GetKeysFromCommand(const CommandAttribu
}

auto cmd = attributes->factory();
if (auto s = cmd->Parse(cmd_tokens); !s) {
cmd->SetAttributes(attributes);
cmd->SetArgs(cmd_tokens);
if (auto s = cmd->Parse(); !s) {
return {Status::NotOK, "Invalid syntax found in this command arguments: " + s.Msg()};
}

Expand Down
Loading
Loading