Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
38 changes: 32 additions & 6 deletions include/redis_command.h
Original file line number Diff line number Diff line change
Expand Up @@ -352,6 +352,9 @@ enum struct RedisCommandType
ELOQVEC_DELETE = 184,
ELOQVEC_SEARCH = 185,
#endif
#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
KEYSPACE = 186,
#endif
};

// Anchors: fail the build if any value above is accidentally shifted
Expand Down Expand Up @@ -381,6 +384,9 @@ static_assert(static_cast<int>(RedisCommandType::FAULT_INJECT) == 177);
static_assert(static_cast<int>(RedisCommandType::ELOQVEC_CREATE) == 178);
static_assert(static_cast<int>(RedisCommandType::ELOQVEC_SEARCH) == 185);
#endif
#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
static_assert(static_cast<int>(RedisCommandType::KEYSPACE) == 186);
#endif

enum RedisResultType
{
Expand Down Expand Up @@ -883,9 +889,6 @@ struct InfoCommand : public DirectCommand
std::string enable_wal_;
uint32_t node_memory_limit_mb_{0};

int64_t data_memory_allocated_{0};
int64_t data_memory_committed_{0};
uint64_t last_ckpt_ts_{0};
int event_dispatcher_num_{0};
std::string os_info_;
std::string executable_path_;
Expand All @@ -902,13 +905,33 @@ struct InfoCommand : public DirectCommand
int64_t multi_cmd_count_{0};
int64_t cmds_per_sec_{0};
double cmd_latency_{0};
std::vector<int64_t> dbsizes_;
};

struct MemoryStatsCommand : public DirectCommand
{
void Execute(RedisServiceImpl *redis_impl,
RedisConnectionContext *ctx) override;

void OutputResult(OutputHandler *reply) const override;

uint64_t data_memory_allocated_{0};
uint64_t data_memory_committed_{0};
uint64_t last_ckpt_ts_{0};
};

#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
struct KeyspaceCommand : public DirectCommand
{
KeyspaceCommand() = default;

void Execute(RedisServiceImpl *redis_impl,
RedisConnectionContext *ctx) override;
void OutputResult(OutputHandler *reply) const override;

std::vector<int64_t> dbsizes_;
uint64_t store_disk_keys_{0};
#endif
};

#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
struct CompactCommand : public DirectCommand
{
CompactCommand() = default;
Expand Down Expand Up @@ -7945,6 +7968,9 @@ std::tuple<bool, InfoCommand> ParseInfoCommand(
const std::vector<std::string_view> &args, OutputHandler *output);

#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
std::tuple<bool, KeyspaceCommand> ParseKeyspaceCommand(
const std::vector<std::string_view> &args, OutputHandler *output);

std::tuple<bool, CompactCommand> ParseCompactCommand(
const std::vector<std::string_view> &args, OutputHandler *output);
#endif
Expand Down
18 changes: 18 additions & 0 deletions include/redis_handler.h
Original file line number Diff line number Diff line change
Expand Up @@ -259,6 +259,24 @@ class InfoCommandHandler : public RedisCommandHandler
};

#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
class KeyspaceCommandHandler : public RedisCommandHandler
{
public:
explicit KeyspaceCommandHandler(RedisServiceImpl *redis_impl)
: redis_impl_(redis_impl)
{
}

brpc::RedisCommandHandlerResult Run(
RedisConnectionContext *ctx,
const std::vector<butil::StringPiece> &args,
brpc::RedisReply *output,
bool /*flush_batched*/) override;

private:
RedisServiceImpl *redis_impl_;
};

