Commit d7e558af authored by Georgios Bitzes's avatar Georgios Bitzes
Browse files

Use separate command type for recovery commands

parent aeb00192
Pipeline #418634 passed with stages
in 31 minutes and 11 seconds
......@@ -126,7 +126,11 @@ struct cmdMapInit {
redis_cmd_map["quarkdb_invalid_command"] = {RedisCommand::QUARKDB_INVALID_COMMAND, CommandType::QUARKDB};
redis_cmd_map["quarkdb_manual_compaction"] = {RedisCommand::QUARKDB_MANUAL_COMPACTION, CommandType::QUARKDB};
redis_cmd_map["quarkdb_level_stats"] = {RedisCommand::QUARKDB_LEVEL_STATS, CommandType::QUARKDB};
redis_cmd_map["recovery_info"] = {RedisCommand::RECOVERY_INFO, CommandType::QUARKDB};
redis_cmd_map["recovery_info"] = {RedisCommand::RECOVERY_INFO, CommandType::RECOVERY};
redis_cmd_map["recovery_set"] = {RedisCommand::RECOVERY_SET, CommandType::RECOVERY};
redis_cmd_map["recovery_get"] = {RedisCommand::RECOVERY_GET, CommandType::RECOVERY};
redis_cmd_map["recovery_del"] = {RedisCommand::RECOVERY_DEL, CommandType::RECOVERY};
}
} cmd_map_init;
......@@ -128,6 +128,10 @@ enum class RedisCommand {
QUARKDB_INVALID_COMMAND, // used in tests
QUARKDB_MANUAL_COMPACTION,
QUARKDB_LEVEL_STATS,
RECOVERY_GET,
RECOVERY_SET,
RECOVERY_DEL,
RECOVERY_INFO
};
......@@ -139,7 +143,8 @@ enum class CommandType {
CONTROL,
RAFT,
QUARKDB,
AUTHENTICATION
AUTHENTICATION,
RECOVERY
};
#define QDB_ALWAYS_INLINE __attribute__((always_inline))
......
......@@ -111,6 +111,10 @@ void Shard::spindown() {
LinkStatus Shard::dispatch(Connection *conn, RedisRequest &req) {
commandMonitor.broadcast(conn->describe(), req);
if(req.getCommandType() == CommandType::RECOVERY) {
return conn->err("recovery commands not allowed, not in recovery mode");
}
switch(req.getCommand()) {
case RedisCommand::MONITOR: {
commandMonitor.addRegistration(conn);
......
......@@ -33,8 +33,14 @@ LinkStatus RecoveryDispatcher::dispatch(Connection *conn, RedisRequest &req) {
}
RedisEncodedResponse RecoveryDispatcher::dispatch(RedisRequest &request) {
if(request.getCommandType() != CommandType::RECOVERY) {
std::string msg = SSTR("unable to dispatch command " << quotes(request[0]) << " - remember we're running in recovery mode, not all operations are available");
qdb_warn(msg);
return Formatter::err(msg);
}
switch(request.getCommand()) {
case RedisCommand::GET: {
case RedisCommand::RECOVERY_GET: {
if(request.size() != 2) return Formatter::errArgs(request[0]);
std::string value;
......@@ -42,11 +48,11 @@ RedisEncodedResponse RecoveryDispatcher::dispatch(RedisRequest &request) {
if(!st.ok()) return Formatter::fromStatus(st);
return Formatter::string(value);
}
case RedisCommand::SET: {
case RedisCommand::RECOVERY_SET: {
if(request.size() != 3) return Formatter::errArgs(request[0]);
return Formatter::fromStatus(editor.set(request[1], request[2]));
}
case RedisCommand::DEL: {
case RedisCommand::RECOVERY_DEL: {
if(request.size() != 2) return Formatter::errArgs(request[0]);
return Formatter::fromStatus(editor.del(request[1]));
}
......@@ -55,9 +61,7 @@ RedisEncodedResponse RecoveryDispatcher::dispatch(RedisRequest &request) {
return Formatter::vector(editor.retrieveMagicValues());
}
default: {
std::string msg = SSTR("unable to dispatch command " << quotes(request[0]) << " - remember we're running in recovery mode, not all operations are available");
qdb_warn(msg);
return Formatter::err(msg);
qdb_throw("should never reach here");
}
}
}
......@@ -28,7 +28,7 @@ RecoveryRunner::RecoveryRunner(const std::string &path, int port)
: editor(path), dispatcher(editor), poller(port, &dispatcher) {
qdb_event("RECOVERY MODE is now active: Issue requests to port " << port << " through redis-cli.");
qdb_info("\nUseful commands: \n"
" GET, SET, DEL:\n"
" RECOVERY_GET, RECOVERY_SET, RECOVERY_DEL:\n"
" Note that these are very different beasts than the traditional GET, SET, DEL offered by QuarkDB.\n"
" These hit directly the rocksdb keys, without a KeyDescriptor or anything else in the middle. \n\n"
" This makes it possible to change low-level details (ie commit-index, last-applied, format, nodes),\n"
......
......@@ -458,6 +458,7 @@ TEST_F(Raft_e2e, test_many_redis_commands) {
futures.emplace_back(tunnel(leaderID)->exec("exists", "hash"));
futures.emplace_back(tunnel(leaderID)->exec("exists", "hash2"));
futures.emplace_back(tunnel(leaderID)->exec("raft_info", "leader"));
futures.emplace_back(tunnel(leaderID)->exec("recovery_get", "test"));
ASSERT_REPLY(futures[0], 1);
ASSERT_REPLY(futures[1], 1);
......@@ -468,6 +469,7 @@ TEST_F(Raft_e2e, test_many_redis_commands) {
ASSERT_REPLY(futures[6], 0);
ASSERT_REPLY(futures[7], 1);
ASSERT_REPLY(futures[8], myself(leaderID).toString() );
ASSERT_REPLY(futures[9], "ERR recovery commands not allowed, not in recovery mode");
futures.clear();
futures.emplace_back(tunnel(leaderID)->exec("hmset", "hmset_test", "f1", "v1", "f2", "v2"));
......
......@@ -95,12 +95,12 @@ TEST(Recovery, RemoveJournalEntriesAndChangeClusterID) {
RecoveryRunner runner("/tmp/quarkdb-recovery-test", 30100);
qclient::QClient qcl("localhost", 30100, {} );
ASSERT_REPLY(qcl.exec("get", KeyConstants::kJournal_ClusterID), "some-cluster-id");
ASSERT_REPLY(qcl.exec("set", KeyConstants::kJournal_ClusterID, "different-cluster-id"), "OK");
ASSERT_REPLY(qcl.exec("set", KeyConstants::kJournal_LogSize, intToBinaryString(2)), "OK");
ASSERT_REPLY(qcl.exec("del", "does-not-exist"), "ERR Invalid argument: key not found, but I inserted a tombstone anyway. Deletion status: OK");
ASSERT_REPLY(qcl.exec("get", SSTR("E" << intToBinaryString(2))), RaftEntry(4, "set", "abc", "cdf").serialize());
ASSERT_REPLY(qcl.exec("del", SSTR("E" << intToBinaryString(2))), "OK");
ASSERT_REPLY(qcl.exec("recovery-get", KeyConstants::kJournal_ClusterID), "some-cluster-id");
ASSERT_REPLY(qcl.exec("recovery-set", KeyConstants::kJournal_ClusterID, "different-cluster-id"), "OK");
ASSERT_REPLY(qcl.exec("recovery-set", KeyConstants::kJournal_LogSize, intToBinaryString(2)), "OK");
ASSERT_REPLY(qcl.exec("recovery-del", "does-not-exist"), "ERR Invalid argument: key not found, but I inserted a tombstone anyway. Deletion status: OK");
ASSERT_REPLY(qcl.exec("recovery-get", SSTR("E" << intToBinaryString(2))), RaftEntry(4, "set", "abc", "cdf").serialize());
ASSERT_REPLY(qcl.exec("recovery-del", SSTR("E" << intToBinaryString(2))), "OK");
std::vector<std::string> rep = {
"RAFT_CURRENT_TERM",
......
Supports Markdown
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