-
Notifications
You must be signed in to change notification settings - Fork 487
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Support slot-based data migration (#430)
A new command CLUSTERX MIGRATE is used for migrate slot data, slot-based migration process mainly includes the following stages: migrating existing data and migrating incremental data. Command format: CLUSTERX MIGRATE $slot $dst_nodeid - $slot is the slot which is to migrate - $dst_nodeid is the node id of destination server in the cluster. We also introduce an internal command CLUSTER IMPORT for importing the migrating slot data into destination server. Migration status are shown into the output of CLUSTER INFO command. After migration slot, you also should use CLUSTERX SETSLOT command to change cluster slot distribution. For more details, please see #412 and #430
- Loading branch information
Showing
41 changed files
with
3,038 additions
and
335 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,207 @@ | ||
| #include "batch_extractor.h" | ||
|
|
||
| #include <rocksdb/write_batch.h> | ||
| #include <glog/logging.h> | ||
|
|
||
| #include "redis_bitmap.h" | ||
| #include "redis_slot.h" | ||
| #include "redis_reply.h" | ||
|
|
||
| void WriteBatchExtractor::LogData(const rocksdb::Slice &blob) { | ||
| log_data_.Decode(blob); | ||
| } | ||
|
|
||
| rocksdb::Status WriteBatchExtractor::PutCF(uint32_t column_family_id, const Slice &key, | ||
| const Slice &value) { | ||
| if (column_family_id == kColumnFamilyIDZSetScore) { | ||
| return rocksdb::Status::OK(); | ||
| } | ||
|
|
||
| std::string ns, user_key, sub_key; | ||
| std::vector<std::string> command_args; | ||
| if (column_family_id == kColumnFamilyIDMetadata) { | ||
| ExtractNamespaceKey(key, &ns, &user_key, is_slotid_encoded_); | ||
| if (slot_ >= 0) { | ||
| if (static_cast<uint16_t>(slot_) != GetSlotNumFromKey(user_key)) return rocksdb::Status::OK(); | ||
| } | ||
| Metadata metadata(kRedisNone); | ||
| metadata.Decode(value.ToString()); | ||
| if (metadata.Type() == kRedisString) { | ||
| command_args = {"SET", user_key, value.ToString().substr(5, value.size() - 5)}; | ||
| resp_commands_[ns].emplace_back(Redis::Command2RESP(command_args)); | ||
| if (metadata.expire > 0) { | ||
| command_args = {"EXPIREAT", user_key, std::to_string(metadata.expire)}; | ||
| resp_commands_[ns].emplace_back(Redis::Command2RESP(command_args)); | ||
| } | ||
| } else if (metadata.expire > 0) { | ||
| auto args = log_data_.GetArguments(); | ||
| if (args->size() > 0) { | ||
| RedisCommand cmd = static_cast<RedisCommand >(std::stoi((*args)[0])); | ||
| if (cmd == kRedisCmdExpire) { | ||
| command_args = {"EXPIREAT", user_key, std::to_string(metadata.expire)}; | ||
| resp_commands_[ns].emplace_back(Redis::Command2RESP(command_args)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return rocksdb::Status::OK(); | ||
| } | ||
|
|
||
| if (column_family_id == kColumnFamilyIDDefault) { | ||
| InternalKey ikey(key, is_slotid_encoded_); | ||
| user_key = ikey.GetKey().ToString(); | ||
| if (slot_ >= 0) { | ||
| if (static_cast<uint16_t>(slot_) != GetSlotNumFromKey(user_key)) return rocksdb::Status::OK(); | ||
| } | ||
| sub_key = ikey.GetSubKey().ToString(); | ||
| ns = ikey.GetNamespace().ToString(); | ||
| switch (log_data_.GetRedisType()) { | ||
| case kRedisHash:command_args = {"HSET", user_key, sub_key, value.ToString()}; | ||
| break; | ||
| case kRedisList: { | ||
| auto args = log_data_.GetArguments(); | ||
| if (args->size() < 1) { | ||
| LOG(ERROR) << "Fail to parse write_batch in putcf type list : args error ,should at least contain cmd"; | ||
| return rocksdb::Status::OK(); | ||
| } | ||
| RedisCommand cmd = static_cast<RedisCommand >(std::stoi((*args)[0])); | ||
| switch (cmd) { | ||
| case kRedisCmdLSet: | ||
| if (args->size() < 2) { | ||
| LOG(ERROR) << "Fail to parse write_batch in putcf cmd lset : args error ,should contain lset index"; | ||
| return rocksdb::Status::OK(); | ||
| } | ||
| command_args = {"LSET", user_key, (*args)[1], value.ToString()}; | ||
| break; | ||
| case kRedisCmdLInsert: | ||
| if (first_seen_) { | ||
| if (args->size() < 4) { | ||
| LOG(ERROR) | ||
| << "Fail to parse write_batch in putcf cmd linsert : args error, should contain before pivot value"; | ||
| return rocksdb::Status::OK(); | ||
| } | ||
| command_args = {"LINSERT", user_key, (*args)[1] == "1" ? "before" : "after", (*args)[2], (*args)[3]}; | ||
| first_seen_ = false; | ||
| } | ||
| break; | ||
| case kRedisCmdLRem: | ||
| // lrem will be parsed in deletecf, so ignore this putcf | ||
| break; | ||
| default:command_args = {cmd == kRedisCmdLPush ? "LPUSH" : "RPUSH", user_key, value.ToString()}; | ||
| } | ||
| break; | ||
| } | ||
| case kRedisSet:command_args = {"SADD", user_key, sub_key}; | ||
| break; | ||
| case kRedisZSet: { | ||
| double score = DecodeDouble(value.data()); | ||
| command_args = {"ZADD", user_key, std::to_string(score), sub_key}; | ||
| break; | ||
| } | ||
| case kRedisBitmap: { | ||
| auto args = log_data_.GetArguments(); | ||
| if (args->size() < 1) { | ||
| LOG(ERROR) << "Fail to parse write_batch in putcf cmd setbit : args error ,should contain setbit offset"; | ||
| return rocksdb::Status::OK(); | ||
| } | ||
| bool bit_value = Redis::Bitmap::GetBitFromValueAndOffset(value.ToString(), std::stoi((*args)[0])); | ||
| command_args = {"SETBIT", user_key, (*args)[0], bit_value ? "1" : "0"}; | ||
| break; | ||
| } | ||
| case kRedisSortedint: { | ||
| if (!to_redis_) { | ||
| command_args = {"SIADD", user_key, std::to_string(DecodeFixed64(sub_key.data()))}; | ||
| } | ||
| break; | ||
| } | ||
| default: break; | ||
| } | ||
| } | ||
|
|
||
| if (!command_args.empty()) { | ||
| resp_commands_[ns].emplace_back(Redis::Command2RESP(command_args)); | ||
| } | ||
| return rocksdb::Status::OK(); | ||
| } | ||
|
|
||
| rocksdb::Status WriteBatchExtractor::DeleteCF(uint32_t column_family_id, const Slice &key) { | ||
| if (column_family_id == kColumnFamilyIDZSetScore) { | ||
| return rocksdb::Status::OK(); | ||
| } | ||
|
|
||
| std::string ns, user_key, sub_key; | ||
| std::vector<std::string> command_args; | ||
| if (column_family_id == kColumnFamilyIDMetadata) { | ||
| ExtractNamespaceKey(key, &ns, &user_key, is_slotid_encoded_); | ||
| if (slot_ >= 0) { | ||
| if (static_cast<uint16_t>(slot_) != GetSlotNumFromKey(user_key)) return rocksdb::Status::OK(); | ||
| } | ||
| command_args = {"DEL", user_key}; | ||
| } else if (column_family_id == kColumnFamilyIDDefault) { | ||
| InternalKey ikey(key, is_slotid_encoded_); | ||
| user_key = ikey.GetKey().ToString(); | ||
| if (slot_ >= 0) { | ||
| if (static_cast<uint16_t>(slot_) != GetSlotNumFromKey(user_key)) return rocksdb::Status::OK(); | ||
| } | ||
| sub_key = ikey.GetSubKey().ToString(); | ||
| ns = ikey.GetNamespace().ToString(); | ||
| switch (log_data_.GetRedisType()) { | ||
| case kRedisHash: command_args = {"HDEL", user_key, sub_key}; | ||
| break; | ||
| case kRedisSet: command_args = {"SREM", user_key, sub_key}; | ||
| break; | ||
| case kRedisZSet: command_args = {"ZREM", user_key, sub_key}; | ||
| break; | ||
| case kRedisList: { | ||
| auto args = log_data_.GetArguments(); | ||
| if (args->size() < 1) { | ||
| LOG(ERROR) << "Fail to parse write_batch in DeleteCF type list : args error ,should contain cmd"; | ||
| return rocksdb::Status::OK(); | ||
| } | ||
| RedisCommand cmd = static_cast<RedisCommand >(std::stoi((*args)[0])); | ||
| switch (cmd) { | ||
| case kRedisCmdLTrim: | ||
| if (first_seen_) { | ||
| if (args->size() < 3) { | ||
| LOG(ERROR) << "Fail to parse write_batch in DeleteCF cmd ltrim : args error ,should contain start,stop"; | ||
| return rocksdb::Status::OK(); | ||
| } | ||
| command_args = {"LTRIM", user_key, (*args)[1], (*args)[2]}; | ||
| first_seen_ = false; | ||
| } | ||
| break; | ||
| case kRedisCmdLRem: | ||
| if (first_seen_) { | ||
| if (args->size() < 3) { | ||
| LOG(ERROR) << "Fail to parse write_batch in DeleteCF cmd lrem : args error ,should contain count,value"; | ||
| return rocksdb::Status::OK(); | ||
| } | ||
| command_args = {"LREM", user_key, (*args)[1], (*args)[2]}; | ||
| first_seen_ = false; | ||
| } | ||
| break; | ||
| default:command_args = {cmd == kRedisCmdLPop ? "LPOP" : "RPOP", user_key}; | ||
| } | ||
| break; | ||
| } | ||
| case kRedisSortedint: { | ||
| if (!to_redis_) { | ||
| command_args = {"SIREM", user_key, std::to_string(DecodeFixed64(sub_key.data()))}; | ||
| } | ||
| break; | ||
| } | ||
| default: break; | ||
| } | ||
| } | ||
|
|
||
| if (!command_args.empty()) { | ||
| resp_commands_[ns].emplace_back(Redis::Command2RESP(command_args)); | ||
| } | ||
| return rocksdb::Status::OK(); | ||
| } | ||
|
|
||
| rocksdb::Status WriteBatchExtractor::DeleteRangeCF(uint32_t column_family_id, | ||
| const Slice& begin_key, const Slice& end_key) { | ||
| // Do nothing about DeleteRange operations | ||
| return rocksdb::Status::OK(); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| #pragma once | ||
| #include <string> | ||
| #include <vector> | ||
| #include <map> | ||
|
|
||
| #include "redis_db.h" | ||
| #include "status.h" | ||
| #include "storage.h" | ||
| #include "redis_metadata.h" | ||
|
|
||
|
|
||
| // An extractor to extract update from raw writebatch | ||
| class WriteBatchExtractor : public rocksdb::WriteBatch::Handler { | ||
| public: | ||
| explicit WriteBatchExtractor(bool is_slotid_encoded, int16_t slot = -1, bool to_redis = false) | ||
| : is_slotid_encoded_(is_slotid_encoded), slot_(slot), to_redis_(to_redis) {} | ||
| void LogData(const rocksdb::Slice &blob) override; | ||
| rocksdb::Status PutCF(uint32_t column_family_id, const Slice &key, | ||
| const Slice &value) override; | ||
|
|
||
| rocksdb::Status DeleteCF(uint32_t column_family_id, const Slice &key) override; | ||
| rocksdb::Status DeleteRangeCF(uint32_t column_family_id, | ||
| const Slice& begin_key, const Slice& end_key) override; | ||
| std::map<std::string, std::vector<std::string>> *GetRESPCommands() { return &resp_commands_; } | ||
| private: | ||
| std::map<std::string, std::vector<std::string>> resp_commands_; | ||
| Redis::WriteBatchLogData log_data_; | ||
| bool first_seen_ = true; | ||
| bool is_slotid_encoded_ = false; | ||
| int slot_; | ||
| bool to_redis_; | ||
| }; |
Oops, something went wrong.