class CompactCommandHandler : public RedisCommandHandler
{
public:
Expand Down
150 changes: 89 additions & 61 deletions src/redis_command.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,7 @@ const std::vector<std::pair<const char *, RedisCommandType>> command_types{{
{"dbsize", RedisCommandType::DBSIZE},
#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
{"compact", RedisCommandType::COMPACT},
{"keyspace", RedisCommandType::KEYSPACE},
#endif
{"time", RedisCommandType::TIME},
{"slowlog", RedisCommandType::SLOWLOG},
Expand Down Expand Up @@ -1779,28 +1780,6 @@ void InfoCommand::Execute(RedisServiceImpl *redis_impl,
enable_wal_ = redis_impl->GetEnableWal();
node_memory_limit_mb_ = redis_impl->GetNodeMemoryLimitMB();

if (set_section_.size() == 0 ||
set_section_.find("memory") != set_section_.end())
{
auto &local_shards = redis_impl->GetTxService()->CcShards();
size_t cnt = local_shards.Count();
CkptTsCc ckpt_req(cnt, node_id_);
for (size_t i = 0; i < cnt; i++)
{
local_shards.EnqueueCcRequest(i, &ckpt_req);
}
ckpt_req.Wait();
data_memory_allocated_ = ckpt_req.GetMemUsage();
data_memory_committed_ = ckpt_req.GetMemCommited();
last_ckpt_ts_ = ckpt_req.GetCkptTs();
}
else
{
data_memory_allocated_ = -1;
data_memory_committed_ = -1;
last_ckpt_ts_ = -1;
}

event_dispatcher_num_ = redis_impl->GetEventDispatcherNum();
version_ = redis_impl->GetVersion();
os_info_ = redis_impl->GetOsInfo();
Expand All @@ -1820,29 +1799,44 @@ void InfoCommand::Execute(RedisServiceImpl *redis_impl,
multi_cmd_count_ = RedisStats::GetMultiObjectCommandsCount();
// cmds_per_sec_ = RedisStats::GetCommandsPerSecond();
}
}

#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
if (set_section_.size() == 0 ||
set_section_.find("keyspace") != set_section_.end())
void MemoryStatsCommand::Execute(RedisServiceImpl *redis_impl,
RedisConnectionContext *ctx)
{
uint32_t node_id = Sharder::Instance().NodeId();
auto &local_shards = redis_impl->GetTxService()->CcShards();
size_t cnt = local_shards.Count();
CkptTsCc ckpt_req(cnt, node_id);
for (size_t i = 0; i < cnt; i++)
{
std::vector<TableName> table_names;
size_t total_redis_table_cnt = redis_impl->GetRedisTableCount();
for (size_t db_id = 0; db_id < total_redis_table_cnt; ++db_id)
{
const TableName *tbn = redis_impl->RedisTableName(db_id);
table_names.emplace_back(
tbn->StringView(), tbn->Type(), TableEngine::EloqKv);
}

dbsizes_ = DBSizeCommand::FetchDBSize(std::move(table_names));
auto *store_hd = redis_impl->GetStoreHandler();
store_disk_keys_ =
store_hd == nullptr ? 0 : store_hd->ApproxStoreKeyCount();
local_shards.EnqueueCcRequest(i, &ckpt_req);
}
#endif
ckpt_req.Wait();
data_memory_allocated_ = ckpt_req.GetMemUsage();
data_memory_committed_ = ckpt_req.GetMemCommited();
last_ckpt_ts_ = ckpt_req.GetCkptTs();
}

#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
void KeyspaceCommand::Execute(RedisServiceImpl *redis_impl,
RedisConnectionContext *ctx)
{
std::vector<TableName> table_names;
size_t total_redis_table_cnt = redis_impl->GetRedisTableCount();
for (size_t db_id = 0; db_id < total_redis_table_cnt; ++db_id)
{
const TableName *tbn = redis_impl->RedisTableName(db_id);
table_names.emplace_back(
tbn->StringView(), tbn->Type(), TableEngine::EloqKv);
}

dbsizes_ = DBSizeCommand::FetchDBSize(std::move(table_names));
auto *store_hd = redis_impl->GetStoreHandler();
store_disk_keys_ =
store_hd == nullptr ? 0 : store_hd->ApproxStoreKeyCount();
}

void CompactCommand::Execute(RedisServiceImpl *redis_impl,
RedisConnectionContext *ctx)
{
Expand Down Expand Up @@ -2585,11 +2579,6 @@ void InfoCommand::OutputResult(OutputHandler *reply) const
std::to_string(GetProcessVmRSSKb(getpid())) + " kb";
result += "\r\ntotal_system_memory:" +
std::to_string(total_system_memory_kb_) + " kb";
result += "\r\ndata_memory_allocated:" +
std::to_string(data_memory_allocated_) + " kb";
result += "\r\ndata_memory_committed:" +
std::to_string(data_memory_committed_) + " kb";
// result += "data_memory_frag_ratio: " + ; // TODO
result +=
"\r\nnode_memory_limit:" + std::to_string(node_memory_limit_mb_) +
" mb ";
Expand All @@ -2605,7 +2594,6 @@ void InfoCommand::OutputResult(OutputHandler *reply) const

result += "# Persistence";

result += "\r\nlast_success_ckpt_ts:" + std::to_string(last_ckpt_ts_);
result += "\r\nenable_data_store:" + enable_data_store_;
result += "\r\nenable_wal:" + enable_wal_;
}
Expand Down Expand Up @@ -2680,31 +2668,48 @@ void InfoCommand::OutputResult(OutputHandler *reply) const
}
}

result += "\r\n";
reply->OnString(result);
}

void MemoryStatsCommand::OutputResult(OutputHandler *reply) const
{
reply->OnArrayStart(6);
reply->OnString("data_memory_allocated");
reply->OnInt(data_memory_allocated_);
reply->OnString("data_memory_committed");
reply->OnInt(data_memory_committed_);
reply->OnString("last_success_ckpt_ts");
reply->OnInt(last_ckpt_ts_);
reply->OnArrayEnd();
}

