This is an automated email from the ASF dual-hosted git repository.
PragmaTwice pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new bfe3cbdf8 refactor(cluster): remove the redis-command migration type
(#3544)
bfe3cbdf8 is described below
commit bfe3cbdf86e19d750673780ae3a82ecd7d33b65d
Author: hulk <[email protected]>
AuthorDate: Mon Jul 6 12:48:43 2026 +0800
refactor(cluster): remove the redis-command migration type (#3544)
This PR removes the `redis-command` slot migration type, leaving
raw-key-value
as the only migration path. The raw-key-value type has been introduced
for a long time, and it outperforms the redis-command type in both
resource consumption and migration speed, so we can remove it to
simplify the slot migration code. Also, the redis command migration
didn't work well after introducing more and more new data types.
To be noticed, there is no fallback anymore, the migration now fails
with the error `destination node doesn't support the APPLYBATCH command`
if the destination
node doesn't support the APPLYBATCH command, instead of silently falling
back to the redis-command type.
Assistant By Claude Fable 5
---
kvrocks.conf | 27 -
src/cluster/slot_migrate.cc | 711 +--------------------
src/cluster/slot_migrate.h | 67 +-
src/config/config.cc | 27 +-
src/config/config.h | 4 -
tests/gocase/integration/cluster/cluster_test.go | 8 +-
.../integration/slotmigrate/slotmigrate_test.go | 156 ++---
7 files changed, 97 insertions(+), 903 deletions(-)
diff --git a/kvrocks.conf b/kvrocks.conf
index f1541c496..b2ab11fbd 100644
--- a/kvrocks.conf
+++ b/kvrocks.conf
@@ -719,33 +719,6 @@ compaction-checker-cron * 0-7 * * *
# rename-command KEYS ""
################################ MIGRATE #####################################
-# Slot migration supports two ways:
-# - redis-command: Migrate data by redis serialization protocol(RESP).
-# - raw-key-value: Migrate the raw key value data of the storage engine
directly.
-# This way eliminates the overhead of converting to the redis
-# command, reduces resource consumption, improves migration
-# efficiency, and can implement a finer rate limit.
-#
-# Default: raw-key-value
-migrate-type raw-key-value
-
-# If the network bandwidth is completely consumed by the migration task,
-# it will affect the availability of kvrocks. To avoid this situation,
-# migrate-speed is adopted to limit the migrating speed.
-# Migrating speed is limited by controlling the duration between sending data,
-# the duration is calculated by: 1000000 * migrate-pipeline-size /
migrate-speed (us).
-# Value: [0,INT_MAX], 0 means no limit
-#
-# Default: 4096
-migrate-speed 4096
-
-# In order to reduce data transmission times and improve the efficiency of
data migration,
-# pipeline is adopted to send multiple data at once. Pipeline size can be set
by this option.
-# Value: [1, INT_MAX], it can't be 0
-#
-# Default: 16
-migrate-pipeline-size 16
-
# In order to reduce the write forbidden time during migrating slot, we will
migrate the incremental
# data several times to reduce the amount of incremental data. Until the
quantity of incremental
# data is reduced to a certain threshold, slot will be forbidden write. The
threshold is set by
diff --git a/src/cluster/slot_migrate.cc b/src/cluster/slot_migrate.cc
index 3edf8a3e4..02e0534c2 100644
--- a/src/cluster/slot_migrate.cc
+++ b/src/cluster/slot_migrate.cc
@@ -23,33 +23,21 @@
#include <memory>
#include <utility>
-#include "db_util.h"
#include "event_util.h"
#include "fmt/format.h"
#include "io_util.h"
-#include "storage/batch_extractor.h"
#include "storage/iterator.h"
#include "storage/redis_metadata.h"
#include "sync_migrate_context.h"
#include "thread_util.h"
#include "time_util.h"
-#include "types/redis_stream_base.h"
-constexpr std::string_view errFailedToSendCommands = "failed to send commands
to restore a key";
constexpr std::string_view errMigrationTaskCanceled = "key migration stopped
due to a task cancellation";
constexpr std::string_view errFailedToSetImportStatus = "failed to set import
status on destination node";
-constexpr std::string_view errUnsupportedMigrationType = "unsupported
migration type";
-
-static std::map<RedisType, std::string> type_to_cmd = {
- {kRedisString, "set"}, {kRedisList, "rpush"}, {kRedisHash, "hmset"},
{kRedisSet, "sadd"},
- {kRedisZSet, "zadd"}, {kRedisBitmap, "setbit"}, {kRedisSortedint,
"siadd"}, {kRedisStream, "xadd"},
-};
SlotMigrator::SlotMigrator(Server *srv)
: Database(srv->storage, kDefaultNamespace),
srv_(srv),
- max_migration_speed_(srv->GetConfig()->migrate_speed),
- max_pipeline_size_(srv->GetConfig()->pipeline_size),
seq_gap_limit_(srv->GetConfig()->sequence_gap),
migrate_batch_bytes_per_sec_(srv->GetConfig()->migrate_batch_rate_limit_mb *
MiB),
migrate_batch_size_bytes_(srv->GetConfig()->migrate_batch_size_kb * KiB)
{
@@ -91,18 +79,7 @@ Status SlotMigrator::PerformSlotRangeMigration(const
std::string &node_id, std::
}
migration_state_ = MigrationState::kStarted;
- auto speed = srv_->GetConfig()->migrate_speed;
auto seq_gap = srv_->GetConfig()->sequence_gap;
- auto pipeline_size = srv_->GetConfig()->pipeline_size;
-
- if (speed <= 0) {
- speed = 0;
- }
-
- if (pipeline_size <= 0) {
- pipeline_size = kDefaultMaxPipelineSize;
- }
-
if (seq_gap <= 0) {
seq_gap = kDefaultSequenceGapLimit;
}
@@ -116,7 +93,7 @@ Status SlotMigrator::PerformSlotRangeMigration(const
std::string &node_id, std::
dst_node_ = node_id;
// Create migration job
- auto job = std::make_unique<SlotMigrationJob>(slot_range, dst_ip, dst_port,
speed, pipeline_size, seq_gap);
+ auto job = std::make_unique<SlotMigrationJob>(slot_range, dst_ip, dst_port,
seq_gap);
{
std::lock_guard<std::mutex> guard(job_mutex_);
migration_job_ = std::move(job);
@@ -158,14 +135,11 @@ void SlotMigrator::loop() {
clean();
return;
}
- INFO("[migrate] Migrating slot(s): {}, dst_ip: {}, dst_port: {},
max_speed: {}, max_pipeline_size: {}",
- migration_job_->slot_range.String(), migration_job_->dst_ip,
migration_job_->dst_port,
- migration_job_->max_speed, migration_job_->max_pipeline_size);
+ INFO("[migrate] Migrating slot(s): {}, dst_ip: {}, dst_port: {}",
migration_job_->slot_range.String(),
+ migration_job_->dst_ip, migration_job_->dst_port);
dst_ip_ = migration_job_->dst_ip;
dst_port_ = migration_job_->dst_port;
- max_migration_speed_ = migration_job_->max_speed;
- max_pipeline_size_ = migration_job_->max_pipeline_size;
seq_gap_limit_ = migration_job_->seq_gap_limit;
runMigrationProcess();
@@ -264,7 +238,6 @@ Status SlotMigrator::startMigration() {
}
wal_begin_seq_ = slot_snapshot_->GetSequenceNumber();
- last_send_time_ = 0;
// Connect to the destination node
auto result = util::SockConnect(dst_ip_, dst_port_);
@@ -289,138 +262,17 @@ Status SlotMigrator::startMigration() {
return s.Prefixed(errFailedToSetImportStatus);
}
- migration_type_ = srv_->GetConfig()->migrate_type;
-
- // If the APPLYBATCH command is not supported on the destination,
- // we will fall back to the redis-command migration type.
- if (migration_type_ == MigrationType::kRawKeyValue) {
- bool supported = GET_OR_RET(supportedApplyBatchCommandOnDstNode(*dst_fd_));
- if (!supported) {
- INFO("APPLYBATCH command is not supported, use redis command for
migration");
- migration_type_ = MigrationType::kRedisCommand;
- }
+ // The migration relies on the APPLYBATCH command on the destination node
+ // to apply the raw key-value batches, so abort the migration if it's
unavailable.
+ bool supported = GET_OR_RET(supportedApplyBatchCommandOnDstNode(*dst_fd_));
+ if (!supported) {
+ return {Status::NotOK, "destination node doesn't support the APPLYBATCH
command"};
}
INFO("[migrate] Start migrating slot(s) {}, connect destination fd {}",
slot_range_.load().String(), *dst_fd_);
return Status::OK();
}
-Status SlotMigrator::sendSnapshot() {
- if (migration_type_ == MigrationType::kRedisCommand) {
- return sendSnapshotByCmd();
- } else if (migration_type_ == MigrationType::kRawKeyValue) {
- return sendSnapshotByRawKV();
- }
- return {Status::NotOK, std::string(errUnsupportedMigrationType)};
-}
-
-Status SlotMigrator::syncWAL() {
- if (migration_type_ == MigrationType::kRedisCommand) {
- return syncWALByCmd();
- } else if (migration_type_ == MigrationType::kRawKeyValue) {
- return syncWALByRawKV();
- }
- return {Status::NotOK, std::string(errUnsupportedMigrationType)};
-}
-
-Status SlotMigrator::sendSnapshotByCmd() {
- uint64_t migrated_key_cnt = 0;
- uint64_t expired_key_cnt = 0;
- uint64_t empty_key_cnt = 0;
- std::string restore_cmds;
- SlotRange slot_range = slot_range_;
- INFO("[migrate] Start migrating snapshot of slot(s): {}",
slot_range.String());
-
- // Construct key prefix to iterate the keys belong to the target slot
- std::string prefix = ComposeSlotKeyPrefix(namespace_, slot_range.start);
- INFO("[migrate] Iterate keys of slot(s), key's prefix: {}", prefix);
-
- std::string upper_bound = ComposeSlotKeyUpperBound(namespace_,
slot_range.end);
- rocksdb::ReadOptions read_options = storage_->DefaultScanOptions();
- read_options.snapshot = slot_snapshot_;
- Slice prefix_slice(prefix);
- Slice upper_bound_slice(upper_bound);
- read_options.iterate_lower_bound = &prefix_slice;
- read_options.iterate_upper_bound = &upper_bound_slice;
- rocksdb::ColumnFamilyHandle *cf_handle =
storage_->GetCFHandle(ColumnFamilyID::Metadata);
- auto iter =
util::UniqueIterator(storage_->GetDB()->NewIterator(read_options, cf_handle));
-
- // Seek to the beginning of keys start with 'prefix' and iterate all these
keys
- int current_slot = slot_range.start;
- for (iter->Seek(prefix); iter->Valid(); iter->Next()) {
- // The migrating task has to be stopped, if server role is changed from
master to slave
- // or flush command (flushdb or flushall) is executed
- if (stop_migration_) {
- return {Status::NotOK, std::string(errMigrationTaskCanceled)};
- }
-
- // Iteration is out of range
- current_slot = ExtractSlotId(iter->key());
- if (!slot_range.Contains(current_slot)) {
- break;
- }
-
- // Get user key
- auto [_, user_key] = ExtractNamespaceKey(iter->key(),
/*slot_id_encoded=*/true);
-
- // Add key's constructed commands to restore_cmds, send pipeline or not
according to task's max_pipeline_size
- auto result = migrateOneKey(user_key, iter->value(), &restore_cmds);
- if (!result.IsOK()) {
- return {Status::NotOK, fmt::format("failed to migrate a key {}: {}",
user_key, result.Msg())};
- }
-
- if (*result == KeyMigrationResult::kMigrated) {
- INFO("[migrate] The key {} successfully migrated", user_key);
- migrated_key_cnt++;
- } else if (*result == KeyMigrationResult::kExpired) {
- INFO("[migrate] The key {} is expired", user_key);
- expired_key_cnt++;
- } else if (*result == KeyMigrationResult::kUnderlyingStructEmpty) {
- INFO("[migrate] The key {} has no elements", user_key);
- empty_key_cnt++;
- } else {
- ERROR("[migrate] Migrated a key {} with unexpected result: {}",
user_key, static_cast<int>(*result));
- return {Status::NotOK};
- }
- }
-
- if (auto s = iter->status(); !s.ok()) {
- auto err_str = s.ToString();
- ERROR("[migrate] Failed to iterate keys of slot {}: {}", current_slot,
err_str);
- return {Status::NotOK, fmt::format("failed to iterate keys of slot {}:
{}", current_slot, err_str)};
- }
-
- // It's necessary to send commands that are still in the pipeline since the
final pipeline may not be sent
- // while iterating keys because its size could be less than
max_pipeline_size_
- auto s = sendCmdsPipelineIfNeed(&restore_cmds, true);
- if (!s.IsOK()) {
- return s.Prefixed(errFailedToSendCommands);
- }
- INFO(
- "[migrate] Succeed to migrate slot(s) snapshot, slot(s): {}, Migrated
keys: {}, Expired keys: {}, Empty keys: {}",
- slot_range.String(), migrated_key_cnt, expired_key_cnt, empty_key_cnt);
-
- return Status::OK();
-}
-
-Status SlotMigrator::syncWALByCmd() {
- // Send incremental data from WAL circularly until new increment less than a
certain amount
- auto s = syncWalBeforeForbiddingSlot();
- if (!s.IsOK()) {
- return s.Prefixed("failed to sync WAL before forbidding a slot");
- }
-
- setForbiddenSlotRange(slot_range_);
-
- // Send last incremental data
- s = syncWalAfterForbiddingSlot();
- if (!s.IsOK()) {
- return s.Prefixed("failed to sync WAL after forbidding a slot");
- }
-
- return Status::OK();
-}
-
Status SlotMigrator::finishSuccessfulMigration() {
if (stop_migration_) {
return {Status::NotOK, std::string(errMigrationTaskCanceled)};
@@ -466,7 +318,6 @@ void SlotMigrator::clean() {
}
current_stage_ = SlotMigrationStage::kNone;
- current_pipeline_size_ = 0;
wal_begin_seq_ = 0;
std::lock_guard<std::mutex> guard(job_mutex_);
migration_job_.reset();
@@ -537,32 +388,6 @@ StatusOr<bool>
SlotMigrator::supportedApplyBatchCommandOnDstNode(int sock_fd) {
Status SlotMigrator::checkSingleResponse(int sock_fd) { return
checkMultipleResponses(sock_fd, 1); }
-// Commands | Response | Instance
-// ++++++++++++++++++++++++++++++++++++++++
-// set Redis::Integer :1\r\n
-// hset Redis::SimpleString +OK\r\n
-// sadd Redis::Integer
-// zadd Redis::Integer
-// siadd Redis::Integer
-// setbit Redis::Integer
-// expire Redis::Integer
-// lpush Redis::Integer
-// rpush Redis::Integer
-// ltrim Redis::SimpleString -Err\r\n
-// linsert Redis::Integer
-// lset Redis::SimpleString
-// hdel Redis::Integer
-// srem Redis::Integer
-// zrem Redis::Integer
-// lpop Redis::NilString $-1\r\n
-// or Redis::BulkString $1\r\n1\r\n
-// rpop Redis::NilString
-// or Redis::BulkString
-// lrem Redis::Integer
-// sirem Redis::Integer
-// del Redis::Integer
-// xadd Redis::BulkString
-// bitfield Redis::Array *1\r\n:0
Status SlotMigrator::checkMultipleResponses(int sock_fd, int total) {
if (sock_fd < 0 || total <= 0) {
return {Status::NotOK, fmt::format("invalid arguments: sock_fd={},
count={}", sock_fd, total)};
@@ -667,370 +492,6 @@ Status SlotMigrator::checkMultipleResponses(int sock_fd,
int total) {
}
}
-StatusOr<KeyMigrationResult> SlotMigrator::migrateOneKey(const rocksdb::Slice
&key,
- const rocksdb::Slice
&encoded_metadata,
- std::string
*restore_cmds) {
- std::string bytes = encoded_metadata.ToString();
- Metadata metadata(kRedisNone, false);
- if (auto s = metadata.Decode(bytes); !s.ok()) {
- return {Status::NotOK, s.ToString()};
- }
-
- if (!metadata.IsEmptyableType() && metadata.size == 0) {
- return KeyMigrationResult::kUnderlyingStructEmpty;
- }
-
- if (metadata.Expired()) {
- return KeyMigrationResult::kExpired;
- }
-
- // Construct command according to type of the key
- switch (metadata.Type()) {
- case kRedisString:
- case kRedisJson: {
- auto s = migrateSimpleKey(key, metadata, bytes, restore_cmds);
- if (!s.IsOK()) {
- return s.Prefixed("failed to migrate simple key");
- }
- break;
- }
- case kRedisList:
- case kRedisZSet:
- case kRedisBitmap:
- case kRedisHash:
- case kRedisSet:
- case kRedisSortedint: {
- auto s = migrateComplexKey(key, metadata, restore_cmds);
- if (!s.IsOK()) {
- return s.Prefixed("failed to migrate complex key");
- }
- break;
- }
- case kRedisStream: {
- StreamMetadata stream_md(false);
- if (auto s = stream_md.Decode(bytes); !s.ok()) {
- return {Status::NotOK, s.ToString()};
- }
-
- auto s = migrateStream(key, stream_md, restore_cmds);
- if (!s.IsOK()) {
- return s.Prefixed("failed to migrate stream key");
- }
- break;
- }
- case kRedisHyperLogLog: {
- // HyperLogLog migration by cmd is not supported,
- // since it's hard to restore the same key structure for HyperLogLog
- // commands.
- break;
- }
- default:
- break;
- }
-
- return KeyMigrationResult::kMigrated;
-}
-
-Status SlotMigrator::migrateSimpleKey(const rocksdb::Slice &key, const
Metadata &metadata, const std::string &bytes,
- std::string *restore_cmds) {
- if (metadata.Type() == kRedisString) {
- std::vector<std::string> command = {"SET", key.ToString(),
bytes.substr(Metadata::GetOffsetAfterExpire(bytes[0]))};
- if (metadata.expire > 0) {
- command.emplace_back("PXAT");
- command.emplace_back(std::to_string(metadata.expire));
- }
- *restore_cmds += redis::ArrayOfBulkStrings(command);
- current_pipeline_size_++;
- } else if (metadata.Type() == kRedisJson) {
- // kRedisJson
- JsonValue json_value;
- if (auto s = redis::Json::FromRawString(bytes, &json_value); !s.ok()) {
- return {Status::NotOK, s.ToString()};
- }
- auto json_bytes = GET_OR_RET(json_value.Dump());
- std::vector<std::string> command = {"JSON.SET", key.ToString(), "$",
std::move(json_bytes)};
- *restore_cmds += redis::ArrayOfBulkStrings(command);
- current_pipeline_size_++;
-
- if (metadata.expire > 0) {
- *restore_cmds += redis::ArrayOfBulkStrings({"PEXPIREAT", key.ToString(),
std::to_string(metadata.expire)});
- current_pipeline_size_++;
- }
- } else {
- return {Status::NotOK, "unsupported simple key type"};
- }
-
- // Check whether pipeline needs to be sent
- // TODO(chrisZMF): Resend data if failed to send data
- auto s = sendCmdsPipelineIfNeed(restore_cmds, false);
- if (!s.IsOK()) {
- return s.Prefixed(errFailedToSendCommands);
- }
-
- return Status::OK();
-}
-
-Status SlotMigrator::migrateComplexKey(const rocksdb::Slice &key, const
Metadata &metadata, std::string *restore_cmds) {
- std::string cmd;
- {
- auto iter = type_to_cmd.find(metadata.Type());
- if (iter != type_to_cmd.end()) {
- cmd = iter->second;
- } else {
- if (metadata.Type() > RedisTypeNames.size()) {
- return {Status::NotOK, "unknown key type: " +
std::to_string(metadata.Type())};
- }
- return {Status::NotOK, fmt::format("unsupported complex key type: {}",
metadata.TypeName())};
- }
- }
-
- std::vector<std::string> user_cmd = {cmd, key.ToString()};
- // Construct key prefix to iterate values of the complex type user key
- std::string slot_key = AppendNamespacePrefix(key);
- std::string prefix_subkey = InternalKey(slot_key, "", metadata.version,
true).Encode();
- rocksdb::ReadOptions read_options = storage_->DefaultScanOptions();
- read_options.snapshot = slot_snapshot_;
- Slice prefix_slice(prefix_subkey);
- read_options.iterate_lower_bound = &prefix_slice;
- // Should use th raw db iterator to avoid reading uncommitted writes in
transaction mode
- auto iter =
util::UniqueIterator(storage_->GetDB()->NewIterator(read_options));
-
- int item_count = 0;
-
- for (iter->Seek(prefix_subkey); iter->Valid(); iter->Next()) {
- if (stop_migration_) {
- return {Status::NotOK, std::string(errMigrationTaskCanceled)};
- }
-
- if (!iter->key().starts_with(prefix_subkey)) {
- break;
- }
-
- // Parse values of the complex key
- // InternalKey is adopted to get complex key's value from the formatted
key return by iterator of rocksdb
- InternalKey inkey(iter->key(), true);
- switch (metadata.Type()) {
- case kRedisSet: {
- user_cmd.emplace_back(inkey.GetSubKey().ToString());
- break;
- }
- case kRedisSortedint: {
- auto id = DecodeFixed64(inkey.GetSubKey().ToString().data());
- user_cmd.emplace_back(std::to_string(id));
- break;
- }
- case kRedisZSet: {
- auto score = DecodeDouble(iter->value().ToString().data());
- user_cmd.emplace_back(util::Float2String(score));
- user_cmd.emplace_back(inkey.GetSubKey().ToString());
- break;
- }
- case kRedisBitmap: {
- auto s = migrateBitmapKey(inkey, &iter, &user_cmd, restore_cmds);
- if (!s.IsOK()) {
- return s.Prefixed("failed to migrate bitmap key");
- }
- break;
- }
- case kRedisHash: {
- user_cmd.emplace_back(inkey.GetSubKey().ToString());
- user_cmd.emplace_back(iter->value().ToString());
- break;
- }
- case kRedisList: {
- user_cmd.emplace_back(iter->value().ToString());
- break;
- }
- case kRedisHyperLogLog: {
- break;
- }
- default:
- break;
- }
-
- // Check item count
- // Exclude bitmap because it does not have hmset-like command
- if (metadata.Type() != kRedisBitmap) {
- item_count++;
- if (item_count >= kMaxItemsInCommand) {
- *restore_cmds += redis::ArrayOfBulkStrings(user_cmd);
- current_pipeline_size_++;
- item_count = 0;
- // Have to clear saved items
- user_cmd.erase(user_cmd.begin() + 2, user_cmd.end());
-
- // Send commands if the pipeline contains enough of them
- auto s = sendCmdsPipelineIfNeed(restore_cmds, false);
- if (!s.IsOK()) {
- return s.Prefixed(errFailedToSendCommands);
- }
- }
- }
- }
-
- if (auto s = iter->status(); !s.ok()) {
- return {Status::NotOK,
- fmt::format("failed to iterate values of the complex key {}: {}",
key.ToString(), s.ToString())};
- }
-
- // Have to check the item count of the last command list
- if (item_count % kMaxItemsInCommand != 0) {
- *restore_cmds += redis::ArrayOfBulkStrings(user_cmd);
- current_pipeline_size_++;
- }
-
- // Add TTL for complex key
- if (metadata.expire > 0) {
- *restore_cmds += redis::ArrayOfBulkStrings({"PEXPIREAT", key.ToString(),
std::to_string(metadata.expire)});
- current_pipeline_size_++;
- }
-
- // Send commands if the pipeline contains enough of them
- auto s = sendCmdsPipelineIfNeed(restore_cmds, false);
- if (!s.IsOK()) {
- return s.Prefixed(errFailedToSendCommands);
- }
-
- return Status::OK();
-}
-
-Status SlotMigrator::migrateStream(const Slice &key, const StreamMetadata
&metadata, std::string *restore_cmds) {
- rocksdb::ReadOptions read_options = storage_->DefaultScanOptions();
- read_options.snapshot = slot_snapshot_;
- std::string ns_key = AppendNamespacePrefix(key);
- // Construct key prefix to iterate values of the stream
- std::string prefix_key = InternalKey(ns_key, "", metadata.version,
true).Encode();
- rocksdb::Slice prefix_key_slice(prefix_key);
- read_options.iterate_lower_bound = &prefix_key_slice;
-
- // Should use th raw db iterator to avoid reading uncommitted writes in
transaction mode
- auto iter =
- util::UniqueIterator(storage_->GetDB()->NewIterator(read_options,
storage_->GetCFHandle(ColumnFamilyID::Stream)));
-
- std::vector<std::string> user_cmd = {type_to_cmd[metadata.Type()],
key.ToString()};
-
- for (iter->Seek(prefix_key); iter->Valid(); iter->Next()) {
- if (stop_migration_) {
- return {Status::NotOK, std::string(errMigrationTaskCanceled)};
- }
-
- if (!iter->key().starts_with(prefix_key)) {
- break;
- }
-
- auto s = WriteBatchExtractor::ExtractStreamAddCommand(true, iter->key(),
iter->value(), &user_cmd);
- if (!s.IsOK()) {
- return s;
- }
- *restore_cmds += redis::ArrayOfBulkStrings(user_cmd);
- current_pipeline_size_++;
-
- user_cmd.erase(user_cmd.begin() + 2, user_cmd.end());
-
- s = sendCmdsPipelineIfNeed(restore_cmds, false);
- if (!s.IsOK()) {
- return s.Prefixed(errFailedToSendCommands);
- }
- }
-
- if (auto s = iter->status(); !s.ok()) {
- return {Status::NotOK,
- fmt::format("failed to iterate values of the stream key {}: {}",
key.ToString(), s.ToString())};
- }
-
- // commands like XTRIM and XDEL affect stream's metadata, but we use only
XADD for a slot migration
- // XSETID is used to adjust stream's info on the destination node according
to the current values on the source
- *restore_cmds += redis::ArrayOfBulkStrings({"XSETID", key.ToString(),
metadata.last_generated_id.ToString(),
- "ENTRIESADDED",
std::to_string(metadata.entries_added), "MAXDELETEDID",
-
metadata.max_deleted_entry_id.ToString()});
- current_pipeline_size_++;
-
- // Add TTL
- if (metadata.expire > 0) {
- *restore_cmds += redis::ArrayOfBulkStrings({"PEXPIREAT", key.ToString(),
std::to_string(metadata.expire)});
- current_pipeline_size_++;
- }
-
- auto s = sendCmdsPipelineIfNeed(restore_cmds, false);
- if (!s.IsOK()) {
- return s.Prefixed(errFailedToSendCommands);
- }
-
- return Status::OK();
-}
-
-Status SlotMigrator::migrateBitmapKey(const InternalKey &inkey,
std::unique_ptr<rocksdb::Iterator> *iter,
- std::vector<std::string> *user_cmd,
std::string *restore_cmds) {
- std::string index_str = inkey.GetSubKey().ToString();
- std::string fragment = (*iter)->value().ToString();
- auto parse_result = ParseInt<int>(index_str, 10);
- if (!parse_result) {
- return {Status::RedisParseErr, "index is not a valid integer"};
- }
-
- uint32_t index = *parse_result;
-
- // Bitmap does not have hmset-like command
- // TODO(chrisZMF): Use hmset-like command for efficiency
- for (int byte_idx = 0; byte_idx < static_cast<int>(fragment.size());
byte_idx++) {
- if (fragment[byte_idx] & 0xff) {
- for (int bit_idx = 0; bit_idx < 8; bit_idx++) {
- if (fragment[byte_idx] & (1 << bit_idx)) {
- uint32_t offset = (index * 8) + (byte_idx * 8) + bit_idx;
- user_cmd->emplace_back(std::to_string(offset));
- user_cmd->emplace_back("1");
- *restore_cmds += redis::ArrayOfBulkStrings(*user_cmd);
- current_pipeline_size_++;
- user_cmd->erase(user_cmd->begin() + 2, user_cmd->end());
- }
- }
-
- auto s = sendCmdsPipelineIfNeed(restore_cmds, false);
- if (!s.IsOK()) {
- return s.Prefixed(errFailedToSendCommands);
- }
- }
- }
-
- return Status::OK();
-}
-
-Status SlotMigrator::sendCmdsPipelineIfNeed(std::string *commands, bool need) {
- if (stop_migration_) {
- return {Status::NotOK, std::string(errMigrationTaskCanceled)};
- }
-
- // Check pipeline
- if (!need && current_pipeline_size_ < max_pipeline_size_) {
- return Status::OK();
- }
-
- if (current_pipeline_size_ == 0) {
- INFO("[migrate] No commands to send");
- return Status::OK();
- }
-
- applyMigrationSpeedLimit();
-
- auto s = util::SockSend(*dst_fd_, *commands);
- if (!s.IsOK()) {
- return s.Prefixed("failed to write data to a socket");
- }
-
- last_send_time_ = util::GetTimeStampUS();
-
- s = checkMultipleResponses(*dst_fd_, current_pipeline_size_);
- if (!s.IsOK()) {
- return s.Prefixed("wrong response from the destination node");
- }
-
- // Clear commands and running pipeline
- commands->clear();
- current_pipeline_size_ = 0;
-
- return Status::OK();
-}
-
void SlotMigrator::setForbiddenSlotRange(const SlotRange &slot_range) {
INFO("[migrate] Setting forbidden slot(s) {}", slot_range.String());
// Block server to set forbidden slot
@@ -1048,158 +509,6 @@ void SlotMigrator::ReleaseForbiddenSlotRange() {
forbidden_slot_range_ = {-1, -1};
}
-void SlotMigrator::applyMigrationSpeedLimit() const {
- if (max_migration_speed_ > 0) {
- uint64_t current_time = util::GetTimeStampUS();
- uint64_t per_request_time = 1000000 * max_pipeline_size_ /
max_migration_speed_;
- if (per_request_time == 0) {
- per_request_time = 1;
- }
- if (last_send_time_ + per_request_time > current_time) {
- uint64_t during = last_send_time_ + per_request_time - current_time;
- INFO("[migrate] Sleep to limit migration speed for: {}", during);
- std::this_thread::sleep_for(std::chrono::microseconds(during));
- }
- }
-}
-
-Status SlotMigrator::generateCmdsFromBatch(rocksdb::BatchResult *batch,
std::string *commands) {
- // Iterate batch to get keys and construct commands for keys
- WriteBatchExtractor write_batch_extractor(storage_->IsSlotIdEncoded(),
slot_range_, false);
- rocksdb::Status status =
batch->writeBatchPtr->Iterate(&write_batch_extractor);
- if (!status.ok()) {
- ERROR("[migrate] Failed to parse write batch, Err: {}", status.ToString());
- return {Status::NotOK};
- }
-
- // Get all constructed commands
- auto resp_commands = write_batch_extractor.GetRESPCommands();
- for (const auto &iter : *resp_commands) {
- for (const auto &it : iter.second) {
- *commands += it;
- current_pipeline_size_++;
- }
- }
-
- return Status::OK();
-}
-
-Status
SlotMigrator::migrateIncrementData(std::unique_ptr<rocksdb::TransactionLogIterator>
*iter, uint64_t end_seq) {
- if (!(*iter) || !(*iter)->Valid()) {
- ERROR("[migrate] WAL iterator is invalid");
- return {Status::NotOK};
- }
-
- uint64_t next_seq = wal_begin_seq_ + 1;
- std::string commands;
-
- while (true) {
- if (stop_migration_) {
- ERROR("[migrate] Migration task end during migrating WAL data");
- return {Status::NotOK};
- }
-
- auto batch = (*iter)->GetBatch();
- if (batch.sequence != next_seq) {
- ERROR("[migrate] WAL iterator is discrete, some seq might be lost,
expected sequence: {}, but got sequence: {}",
- next_seq, batch.sequence);
- return {Status::NotOK};
- }
-
- // Generate commands by iterating write batch
- auto s = generateCmdsFromBatch(&batch, &commands);
- if (!s.IsOK()) {
- ERROR("[migrate] Failed to generate commands from write batch");
- return {Status::NotOK};
- }
-
- // Check whether command pipeline should be sent
- s = sendCmdsPipelineIfNeed(&commands, false);
- if (!s.IsOK()) {
- ERROR("[migrate] Failed to send WAL commands pipeline");
- return {Status::NotOK};
- }
-
- next_seq = batch.sequence + batch.writeBatchPtr->Count();
- if (next_seq > end_seq) {
- INFO("[migrate] Migrate incremental data an epoch OK, seq from {}, to
{}", wal_begin_seq_, end_seq);
- break;
- }
-
- (*iter)->Next();
- if (!(*iter)->Valid()) {
- ERROR("[migrate] WAL iterator is invalid, expected end seq: {}, next
seq: {}", end_seq, next_seq);
- return {Status::NotOK};
- }
- }
-
- // Send the left data of this epoch
- auto s = sendCmdsPipelineIfNeed(&commands, true);
- if (!s.IsOK()) {
- ERROR("[migrate] Failed to send WAL last commands in pipeline");
- return {Status::NotOK};
- }
-
- return Status::OK();
-}
-
-Status SlotMigrator::syncWalBeforeForbiddingSlot() {
- uint32_t count = 0;
-
- while (count < kMaxLoopTimes) {
- uint64_t latest_seq = storage_->GetDB()->GetLatestSequenceNumber();
- uint64_t gap = latest_seq - wal_begin_seq_;
- if (gap <= static_cast<uint64_t>(seq_gap_limit_)) {
- INFO("[migrate] Incremental data sequence: {}, less than limit: {}, go
to set forbidden slot", gap,
- seq_gap_limit_);
- break;
- }
-
- std::unique_ptr<rocksdb::TransactionLogIterator> iter = nullptr;
- auto s = storage_->GetWALIter(wal_begin_seq_ + 1, &iter);
- if (!s.IsOK()) {
- ERROR("[migrate] Failed to generate WAL iterator before setting
forbidden slot, Err: {}", s.Msg());
- return {Status::NotOK};
- }
-
- // Iterate wal and migrate data
- s = migrateIncrementData(&iter, latest_seq);
- if (!s.IsOK()) {
- ERROR("[migrate] Failed to migrate WAL data before setting forbidden
slot");
- return {Status::NotOK};
- }
-
- wal_begin_seq_ = latest_seq;
- count++;
- }
- INFO("[migrate] Succeed to migrate incremental data before setting forbidden
slot, end epoch: {}", count);
- return Status::OK();
-}
-
-Status SlotMigrator::syncWalAfterForbiddingSlot() {
- uint64_t latest_seq = storage_->GetDB()->GetLatestSequenceNumber();
-
- // No incremental data
- if (latest_seq <= wal_begin_seq_) return Status::OK();
-
- // Get WAL iter
- std::unique_ptr<rocksdb::TransactionLogIterator> iter = nullptr;
- auto s = storage_->GetWALIter(wal_begin_seq_ + 1, &iter);
- if (!s.IsOK()) {
- ERROR("[migrate] Failed to generate WAL iterator after setting forbidden
slot, Err: {}", s.Msg());
- return {Status::NotOK};
- }
-
- // Send incremental data
- s = migrateIncrementData(&iter, latest_seq);
- if (!s.IsOK()) {
- ERROR("[migrate] Failed to migrate WAL data after setting forbidden slot");
- return {Status::NotOK};
- }
-
- return Status::OK();
-}
-
void SlotMigrator::GetMigrationInfo(std::string *info) const {
info->clear();
if (!slot_range_.load().IsValid() && !forbidden_slot_range_.load().IsValid()
&&
@@ -1255,7 +564,7 @@ Status SlotMigrator::sendMigrationBatch(BatchSender
*batch) {
return batch->Send();
}
-Status SlotMigrator::sendSnapshotByRawKV() {
+Status SlotMigrator::sendSnapshot() {
uint64_t start_ts = util::GetTimeStampMS();
auto slot_range = slot_range_.load();
INFO("[migrate] Migrating snapshot of slot(s) {} by raw key value",
slot_range.String());
@@ -1335,7 +644,7 @@ Status SlotMigrator::sendSnapshotByRawKV() {
return Status::OK();
}
-Status SlotMigrator::syncWALByRawKV() {
+Status SlotMigrator::syncWAL() {
uint64_t start_ts = util::GetTimeStampMS();
INFO("[migrate] Syncing WAL of slot(s) {} by raw key value",
slot_range_.load().String());
BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_,
migrate_batch_bytes_per_sec_);
diff --git a/src/cluster/slot_migrate.h b/src/cluster/slot_migrate.h
index 71107a5af..b5d6a1a28 100644
--- a/src/cluster/slot_migrate.h
+++ b/src/cluster/slot_migrate.h
@@ -29,7 +29,6 @@
#include <string>
#include <thread>
#include <utility>
-#include <vector>
#include "batch_sender.h"
#include "logging.h"
@@ -38,33 +37,13 @@
#include "storage/redis_db.h"
#include "unique_fd.h"
-enum class MigrationType {
- /// Use Redis commands to migrate data.
- /// It will try to extract commands from existing data and log, then replay
- /// them on the destination node.
- kRedisCommand = 0,
- /// Using raw key-value and "APPLYBATCH" command in kvrocks to migrate data.
- ///
- /// If downstream is not compatible with raw key-value, this migration type
will
- /// auto switch to kRedisCommand.
- kRawKeyValue
-};
-
enum class MigrationState { kNone = 0, kStarted, kSuccess, kFailed };
enum class SlotMigrationStage { kNone, kStart, kSnapshot, kWAL, kSuccess,
kFailed, kClean };
-enum class KeyMigrationResult { kMigrated, kExpired, kUnderlyingStructEmpty };
-
struct SlotMigrationJob {
- SlotMigrationJob(const SlotRange &slot_range_in, std::string dst_ip, int
dst_port, int speed, int pipeline_size,
- int seq_gap)
- : slot_range(slot_range_in),
- dst_ip(std::move(dst_ip)),
- dst_port(dst_port),
- max_speed(speed),
- max_pipeline_size(pipeline_size),
- seq_gap_limit(seq_gap) {}
+ SlotMigrationJob(const SlotRange &slot_range_in, std::string dst_ip, int
dst_port, int seq_gap)
+ : slot_range(slot_range_in), dst_ip(std::move(dst_ip)),
dst_port(dst_port), seq_gap_limit(seq_gap) {}
SlotMigrationJob(const SlotMigrationJob &other) = delete;
SlotMigrationJob &operator=(const SlotMigrationJob &other) = delete;
~SlotMigrationJob() = default;
@@ -72,8 +51,6 @@ struct SlotMigrationJob {
SlotRange slot_range;
std::string dst_ip;
int dst_port;
- int max_speed;
- int max_pipeline_size;
int seq_gap_limit;
};
@@ -90,12 +67,6 @@ class SlotMigrator : public redis::Database {
Status PerformSlotRangeMigration(const std::string &node_id, std::string
&dst_ip, int dst_port,
const SlotRange &range, SyncMigrateContext
*blocking_ctx = nullptr);
void ReleaseForbiddenSlotRange();
- void SetMaxMigrationSpeed(int value) {
- if (value >= 0) max_migration_speed_ = value;
- }
- void SetMaxPipelineSize(int value) {
- if (value > 0) max_pipeline_size_ = value;
- }
void SetSequenceGapLimit(int value) {
if (value > 0) seq_gap_limit_ = value;
}
@@ -115,8 +86,6 @@ class SlotMigrator : public redis::Database {
void runMigrationProcess();
bool isTerminated() const { return thread_state_ == ThreadState::Terminated;
}
Status startMigration();
- Status sendSnapshot();
- Status syncWAL();
Status finishSuccessfulMigration();
Status finishFailedMigration();
void clean();
@@ -125,30 +94,12 @@ class SlotMigrator : public redis::Database {
Status setImportStatusOnDstNode(int sock_fd, int status);
static StatusOr<bool> supportedApplyBatchCommandOnDstNode(int sock_fd);
- Status sendSnapshotByCmd();
- Status syncWALByCmd();
Status checkSingleResponse(int sock_fd);
Status checkMultipleResponses(int sock_fd, int total);
- StatusOr<KeyMigrationResult> migrateOneKey(const rocksdb::Slice &key, const
rocksdb::Slice &encoded_metadata,
- std::string *restore_cmds);
- Status migrateSimpleKey(const rocksdb::Slice &key, const Metadata &metadata,
const std::string &bytes,
- std::string *restore_cmds);
- Status migrateComplexKey(const rocksdb::Slice &key, const Metadata
&metadata, std::string *restore_cmds);
- Status migrateStream(const rocksdb::Slice &key, const StreamMetadata
&metadata, std::string *restore_cmds);
- Status migrateBitmapKey(const InternalKey &inkey,
std::unique_ptr<rocksdb::Iterator> *iter,
- std::vector<std::string> *user_cmd, std::string
*restore_cmds);
-
- Status sendCmdsPipelineIfNeed(std::string *commands, bool need);
- void applyMigrationSpeedLimit() const;
- Status generateCmdsFromBatch(rocksdb::BatchResult *batch, std::string
*commands);
- Status migrateIncrementData(std::unique_ptr<rocksdb::TransactionLogIterator>
*iter, uint64_t end_seq);
- Status syncWalBeforeForbiddingSlot();
- Status syncWalAfterForbiddingSlot();
-
Status sendMigrationBatch(BatchSender *batch);
- Status sendSnapshotByRawKV();
- Status syncWALByRawKV();
+ Status sendSnapshot();
+ Status syncWAL();
bool catchUpIncrementalWAL();
Status migrateIncrementalDataByRawKV(uint64_t end_seq, BatchSender
*batch_sender);
@@ -160,16 +111,11 @@ class SlotMigrator : public redis::Database {
enum class ParserState { ArrayLen, BulkLen, BulkData, ArrayData, OneRspEnd };
enum class ThreadState { Uninitialized, Running, Terminated };
- static constexpr int kDefaultMaxPipelineSize = 16;
- static constexpr int kDefaultMaxMigrationSpeed = 4096;
static constexpr int kDefaultSequenceGapLimit = 10000;
- static constexpr int kMaxItemsInCommand = 16; // number of items in every
write command of complex keys
static constexpr int kMaxLoopTimes = 10;
Server *srv_;
- int max_migration_speed_ = kDefaultMaxMigrationSpeed;
- int max_pipeline_size_ = kDefaultMaxPipelineSize;
uint64_t seq_gap_limit_ = kDefaultSequenceGapLimit;
std::atomic<size_t> migrate_batch_bytes_per_sec_ = 1 * GiB;
std::atomic<size_t> migrate_batch_size_bytes_;
@@ -179,9 +125,6 @@ class SlotMigrator : public redis::Database {
std::atomic<ThreadState> thread_state_ = ThreadState::Uninitialized;
std::atomic<MigrationState> migration_state_ = MigrationState::kNone;
- int current_pipeline_size_ = 0;
- uint64_t last_send_time_ = 0;
-
std::thread t_;
std::mutex job_mutex_;
std::condition_variable job_cv_;
@@ -192,8 +135,6 @@ class SlotMigrator : public redis::Database {
int dst_port_ = -1;
UniqueFD dst_fd_;
- MigrationType migration_type_ = MigrationType::kRedisCommand;
-
static_assert(std::atomic<SlotRange>::is_always_lock_free, "SlotRange is not
lock free.");
std::atomic<SlotRange> forbidden_slot_range_ = SlotRange{-1, -1};
std::atomic<SlotRange> slot_range_ = SlotRange{-1, -1};
diff --git a/src/config/config.cc b/src/config/config.cc
index 492eb975b..3932574cd 100644
--- a/src/config/config.cc
+++ b/src/config/config.cc
@@ -90,9 +90,6 @@ const std::vector<ConfigEnum<BlockCacheType>> cache_types{[] {
return res;
}()};
-const std::vector<ConfigEnum<MigrationType>> migration_types{{"redis-command",
MigrationType::kRedisCommand},
- {"raw-key-value",
MigrationType::kRawKeyValue}};
-
const std::vector<ConfigEnum<HashSubkeyEncodingMode>>
hash_subkey_encoding_modes{
{"legacy", HashSubkeyEncodingMode::kLegacy},
{"field-expiration", HashSubkeyEncodingMode::kFieldExpiration},
@@ -140,7 +137,13 @@ Status SetRocksdbCompression(Server *srv, const
rocksdb::CompressionType compres
};
Config::Config() {
- deprecated_fields_ = {"rocksdb.row_cache_size"};
+ // The `migrate-type` config was deprecated after removing the
`redis-command` migration type,
+ // and the server will always migrate slots in the way of the raw key-value.
+ // The `migrate-speed` and `migrate-pipeline-size` configs only took effect
in the
+ // `redis-command` migration type, so they were deprecated as well. Use the
+ // `migrate-batch-rate-limit-mb` and `migrate-batch-size-kb` configs to
control the
+ // migration speed instead.
+ deprecated_fields_ = {"rocksdb.row_cache_size", "migrate-type",
"migrate-speed", "migrate-pipeline-size"};
struct FieldWrapper {
std::string name;
@@ -233,11 +236,7 @@ Config::Config() {
{"rename-command", true, new MultiStringField(&rename_command_,
std::vector<std::string>{})},
{"fullsync-recv-file-delay", false, new
IntField(&fullsync_recv_file_delay, 0, 0, INT_MAX)},
{"cluster-enabled", true, new YesNoField(&cluster_enabled, false)},
- {"migrate-speed", false, new IntField(&migrate_speed, 4096, 0, INT_MAX)},
- {"migrate-pipeline-size", false, new IntField(&pipeline_size, 16, 1,
INT_MAX)},
{"migrate-sequence-gap", false, new IntField(&sequence_gap, 10000, 1,
INT_MAX)},
- {"migrate-type", false,
- new EnumField<MigrationType>(&migrate_type, migration_types,
MigrationType::kRawKeyValue)},
{"migrate-batch-size-kb", false, new IntField(&migrate_batch_size_kb,
16, 1, INT_MAX)},
{"migrate-batch-rate-limit-mb", false, new
IntField(&migrate_batch_rate_limit_mb, 16, 1, INT_MAX)},
{"unixsocket", true, new StringField(&unixsocket, "")},
@@ -603,18 +602,6 @@ void Config::initFieldCallback() {
srv->GetPerfLog()->SetMaxEntries(profiling_sample_record_max_len);
return Status::OK();
}},
- {"migrate-speed",
- [this](Server *srv, [[maybe_unused]] const std::string &k,
[[maybe_unused]] const std::string &v) -> Status {
- if (!srv) return Status::OK();
- if (cluster_enabled)
srv->slot_migrator->SetMaxMigrationSpeed(migrate_speed);
- return Status::OK();
- }},
- {"migrate-pipeline-size",
- [this](Server *srv, [[maybe_unused]] const std::string &k,
[[maybe_unused]] const std::string &v) -> Status {
- if (!srv) return Status::OK();
- if (cluster_enabled)
srv->slot_migrator->SetMaxPipelineSize(pipeline_size);
- return Status::OK();
- }},
{"migrate-sequence-gap",
[this](Server *srv, [[maybe_unused]] const std::string &k,
[[maybe_unused]] const std::string &v) -> Status {
if (!srv) return Status::OK();
diff --git a/src/config/config.h b/src/config/config.h
index c17b331e7..62c13c491 100644
--- a/src/config/config.h
+++ b/src/config/config.h
@@ -37,7 +37,6 @@
// forward declaration
class Server;
-enum class MigrationType;
namespace engine {
class Storage;
}
@@ -169,10 +168,7 @@ struct Config {
bool slot_id_encoded = false;
bool cluster_enabled = false;
- int migrate_speed;
- int pipeline_size;
int sequence_gap;
- MigrationType migrate_type;
int migrate_batch_size_kb;
int migrate_batch_rate_limit_mb;
diff --git a/tests/gocase/integration/cluster/cluster_test.go
b/tests/gocase/integration/cluster/cluster_test.go
index 355b5c10c..8bbf02410 100644
--- a/tests/gocase/integration/cluster/cluster_test.go
+++ b/tests/gocase/integration/cluster/cluster_test.go
@@ -576,9 +576,11 @@ func TestClusterReset(t *testing.T) {
slotNum := 2
// slow down the migration speed to ensure we can observe the
"start" state
// before migration completes (especially on fast hardware like
macOS ARM)
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-speed",
"64").Err())
- for i := 0; i < 2048; i++ {
- require.NoError(t, rdb0.RPush(ctx, "my-list",
fmt.Sprintf("element%d", i)).Err())
+ require.NoError(t, rdb0.ConfigSet(ctx, "migrate-batch-size-kb",
"1").Err())
+ require.NoError(t, rdb0.ConfigSet(ctx,
"migrate-batch-rate-limit-mb", "1").Err())
+ value := strings.Repeat("a", 512)
+ for i := 0; i < 4096; i++ {
+ require.NoError(t, rdb0.RPush(ctx, "my-list",
value).Err())
}
require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate",
slotNum, id1).Val())
diff --git a/tests/gocase/integration/slotmigrate/slotmigrate_test.go
b/tests/gocase/integration/slotmigrate/slotmigrate_test.go
index dee732408..6e430ce63 100644
--- a/tests/gocase/integration/slotmigrate/slotmigrate_test.go
+++ b/tests/gocase/integration/slotmigrate/slotmigrate_test.go
@@ -35,7 +35,6 @@ import (
type SlotMigrationState string
type SlotImportState string
-type SlotMigrationType string
const (
SlotMigrationStateStarted SlotMigrationState = "start"
@@ -44,9 +43,6 @@ const (
SlotImportStateSuccess SlotImportState = "success"
SlotImportStateFailed SlotImportState = "error"
-
- MigrationTypeRedisCommand SlotMigrationType = "redis-command"
- MigrationTypeRawKeyValue SlotMigrationType = "raw-key-value"
)
var testSlot = 0
@@ -236,20 +232,19 @@ func TestSlotMigrateSourceServerFlushedOrKilled(t
*testing.T) {
t.Run("MIGRATE - Fail to migrate slot because source server is
flushed", func(t *testing.T) {
slot := 11
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-speed",
"32").Err())
- // FLUSHDB only allowed in `redis-command` migrate type
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-type",
"redis-command").Err())
- defer func() {
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-type",
"raw-key-value").Err())
- }()
+ // Slow down the migration to ensure it's still running while
flushing the DB
+ require.NoError(t, rdb0.ConfigSet(ctx, "migrate-batch-size-kb",
"1").Err())
+ require.NoError(t, rdb0.ConfigSet(ctx,
"migrate-batch-rate-limit-mb", "1").Err())
+ value := strings.Repeat("a", 512)
for i := 0; i < 20000; i++ {
- require.NoError(t, rdb0.LPush(ctx,
util.SlotTable[slot], i).Err())
+ require.NoError(t, rdb0.LPush(ctx,
util.SlotTable[slot], value).Err())
}
require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate",
slot, id1).Val())
waitForMigrateState(t, rdb0, slot, SlotMigrationStateStarted)
require.NoError(t, rdb0.FlushDB(ctx).Err())
- time.Sleep(time.Second)
- waitForMigrateState(t, rdb0, slot, SlotMigrationStateFailed)
+ // The stop flag set by FLUSHDB is only checked when finishing
the migration,
+ // so it takes a while for the rate-limited migration to reach
the failed state.
+ waitForMigrateStateInDuration(t, rdb0, slot,
SlotMigrationStateFailed, time.Minute)
})
t.Run("MIGRATE - Fail to migrate slot because source server is killed
while migrating", func(t *testing.T) {
@@ -548,9 +543,7 @@ func TestSlotMigrateDataType(t *testing.T) {
require.EqualValues(t, cnt, rdb1.LLen(ctx,
util.SlotTable[slot]).Val())
})
- migrateAllTypes := func(t *testing.T, migrateType SlotMigrationType,
sync bool) {
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-type",
string(migrateType)).Err())
-
+ migrateAllTypes := func(t *testing.T, sync bool) {
testSlot += 1
keys := make(map[string]string, 0)
for _, typ := range []string{"string", "expired_string",
"list", "hash", "set", "zset", "bitmap", "sortint", "stream", "json"} {
@@ -680,8 +673,7 @@ func TestSlotMigrateDataType(t *testing.T) {
}
}
- migrateIncrementalStream := func(t *testing.T, migrateType
SlotMigrationType) {
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-type",
string(migrateType)).Err())
+ migrateIncrementalStream := func(t *testing.T) {
testSlot += 1
keys := make(map[string]string, 0)
for _, typ := range []string{"stream"} {
@@ -702,11 +694,6 @@ func TestSlotMigrateDataType(t *testing.T) {
require.EqualValues(t, "0-0", streamInfo.MaxDeletedEntryID)
require.EqualValues(t, 999, streamInfo.Length)
- // Slowdown the migration speed to prevent running before next
increment commands
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-speed",
"256").Err())
- defer func() {
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-speed",
"4096").Err())
- }()
require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate",
testSlot, id1).Val())
newStreamID := "1001"
require.NoError(t, rdb0.XAdd(ctx, &redis.XAddArgs{
@@ -725,9 +712,7 @@ func TestSlotMigrateDataType(t *testing.T) {
require.EqualValues(t, 999, streamInfo.Length)
}
- migrateEmptyStream := func(t *testing.T, migrateType SlotMigrationType)
{
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-type",
string(migrateType)).Err())
-
+ migrateEmptyStream := func(t *testing.T) {
testSlot += 1
key := fmt.Sprintf("stream_{%s}", util.SlotTable[testSlot])
@@ -768,9 +753,7 @@ func TestSlotMigrateDataType(t *testing.T) {
require.EqualValues(t, originRes.Length, migratedRes.Length)
}
- migrateStreamWithDeletedEntries := func(t *testing.T, migrateType
SlotMigrationType) {
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-type",
string(migrateType)).Err())
-
+ migrateStreamWithDeletedEntries := func(t *testing.T) {
testSlot += 1
key := fmt.Sprintf("stream_{%s}", util.SlotTable[testSlot])
@@ -809,13 +792,12 @@ func TestSlotMigrateDataType(t *testing.T) {
require.EqualValues(t, originRes.Length, migratedRes.Length)
}
- migrateIncrementalData := func(t *testing.T, migrateType
SlotMigrationType) {
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-type",
string(migrateType)).Err())
+ migrateIncrementalData := func(t *testing.T) {
testSlot += 1
migratingSlot := testSlot
hashtag := util.SlotTable[migratingSlot]
keys := []string{
- // key for slowing migrate-speed when migrating
existing data
+ // key for slowing down the migration when migrating
existing data
hashtag,
fmt.Sprintf("{%s}_key1", hashtag),
fmt.Sprintf("{%s}_key2", hashtag),
@@ -831,14 +813,10 @@ func TestSlotMigrateDataType(t *testing.T) {
require.NoError(t, rdb0.Del(ctx, key).Err())
}
- valuePrefix := "value"
- if migrateType == MigrationTypeRedisCommand {
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-speed",
"64").Err())
- } else {
- // Create enough data
- valuePrefix = strings.Repeat("value", 1024)
- require.NoError(t, rdb0.ConfigSet(ctx,
"migrate-batch-rate-limit-mb", "1").Err())
- }
+ // Create enough data to slow down the migration with the rate
limit,
+ // so that the increment commands below are synced via the WAL
+ valuePrefix := strings.Repeat("value", 1024)
+ require.NoError(t, rdb0.ConfigSet(ctx,
"migrate-batch-rate-limit-mb", "1").Err())
cnt := 2000
for i := 0; i < cnt; i++ {
@@ -962,33 +940,29 @@ func TestSlotMigrateDataType(t *testing.T) {
require.EqualValues(t, 0, rdb0.Exists(ctx,
util.SlotTable[slotWithDeletedKey]).Val())
}
- testMigrationTypes := []SlotMigrationType{MigrationTypeRedisCommand,
MigrationTypeRawKeyValue}
-
- for _, testType := range testMigrationTypes {
- t.Run(fmt.Sprintf("MIGRATE - Slot migrate all types of existing
data using %s", testType), func(t *testing.T) {
- migrateAllTypes(t, testType, false)
- })
+ t.Run("MIGRATE - Slot migrate all types of existing data", func(t
*testing.T) {
+ migrateAllTypes(t, false)
+ })
- t.Run(fmt.Sprintf("MIGRATE - Slot migrate all types of existing
data (sync) using %s", testType), func(t *testing.T) {
- migrateAllTypes(t, testType, true)
- })
+ t.Run("MIGRATE - Slot migrate all types of existing data (sync)",
func(t *testing.T) {
+ migrateAllTypes(t, true)
+ })
- t.Run(fmt.Sprintf("MIGRATE - increment sync stream from WAL
using %s", testType), func(t *testing.T) {
- migrateIncrementalStream(t, testType)
- })
+ t.Run("MIGRATE - increment sync stream from WAL", func(t *testing.T) {
+ migrateIncrementalStream(t)
+ })
- t.Run(fmt.Sprintf("MIGRATE - Migrating empty stream using %s",
testType), func(t *testing.T) {
- migrateEmptyStream(t, testType)
- })
+ t.Run("MIGRATE - Migrating empty stream", func(t *testing.T) {
+ migrateEmptyStream(t)
+ })
- t.Run(fmt.Sprintf("MIGRATE - Migrating stream with deleted
entries using %s", testType), func(t *testing.T) {
- migrateStreamWithDeletedEntries(t, testType)
- })
+ t.Run("MIGRATE - Migrating stream with deleted entries", func(t
*testing.T) {
+ migrateStreamWithDeletedEntries(t)
+ })
- t.Run(fmt.Sprintf("MIGRATE - Migrate incremental data via
parsing and filtering data in WAL using %s", testType), func(t *testing.T) {
- migrateIncrementalData(t, testType)
- })
- }
+ t.Run("MIGRATE - Migrate incremental data via parsing and filtering
data in WAL", func(t *testing.T) {
+ migrateIncrementalData(t)
+ })
t.Run("MIGRATE - Accessing slot is forbidden on source server but not
on destination server", func(t *testing.T) {
testSlot += 1
@@ -1027,15 +1001,14 @@ func TestSlotMigrateDataType(t *testing.T) {
})
t.Run("MIGRATE - Slow migrate speed", func(t *testing.T) {
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-type",
string(MigrationTypeRedisCommand)).Err())
testSlot += 1
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-speed",
"16").Err())
- require.Equal(t, map[string]string{"migrate-speed": "16"},
rdb0.ConfigGet(ctx, "migrate-speed").Val())
+ require.NoError(t, rdb0.ConfigSet(ctx, "migrate-batch-size-kb",
"1").Err())
+ require.NoError(t, rdb0.ConfigSet(ctx,
"migrate-batch-rate-limit-mb", "1").Err())
require.NoError(t, rdb0.Del(ctx,
util.SlotTable[testSlot]).Err())
- // more than pipeline size(16) and max items(16) in command
- cnt := 1000
+ value := strings.Repeat("a", 512)
+ cnt := 10000
for i := 0; i < cnt; i++ {
- require.NoError(t, rdb0.LPush(ctx,
util.SlotTable[testSlot], i).Err())
+ require.NoError(t, rdb0.LPush(ctx,
util.SlotTable[testSlot], value).Err())
}
require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate",
testSlot, id1).Val())
@@ -1045,7 +1018,7 @@ func TestSlotMigrateDataType(t *testing.T) {
// should not finish 1.5s
time.Sleep(1500 * time.Millisecond)
requireMigrateState(t, rdb0, testSlot,
SlotMigrationStateStarted)
- waitForMigrateState(t, rdb0, testSlot,
SlotMigrationStateSuccess)
+ waitForMigrateStateInDuration(t, rdb0, testSlot,
SlotMigrationStateSuccess, time.Minute)
})
t.Run("MIGRATE - Data of migrated slot can't be written to source but
can be written to destination", func(t *testing.T) {
@@ -1070,14 +1043,20 @@ func TestSlotMigrateDataType(t *testing.T) {
srcListName := fmt.Sprintf("list_src_{%s}",
util.SlotTable[testSlot])
dstListName := fmt.Sprintf("list_dst_{%s}",
util.SlotTable[testSlot])
+ fillerListName := fmt.Sprintf("list_filler_{%s}",
util.SlotTable[testSlot])
require.NoError(t, rdb0.Del(ctx, srcListName).Err())
require.NoError(t, rdb0.Del(ctx, dstListName).Err())
+ require.NoError(t, rdb0.Del(ctx, fillerListName).Err())
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-speed",
"512").Err())
- defer func() {
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-speed",
"4096").Err())
- }()
+ // Slow down the migration with the rate limit and filler data,
+ // so that the LMOVE commands below are synced via the WAL
+ require.NoError(t, rdb0.ConfigSet(ctx, "migrate-batch-size-kb",
"1").Err())
+ require.NoError(t, rdb0.ConfigSet(ctx,
"migrate-batch-rate-limit-mb", "1").Err())
+ fillerValue := strings.Repeat("a", 512)
+ for i := 0; i < 10000; i++ {
+ require.NoError(t, rdb0.LPush(ctx, fillerListName,
fillerValue).Err())
+ }
for i := 0; i < 1000; i++ {
require.NoError(t, rdb0.RPush(ctx, srcListName,
fmt.Sprintf("element%d", i)).Err())
@@ -1089,7 +1068,7 @@ func TestSlotMigrateDataType(t *testing.T) {
for i := 0; i < 10; i++ {
require.NoError(t, rdb0.LMove(ctx, srcListName,
dstListName, "RIGHT", "LEFT").Err())
}
- waitForMigrateState(t, rdb0, testSlot,
SlotMigrationStateSuccess)
+ waitForMigrateStateInDuration(t, rdb0, testSlot,
SlotMigrationStateSuccess, time.Minute)
require.ErrorContains(t, rdb0.RPush(ctx, srcListName,
"element1000").Err(), "MOVED")
require.Equal(t, int64(10), rdb1.LLen(ctx, dstListName).Val())
@@ -1104,13 +1083,19 @@ func TestSlotMigrateDataType(t *testing.T) {
testSlot += 1
srcListName := fmt.Sprintf("list_src_{%s}",
util.SlotTable[testSlot])
+ fillerListName := fmt.Sprintf("list_filler_{%s}",
util.SlotTable[testSlot])
require.NoError(t, rdb0.Del(ctx, srcListName).Err())
+ require.NoError(t, rdb0.Del(ctx, fillerListName).Err())
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-speed",
"512").Err())
- defer func() {
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-speed",
"4096").Err())
- }()
+ // Slow down the migration with the rate limit and filler data,
+ // so that the LMOVE commands below are synced via the WAL
+ require.NoError(t, rdb0.ConfigSet(ctx, "migrate-batch-size-kb",
"1").Err())
+ require.NoError(t, rdb0.ConfigSet(ctx,
"migrate-batch-rate-limit-mb", "1").Err())
+ fillerValue := strings.Repeat("a", 512)
+ for i := 0; i < 10000; i++ {
+ require.NoError(t, rdb0.LPush(ctx, fillerListName,
fillerValue).Err())
+ }
srcLen := 1_000
@@ -1124,7 +1109,7 @@ func TestSlotMigrateDataType(t *testing.T) {
for i := 0; i < 10; i++ {
require.NoError(t, rdb0.LMove(ctx, srcListName,
srcListName, "RIGHT", "LEFT").Err())
}
- waitForMigrateState(t, rdb0, testSlot,
SlotMigrationStateSuccess)
+ waitForMigrateStateInDuration(t, rdb0, testSlot,
SlotMigrationStateSuccess, time.Minute)
require.ErrorContains(t, rdb0.RPush(ctx, srcListName,
"element1000").Err(), "MOVED")
require.Equal(t, int64(srcLen), rdb1.LLen(ctx,
srcListName).Val())
@@ -1135,12 +1120,14 @@ func TestSlotMigrateDataType(t *testing.T) {
})
}
-func TestSlotMigrateTypeFallback(t *testing.T) {
+func TestSlotMigrateDestNotSupportApplyBatch(t *testing.T) {
ctx := context.Background()
srv0 := util.StartServer(t, map[string]string{
"cluster-enabled": "yes",
- "migrate-type": "raw-key-value",
+ // The `migrate-type` config was deprecated after removing the
`redis-command`
+ // migration type, but the server should still be able to start
with it.
+ "migrate-type": "redis-command",
})
defer srv0.Close()
@@ -1164,7 +1151,7 @@ func TestSlotMigrateTypeFallback(t *testing.T) {
require.NoError(t, rdb0.Do(ctx, "clusterx", "setnodes", clusterNodes,
"1").Err())
require.NoError(t, rdb1.Do(ctx, "clusterx", "setnodes", clusterNodes,
"1").Err())
- t.Run("MIGRATE - Fall back to redis-command migration type when the
destination does not support APPLYBATCH", func(t *testing.T) {
+ t.Run("MIGRATE - Fail to migrate slot when the destination does not
support APPLYBATCH", func(t *testing.T) {
info, err := rdb1.Do(ctx, "command", "info",
"applybatch").Slice()
require.NoError(t, err)
require.Len(t, info, 1)
@@ -1174,8 +1161,8 @@ func TestSlotMigrateTypeFallback(t *testing.T) {
value := "value"
require.NoError(t, rdb0.Set(ctx, key, value, 0).Err())
require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate",
testSlot, id1).Val())
- waitForMigrateState(t, rdb0, testSlot,
SlotMigrationStateSuccess)
- require.Equal(t, value, rdb1.Get(ctx, key).Val())
+ waitForMigrateState(t, rdb0, testSlot, SlotMigrationStateFailed)
+ require.ErrorContains(t, rdb1.Exists(ctx, key).Err(), "MOVED")
})
}
@@ -1204,7 +1191,6 @@ func TestSlotMigrateCuckooFilter(t *testing.T) {
t.Run("MIGRATE - Cuckoo filter is preserved by raw-key-value
migration", func(t *testing.T) {
slot := 31
key := fmt.Sprintf("cf_{%s}", util.SlotTable[slot])
- require.NoError(t, rdb0.ConfigSet(ctx, "migrate-type",
string(MigrationTypeRawKeyValue)).Err())
require.NoError(t, rdb0.Do(ctx, "cf.reserve", key,
"1000").Err())
for i := 0; i < 20; i++ {
require.NoError(t, rdb0.Do(ctx, "cf.add", key,
fmt.Sprintf("item_%d", i)).Err())