#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
if (set_section_.size() == 0 ||
set_section_.find("keyspace") != set_section_.end())
void KeyspaceCommand::OutputResult(OutputHandler *reply) const
{
size_t db_count = 0;
for (int64_t dbsize : dbsizes_)
{
if (result.size() > 0)
if (dbsize > 0)
{
result += "\r\n";
++db_count;
}
}

result += "# Keyspace";
for (size_t idx = 0; idx < dbsizes_.size(); ++idx)
reply->OnArrayStart(db_count * 2 + 2);
for (size_t idx = 0; idx < dbsizes_.size(); ++idx)
{
if (dbsizes_[idx] > 0)
{
if (dbsizes_[idx] > 0)
{
result += "\r\ndb" + std::to_string(idx) +
":keys=" + std::to_string(dbsizes_[idx]);
}
reply->OnString("db" + std::to_string(idx) + ":keys");
reply->OnInt(dbsizes_[idx]);
}
result += "\r\nstore:disk_keys=" + std::to_string(store_disk_keys_);
}
#endif

result += "\r\n";
reply->OnString(result);
reply->OnString("store:disk_keys");
reply->OnInt(store_disk_keys_);
reply->OnArrayEnd();
}
#endif

void ClusterCommand::OutputResult(OutputHandler *reply) const
{
Expand Down Expand Up @@ -8452,6 +8457,17 @@ ParseMultiCommand(RedisServiceImpl *redis_impl,
DirectRequest(ctx, std::make_unique<InfoCommand>(std::move(cmd)))};
}
#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
case RedisCommandType::KEYSPACE:
{
auto [success, cmd] = ParseKeyspaceCommand(args, output);
if (!success)
{
return {false, DirectRequest{}};
}
return {success,
DirectRequest(
ctx, std::make_unique<KeyspaceCommand>(std::move(cmd)))};
}
case RedisCommandType::COMPACT:
{
auto [success, cmd] = ParseCompactCommand(args, output);
Expand Down Expand Up @@ -10764,6 +10780,18 @@ std::tuple<bool, InfoCommand> ParseInfoCommand(
}

#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
std::tuple<bool, KeyspaceCommand> ParseKeyspaceCommand(
const std::vector<std::string_view> &args, OutputHandler *output)
{
if (args.size() != 1)
{
output->OnError("ERR wrong number of arguments for 'keyspace' command");
return {false, KeyspaceCommand()};
}

return {true, KeyspaceCommand()};
}

std::tuple<bool, CompactCommand> ParseCompactCommand(
const std::vector<std::string_view> &args, OutputHandler *output)
{
Expand Down
39 changes: 38 additions & 1 deletion src/redis_handler.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -650,6 +650,26 @@ brpc::RedisCommandHandlerResult InfoCommandHandler::Run(
}

#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
brpc::RedisCommandHandlerResult KeyspaceCommandHandler::Run(
RedisConnectionContext *ctx,
const std::vector<butil::StringPiece> &args,
brpc::RedisReply *output,
bool /*flush_batched*/)
{
assert(args[0] == "keyspace");

RedisReplier reply(output);
std::vector<std::string_view> cmd_arg_list = Transform(args);

auto [success, cmd] = ParseKeyspaceCommand(cmd_arg_list, &reply);
if (success)
{
redis_impl_->ExecuteCommand(ctx, &cmd, &reply);
}

return brpc::REDIS_CMD_HANDLED;
}

brpc::RedisCommandHandlerResult CompactCommandHandler::Run(
RedisConnectionContext *ctx,
const std::vector<butil::StringPiece> &args,
Expand Down Expand Up @@ -4962,7 +4982,11 @@ brpc::RedisCommandHandlerResult MemoryHandler::Run(

RedisReplier reply(output);
std::vector<std::string_view> cmd_arg_list = Transform(args);
assert(cmd_arg_list.size() > 1);
if (cmd_arg_list.size() < 2)
{
reply.OnError("ERR wrong number of arguments for 'memory' command");
return brpc::REDIS_CMD_HANDLED;
Comment thread
thweetkomputer marked this conversation as resolved.
Resolved
}

if (stringcomp(cmd_arg_list[1], "usage", 1))
{
Expand Down Expand Up @@ -4990,6 +5014,19 @@ brpc::RedisCommandHandlerResult MemoryHandler::Run(
in_tx);
}
}
else if (stringcomp(cmd_arg_list[1], "stats", 1))
{
if (cmd_arg_list.size() != 2)
{
reply.OnError(
"wrong number of arguments for 'memory stats' command");
}
else
{
MemoryStatsCommand cmd;
redis_impl_->ExecuteCommand(ctx, &cmd, &reply);
}
}
else
{
std::string err_msg = "ERR unknown subcommand '";
Expand Down
4 changes: 4 additions & 0 deletions src/redis_service.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1418,6 +1418,10 @@ void RedisServiceImpl::AddHandlers()
AddCommandHandler("info", info_hd.get());

#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD
auto &keyspace_hd =
hd_vec_.emplace_back(std::make_unique<KeyspaceCommandHandler>(this));
AddCommandHandler("keyspace", keyspace_hd.get());

auto &compact_hd =
hd_vec_.emplace_back(std::make_unique<CompactCommandHandler>(this));
AddCommandHandler("compact", compact_hd.get());
Expand Down
Loading