diff options
Diffstat (limited to 'examples/redis-unstable/tests/unit/cluster/atomic-slot-migration.tcl')
| -rw-r--r-- | examples/redis-unstable/tests/unit/cluster/atomic-slot-migration.tcl | 3063 |
1 files changed, 3063 insertions, 0 deletions
diff --git a/examples/redis-unstable/tests/unit/cluster/atomic-slot-migration.tcl b/examples/redis-unstable/tests/unit/cluster/atomic-slot-migration.tcl new file mode 100644 index 0000000..f04257f --- /dev/null +++ b/examples/redis-unstable/tests/unit/cluster/atomic-slot-migration.tcl @@ -0,0 +1,3063 @@ +set ::slot_prefixes [dict create \ + 0 "{06S}" \ + 1 "{Qi}" \ + 2 "{5L5}" \ + 3 "{4Iu}" \ + 4 "{4gY}" \ + 5 "{460}" \ + 6 "{1Y7}" \ + 7 "{1LV}" \ + 101 "{1j2}" \ + 102 "{75V}" \ + 103 "{bno}" \ + 5462 "{450}"\ + 5463 "{4dY}"\ + 6000 "{4L7}" \ + 6001 "{4YV}" \ + 6002 "{0bx}" \ + 6003 "{AJ}" \ + 6004 "{of}" \ + 16383 "{6ZJ}" \ +] + +# Helper functions +proc get_port {node_id} { + if {$::tls} { + return [lindex [R $node_id config get tls-port] 1] + } else { + return [lindex [R $node_id config get port] 1] + } +} + +# return the prefix for the given slot +proc slot_prefix {slot} { + return [dict get $::slot_prefixes $slot] +} + +# return a key for the given slot +proc slot_key {slot {suffix ""}} { + return "[slot_prefix $slot]$suffix" +} + +# Populate a slot with keys +# TODO: Consider merging with populate() +proc populate_slot {num args} { + # Default values + set prefix "key:" + set size 3 + set idx 0 + set prints false + set expires 0 + set slot -1 + + # Parse named arguments + foreach {key value} $args { + switch -- $key { + -prefix { set prefix $value } + -size { set size $value } + -idx { set idx $value } + -prints { set prints $value } + -expires { set expires $value } + -slot { set slot $value } + default { error "Unknown option: $key" } + } + } + + # If slot is specified, use slot prefix from table + if {$slot >= 0} { + if {[dict exists $::slot_prefixes $slot]} { + set prefix [dict get $::slot_prefixes $slot] + } else { + error "Slot $slot not supported in slot_prefixes table, add it manually" + } + } + + R $idx deferred 1 + if {$num > 16} {set pipeline 16} else {set pipeline $num} + set val [string repeat A $size] + for {set j 0} {$j < $pipeline} {incr j} { + if {$expires > 0} { + R $idx set $prefix$j $val ex $expires + } else { + R $idx set $prefix$j $val + } + if {$prints} {puts $j} + } + for {} {$j < $num} {incr j} { + if {$expires > 0} { + R $idx set $prefix$j $val ex $expires + } else { + R $idx set $prefix$j $val + } + R $idx read + if {$prints} {puts $j} + } + for {set j 0} {$j < $pipeline} {incr j} { + R $idx read + if {$prints} {puts $j} + } + R $idx deferred 0 +} + +# Return 1 if all instances are idle +proc asm_all_instances_idle {total} { + for {set i 0} {$i < $total} {incr i} { + if {[CI $i cluster_slot_migration_active_tasks] != 0} { return 0 } + if {[CI $i cluster_slot_migration_active_trim_running] != 0} { return 0 } + } + return 1 +} + +# Wait for all ASM tasks to complete in the cluster +proc wait_for_asm_done {} { + set total_instances [expr {$::cluster_master_nodes + $::cluster_replica_nodes}] + + wait_for_condition 1000 10 { + [asm_all_instances_idle $total_instances] == 1 + } else { + # Print the number of active tasks on each instance + for {set i 0} {$i < $total_instances} {incr i} { + set migration_count [CI $i cluster_slot_migration_active_tasks] + set trim_count [CI $i cluster_slot_migration_active_trim_running] + puts "Instance $i: migration_tasks=$migration_count, trim_tasks=$trim_count" + } + fail "ASM tasks did not complete on all instances" + } + # wait all nodes to reach the same cluster config after ASM + wait_for_cluster_propagation +} + +proc failover_and_wait_for_done {node_id {failover_arg ""}} { + set max_attempts 5 + for {set attempt 1} {$attempt <= $max_attempts} {incr attempt} { + if {$failover_arg eq ""} { + R $node_id cluster failover + } else { + R $node_id cluster failover $failover_arg + } + + set completed 1 + wait_for_condition 1000 10 { + [string match "*master*" [R $node_id role]] + } else { + set completed 0 + } + + if {$completed} { + wait_for_cluster_propagation + return + } + } + fail "Failover did not complete after $max_attempts attempts for node $node_id" +} + +proc migration_status {node_id task_id field} { + set status [R $node_id CLUSTER MIGRATION STATUS ID $task_id] + + # STATUS ID returns single task, so get first element + if {[llength $status] == 0} { + return "" + } + + set task_status [lindex $status 0] + set field_value "" + + # Parse the key-value pairs in the task + for {set i 0} {$i < [llength $task_status]} {incr i 2} { + set key [lindex $task_status $i] + set value [lindex $task_status [expr $i + 1]] + + if {$key eq $field} { + set field_value $value + break + } + } + + return $field_value +} + +# Setup slot migration test with keys and delay, then start migration +# Returns the task_id for the migration +proc setup_slot_migration_with_delay {src_node dst_node start_slot end_slot {keys 2} {delay 1000000}} { + # Two keys on the start slot + populate_slot $keys -idx $src_node -slot $start_slot + + # we set a delay to ensure migration takes time for testing, + # with default parameters, two keys cost 2s to save + R $src_node config set rdb-key-save-delay $delay + + # migrate slot range from src_node to dst_node + set task_id [R $dst_node CLUSTER MIGRATION IMPORT $start_slot $end_slot] + wait_for_condition 2000 10 { + [string match {*send-bulk-and-stream*} [migration_status $src_node $task_id state]] + } else { + fail "ASM task did not start" + } + + return $task_id +} + +# Helper function to clear module internal event logs +proc clear_module_event_log {} { + for {set i 0} {$i < $::cluster_master_nodes + $::cluster_replica_nodes} {incr i} { + R $i asm.clear_event_log + } +} + +proc reset_default_trim_method {} { + for {set i 0} {$i < $::cluster_master_nodes + $::cluster_replica_nodes} {incr i} { + R $i debug asm-trim-method default + } +} + +start_cluster 3 3 {tags {external:skip cluster} overrides {cluster-node-timeout 60000 cluster-allow-replica-migration no}} { + foreach trim_method {"active" "bg"} { + test "Simple slot migration (trim method: $trim_method)" { + R 0 debug asm-trim-method $trim_method + R 3 debug asm-trim-method $trim_method + + set slot0_key [slot_key 0 mykey] + R 0 set $slot0_key "a" + set slot1_key [slot_key 1 mykey] + R 0 set $slot1_key "b" + set slot101_key [slot_key 101 mykey] + R 0 set $slot101_key "c" + # 3 keys cost 3s to save + R 0 config set rdb-key-save-delay 1000000 + + # load a function + R 0 function load {#!lua name=test1 + redis.register_function('test1', function() return 'hello1' end) + } + + # migrate slot 0-100 to R 1 + set task_id [R 1 CLUSTER MIGRATION IMPORT 0 100] + # migration is start, and in accumulating buffer stage + wait_for_condition 1000 50 { + [string match {*send-bulk-and-stream*} [migration_status 0 $task_id state]] && + [string match {*accumulate-buffer*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not start" + } + + # append 99 times during migration + for {set i 0} {$i < 99} {incr i} { + R 0 multi + R 0 append $slot0_key "a" + R 0 exec + R 0 append $slot1_key "b" + R 0 append $slot101_key "c" + } + + # wait until migration of 0-100 successful + wait_for_asm_done + + # verify task state became completed + assert_equal "completed" [migration_status 0 $task_id state] + assert_equal "completed" [migration_status 1 $task_id state] + + # the appended 99 times should also be migrated + assert_equal [string repeat a 100] [R 1 get $slot0_key] + assert_equal [string repeat b 100] [R 1 get $slot1_key] + + # function should be migrated + assert_equal [R 0 function dump] [R 1 function dump] + # the slave should also get the data + wait_for_ofs_sync [Rn 1] [Rn 4] + + R 4 readonly + assert_equal [string repeat a 100] [R 4 get $slot0_key] + assert_equal [string repeat b 100] [R 4 get $slot1_key] + assert_equal [R 0 function dump] [R 4 function dump] + + # verify key that was not in the slot range is not migrated + assert_equal [string repeat c 100] [R 0 get $slot101_key] + # verify changes in replica + wait_for_ofs_sync [Rn 0] [Rn 3] + R 3 readonly + assert_equal [string repeat c 100] [R 3 get $slot101_key] + + # cleanup + R 0 config set rdb-key-save-delay 0 + R 0 flushall + R 0 function flush + R 1 flushall + R 1 function flush + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + } + } +} + +# Skip most of the tests when running under valgrind since it is hard to +# stabilize tests under valgrind. +if {!$::valgrind} { +start_cluster 3 3 {tags {external:skip cluster} overrides {cluster-node-timeout 60000 cluster-allow-replica-migration no}} { + test "Test CLUSTER MIGRATION IMPORT input validation" { + # invalid arguments + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION} + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION IMPORT} + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION IMPORT 100} + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION IMPORT 100 200 300} + assert_error {*unknown argument*} {R 0 CLUSTER MIGRATION UNKNOWN 1 2} + + # invalid slot range + assert_error {*greater than end slot number*} {R 0 CLUSTER MIGRATION IMPORT 200 100} + assert_error {*out of range slot*} {R 0 CLUSTER MIGRATION IMPORT 17000 18000} + assert_error {*out of range slot*} {R 0 CLUSTER MIGRATION IMPORT 14000 18000} + assert_error {*out of range slot*} {R 0 CLUSTER MIGRATION IMPORT 0 16384} + assert_error {*out of range slot*} {R 0 CLUSTER MIGRATION IMPORT 0 -1} + assert_error {*out of range slot*} {R 0 CLUSTER MIGRATION IMPORT -1 2} + assert_error {*out of range slot*} {R 0 CLUSTER MIGRATION IMPORT -2 -1} + assert_error {*out of range slot*} {R 0 CLUSTER MIGRATION IMPORT 10 a} + assert_error {*out of range slot*} {R 0 CLUSTER MIGRATION IMPORT sd sd} + assert_error {*already the owner of the slot*} {R 0 CLUSTER MIGRATION IMPORT 100 200} + } + + test "Test CLUSTER MIGRATION CANCEL input validation" { + # invalid arguments + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION CANCEL} + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION CANCEL ID} + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION CANCEL ID 12345 EXTRAARG} + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION CANCEL ALL EXTRAARG} + assert_error {*unknown argument*} {R 0 CLUSTER MIGRATION CANCEL UNKNOWNARG} + assert_error {*unknown argument*} {R 0 CLUSTER MIGRATION CANCEL abc def} + # empty string id should not cancel any task + assert_equal 0 [R 0 CLUSTER MIGRATION CANCEL ID ""] + } + + test "Test CLUSTER MIGRATION STATUS input validation" { + # invalid arguments + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION STATUS} + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION STATUS ID} + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION STATUS ID id EXTRAARG} + assert_error {*wrong number of arguments*} {R 0 CLUSTER MIGRATION STATUS ALL EXTRAARG} + assert_error {*unknown argument*} {R 0 CLUSTER MIGRATION STATUS ABC DEF} + assert_error {*unknown argument*} {R 0 CLUSTER MIGRATION STATUS UNKNOWNARG} + # empty string id should not list any task + assert_equal {} [R 0 CLUSTER MIGRATION STATUS ID ""] + } + + test "Test TRIMSLOTS input validation" { + # Wrong number of arguments + assert_error {*wrong number of arguments*} {R 0 TRIMSLOTS} + assert_error {*wrong number of arguments*} {R 0 TRIMSLOTS RANGES} + assert_error {*wrong number of arguments*} {R 0 TRIMSLOTS RANGES 1} + assert_error {*wrong number of arguments*} {R 0 TRIMSLOTS RANGES 2 100} + assert_error {*wrong number of arguments*} {R 0 TRIMSLOTS RANGES 17000 1} + assert_error {*wrong number of arguments*} {R 0 TRIMSLOTS RANGES abc} + + # Missing ranges argument + assert_error {*missing ranges argument*} {R 0 TRIMSLOTS UNKNOWN 1 100 200} + + # Invalid number of ranges + assert_error {*invalid number of ranges*} {R 0 TRIMSLOTS RANGES 0 1 1} + assert_error {*invalid number of ranges*} {R 0 TRIMSLOTS RANGES -1 2 2} + assert_error {*invalid number of ranges*} {R 0 TRIMSLOTS RANGES 17000 1 2} + assert_error {*invalid number of ranges*} {R 0 TRIMSLOTS RANGES 2 100 200 300} + + # Invalid slot numbers + assert_error {*out of range slot*} {R 0 TRIMSLOTS RANGES 1 -1 0} + assert_error {*out of range slot*} {R 0 TRIMSLOTS RANGES 1 -2 -1} + assert_error {*out of range slot*} {R 0 TRIMSLOTS RANGES 1 0 16384} + assert_error {*out of range slot*} {R 0 TRIMSLOTS RANGES 1 abc def} + assert_error {*out of range slot*} {R 0 TRIMSLOTS RANGES 1 100 abc} + + # Start slot greater than end slot + assert_error {*greater than end slot number*} {R 0 TRIMSLOTS RANGES 1 200 100} + } + + test "Test IMPORT not allowed on replica" { + assert_error {* not allowed on replica*} {R 4 CLUSTER MIGRATION IMPORT 100 200} + } + + test "Test IMPORT not allowed during manual migration" { + set dst_id [R 1 CLUSTER MYID] + + # Set a slot to IMPORTING + R 0 CLUSTER SETSLOT 15000 IMPORTING $dst_id + assert_error {*must be STABLE to start*slot migration*} {R 0 CLUSTER MIGRATION IMPORT 100 200} + # Revert the change + R 0 CLUSTER SETSLOT 15000 STABLE + + # Same test with setting a slot to MIGRATING + R 0 CLUSTER SETSLOT 5000 MIGRATING $dst_id + assert_error {*must be STABLE to start*slot migration*} {R 0 CLUSTER MIGRATION IMPORT 100 200} + # Revert the change + R 0 CLUSTER SETSLOT 5000 STABLE + } + + test "Test IMPORT not allowed if the node is already the owner" { + assert_error {*already the owner of the slot*} {R 0 CLUSTER MIGRATION IMPORT 100 100} + } + + test "Test IMPORT not allowed for a slot without an owner" { + # Slot will have no owner + R 0 CLUSTER DELSLOTS 5000 + + assert_error {*slot has no owner: 5000*} {R 0 CLUSTER MIGRATION IMPORT 5000 5000} + + # Revert the change + R 0 CLUSTER ADDSLOTS 5000 + } + + test "Test IMPORT not allowed if slot ranges belong to different nodes" { + assert_error {*slots belong to different source nodes*} {R 0 CLUSTER MIGRATION IMPORT 7000 15000} + assert_error {*slots belong to different source nodes*} {R 0 CLUSTER MIGRATION IMPORT 7000 8000 14000 15000} + } + + test "Test IMPORT not allowed if slot is given multiple times" { + assert_error {*Slot*specified multiple times*} {R 0 CLUSTER MIGRATION IMPORT 7000 8000 8000 9000} + assert_error {*Slot*specified multiple times*} {R 0 CLUSTER MIGRATION IMPORT 7000 8000 7900 9000} + } + + test "Test CLUSTER MIGRATION STATUS ALL lists all tasks" { + # Create 3 completed tasks + R 0 CLUSTER MIGRATION IMPORT 7000 7001 + wait_for_asm_done + R 0 CLUSTER MIGRATION IMPORT 7002 7003 + wait_for_asm_done + R 0 CLUSTER MIGRATION IMPORT 7004 7005 + wait_for_asm_done + + # Get node IDs for verification + set node0_id [R 0 cluster myid] + set node1_id [R 1 cluster myid] + + # Verify CLUSTER MIGRATION STATUS ALL reply from both nodes + foreach node_idx {0 1} { + set tasks [R $node_idx CLUSTER MIGRATION STATUS ALL] + assert_equal 3 [llength $tasks] + + for {set i 0} {$i < 3} {incr i} { + set task [lindex $tasks $i] + + # Verify field order + set expected_fields {id slots source dest operation state + last_error retries create_time start_time + end_time write_pause_ms} + for {set j 0} {$j < [llength $expected_fields]} {incr j} { + set expected_field [lindex $expected_fields $j] + set actual_field [lindex $task [expr $j * 2]] + assert_equal $expected_field $actual_field + } + + # Verify basic fields + assert_equal "completed" [dict get $task state] + assert_equal "" [dict get $task last_error] + assert_equal 0 [dict get $task retries] + assert {[dict get $task write_pause_ms] >= 0} + + # Verify operation based on node + if {$node_idx == 0} { + assert_equal "import" [dict get $task operation] + } else { + assert_equal "migrate" [dict get $task operation] + } + + # Verify node IDs (all tasks: node1 -> node0) + assert_equal $node1_id [dict get $task source] + assert_equal $node0_id [dict get $task dest] + + # Verify timestamps exist and are reasonable + set create_time [dict get $task create_time] + set start_time [dict get $task start_time] + set end_time [dict get $task end_time] + assert {$create_time > 0} + assert {$start_time >= $create_time} + assert {$end_time >= $start_time} + + # Verify specific slot ranges for each task + set slots [dict get $task slots] + if {$i == 0} { + assert_equal "7004-7005" $slots + } elseif {$i == 1} { + assert_equal "7002-7003" $slots + } elseif {$i == 2} { + assert_equal "7000-7001" $slots + } + } + } + + # cleanup + R 1 CLUSTER MIGRATION IMPORT 7000 7005 + wait_for_asm_done + } + + test "Test IMPORT not allowed if there is an overlapping import" { + # Let slot migration take long time, so that we can test overlapping import + R 1 config set rdb-key-save-delay 1000000 + R 1 set tag22273 tag22273 ;# slot hash is 7000 + R 1 set tag9283 tag9283 ;# slot hash is 8000 + + set task_id [R 0 CLUSTER MIGRATION IMPORT 7000 8000] + assert_error {*overlapping import exists*} {R 0 CLUSTER MIGRATION IMPORT 8000 9000} + assert_error {*overlapping import exists*} {R 0 CLUSTER MIGRATION IMPORT 7500 8500} + assert_error {*overlapping import exists*} {R 0 CLUSTER MIGRATION IMPORT 6000 7000} + assert_error {*overlapping import exists*} {R 0 CLUSTER MIGRATION IMPORT 6500 7500} + + wait_for_condition 1000 50 { + [string match {*completed*} [migration_status 0 $task_id state]] && + [string match {*completed*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not start" + } + assert_equal "tag22273" [R 0 get tag22273] + assert_equal "tag9283" [R 0 get tag9283] + R 1 config set rdb-key-save-delay 0 + + # revert the migration + R 1 CLUSTER MIGRATION IMPORT 7000 8000 + wait_for_asm_done + } + + test "Test IMPORT with unsorted and adjacent ranges" { + # Redis should sort and merge adjacent ranges + # Adjacent means: prev.end + 1 == next.start + # e.g. 7000-7001 7002-7003 7004-7005 => 7000-7005 + + # Test with adjacent ranges + set task_id [R 0 CLUSTER MIGRATION IMPORT 7000 7001 7002 7100] + wait_for_asm_done + # verify migration is successfully completed on both nodes + assert_equal "completed" [migration_status 0 $task_id state] + assert_equal "completed" [migration_status 1 $task_id state] + # verify slot ranges are merged correctly + assert_equal "7000-7100" [migration_status 0 $task_id slots] + assert_equal "7000-7100" [migration_status 1 $task_id slots] + + # Test with unsorted and adjacent ranges + set task_id [R 1 CLUSTER MIGRATION IMPORT 7050 7051 7010 7049 7000 7005] + wait_for_asm_done + # verify migration is successfully completed on both nodes + assert_equal "completed" [migration_status 0 $task_id state] + assert_equal "completed" [migration_status 1 $task_id state] + # verify slot ranges are merged correctly + assert_equal "7000-7005 7010-7051" [migration_status 0 $task_id slots] + assert_equal "7000-7005 7010-7051" [migration_status 1 $task_id slots] + + # Another test with unsorted and adjacent ranges + set task_id [R 1 CLUSTER MIGRATION IMPORT 7007 7007 7008 7009 7006 7006] + wait_for_asm_done + # verify migration is successfully completed on both nodes + assert_equal "completed" [migration_status 0 $task_id state] + assert_equal "completed" [migration_status 1 $task_id state] + # verify slot ranges are merged correctly + assert_equal "7006-7009" [migration_status 0 $task_id slots] + assert_equal "7006-7009" [migration_status 1 $task_id slots] + } + + test "Simple slot migration with write load" { + # Perform slot migration while traffic is on and verify data consistency. + # Trimming is disabled on source nodes so, we can compare the dbs after + # migration via DEBUG DIGEST to ensure no data loss during migration. + # Steps: + # 1. Disable trimming on both nodes + # 2. Populate slot 0 on node-0 and slot 6000 on node-1 + # 2. Start write traffic on both nodes + # 3. Migrate slot 0 from node-0 to node-1 + # 4. Migrate slot 6000 from node-1 to node-0 + # 5. Stop write traffic, verify db's are identical. + + # This test runs slowly under the thread sanitizer. + # 1. Increase the lag threshold from the default 1 MB to 10 MB to let the destination catch up easily. + # 2. Increase the write pause timeout from the default 10s to 60s so the source can wait longer. + set prev_config_lag [lindex [R 0 config get cluster-slot-migration-handoff-max-lag-bytes] 1] + R 0 config set cluster-slot-migration-handoff-max-lag-bytes 10mb + R 1 config set cluster-slot-migration-handoff-max-lag-bytes 10mb + set prev_config_timeout [lindex [R 0 config get cluster-slot-migration-write-pause-timeout] 1] + R 0 config set cluster-slot-migration-write-pause-timeout 60000 + R 1 config set cluster-slot-migration-write-pause-timeout 60000 + + R 0 flushall + R 0 debug asm-trim-method none + populate_slot 10000 -idx 0 -slot 0 + + R 1 flushall + R 1 debug asm-trim-method none + populate_slot 10000 -idx 1 -slot 6000 + + # Start write traffic on node-0 + # Throws -MOVED error once asm is completed, catch block will ignore it. + catch { + # Start the slot 0 write load on the R 0 + set port [get_port 0] + set key [slot_key 0 mykey] + set load_handle0 [start_write_load "127.0.0.1" $port 100 $key 0 5] + } + + # Start write traffic on node-1 + # Throws -MOVED error once asm is completed, catch block will ignore it. + catch { + # Start the slot 6000 write load on the R 1 + set port [get_port 1] + set key [slot_key 6000 mykey] + set load_handle1 [start_write_load "127.0.0.1" $port 100 $key 0 5] + } + + # Migrate keys + R 1 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + R 0 CLUSTER MIGRATION IMPORT 6000 6100 + wait_for_asm_done + + stop_write_load $load_handle0 + stop_write_load $load_handle1 + + # verify data + assert_morethan [R 0 dbsize] 0 + assert_equal [R 0 debug digest] [R 1 debug digest] + + # cleanup + R 0 config set cluster-slot-migration-handoff-max-lag-bytes $prev_config_lag + R 0 config set cluster-slot-migration-write-pause-timeout $prev_config_timeout + R 0 debug asm-trim-method default + R 0 flushall + R 1 config set cluster-slot-migration-handoff-max-lag-bytes $prev_config_lag + R 1 config set cluster-slot-migration-write-pause-timeout $prev_config_timeout + R 1 debug asm-trim-method default + R 1 flushall + + R 1 CLUSTER MIGRATION IMPORT 6000 6100 + wait_for_asm_done + } + + test "Verify expire time is migrated correctly" { + R 0 flushall + R 1 flushall + + set string_key [slot_key 0 string_key] + set list_key [slot_key 0 list_key] + set hash_key [slot_key 0 hash_key] + set stream_key [slot_key 0 stream_key] + + for {set i 0} {$i < 20} {incr i} { + R 1 hset $hash_key $i $i + R 1 xadd $stream_key * item $i + } + for {set i 0} {$i < 2000} {incr i} { + R 1 lpush $list_key $i + } + + # set expire time of some keys + R 1 set $string_key "a" EX 1000 + R 1 EXPIRE $list_key 1000 + R 1 EXPIRE $hash_key 1000 + + # migrate slot 0-100 to R 0 + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + + # check expire times are migrated correctly + assert_range [R 0 ttl $string_key] 900 1000 + assert_range [R 0 ttl $list_key] 900 1000 + assert_range [R 0 ttl $hash_key] 900 1000 + assert_equal -1 [R 0 ttl $stream_key] + + # cleanup + R 0 flushall + R 1 flushall + R 1 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + } + + test "Slot migration with complex data types can work well" { + R 0 flushall + R 1 flushall + + set list_key [slot_key 0 list_key] + set set_key [slot_key 0 set_key] + set zset_key [slot_key 0 zset_key] + set hash_key [slot_key 0 hash_key] + set stream_key [slot_key 0 stream_key] + + # generate big keys for each data type + for {set i 0} {$i < 1000} {incr i} { + R 1 lpush $list_key $i + R 1 sadd $set_key $i + R 1 zadd $zset_key $i $i + R 1 hset $hash_key $i $i + R 1 xadd $stream_key * item $i + } + + # migrate slot 0-100 to R 0 + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + # check the data on destination node is correct + assert_equal 1000 [R 0 llen $list_key] + assert_equal 1000 [R 0 scard $set_key] + assert_equal 1000 [R 0 zcard $zset_key] + assert_equal 1000 [R 0 hlen $hash_key] + assert_equal 1000 [R 0 xlen $stream_key] + # migrate slot 0-100 to R 1 + R 1 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + } + + proc asm_basic_error_handling_test {operation channel all_states} { + foreach state $all_states { + if {$::verbose} { puts "Testing $operation $channel channel with state: $state"} + + # For states that need incremental data streaming, set a longer delay + set streaming_states [list "streaming-buffer" "accumulate-buffer" "send-bulk-and-stream" "send-stream"] + if {$state in $streaming_states} { + R 1 config set rdb-key-save-delay 1000000 + } + + # Let the destination node take time to stream buffer, so the source node will handle + # slot snapshot child process exit, and then enter "send-stream" state. + if {$state == "send-stream"} { + R 0 config set key-load-delay 100000 + } + + # Start the slot 0 write load on the R 1 + set slot0_key [slot_key 0 mykey] + set load_handle [start_write_load "127.0.0.1" [get_port 1] 100 $slot0_key 500] + + # clear old fail points and set the new fail point + assert_equal {OK} [R 0 debug asm-failpoint "" ""] + assert_equal {OK} [R 1 debug asm-failpoint "" ""] + if {$operation eq "import"} { + assert_equal {OK} [R 0 debug asm-failpoint "import-$channel-channel" $state] + } elseif {$operation eq "migrate"} { + assert_equal {OK} [R 1 debug asm-failpoint "migrate-$channel-channel" $state] + } else { + fail "Unknown operation: $operation" + } + + # Start the migration + set task_id [R 0 CLUSTER MIGRATION IMPORT 0 100] + + # The task should be failed due to the fail point + wait_for_condition 2000 10 { + [string match -nocase "*$channel*${state}*" [migration_status 0 $task_id last_error]] || + [string match -nocase "*$channel*${state}*" [migration_status 1 $task_id last_error]] + } else { + fail "ASM task did not fail with expected error - + (dst: [migration_status 0 $task_id last_error] + src: [migration_status 1 $task_id last_error] + expected: $channel $state)" + } + stop_write_load $load_handle + + # Cancel the task + R 0 CLUSTER MIGRATION CANCEL ID $task_id + R 1 CLUSTER MIGRATION CANCEL ID $task_id + + R 1 config set rdb-key-save-delay 0 + R 0 config set key-load-delay 0 + } + } + + test "Destination node main channel basic error-handling tests " { + set all_states [list \ + "connecting" \ + "auth-reply" \ + "handshake-reply" \ + "syncslots-reply" \ + "accumulate-buffer" \ + "streaming-buffer" \ + "wait-stream-eof" \ + ] + asm_basic_error_handling_test "import" "main" $all_states + } + + test "Destination node rdb channel basic error-handling tests" { + set all_states [list \ + "connecting" \ + "auth-reply" \ + "rdbchannel-reply" \ + "rdbchannel-transfer" \ + ] + asm_basic_error_handling_test "import" "rdb" $all_states + } + + test "Source node main channel basic error-handling tests " { + set all_states [list \ + "wait-rdbchannel" \ + "send-bulk-and-stream" \ + "send-stream" \ + "handoff" \ + ] + asm_basic_error_handling_test "migrate" "main" $all_states + } + + test "Source node rdb channel basic error-handling tests" { + set all_states [list \ + "wait-bgsave-start" \ + "send-bulk-and-stream" \ + ] + asm_basic_error_handling_test "migrate" "rdb" $all_states + } + + test "Migration will be successful after fail points are cleared" { + R 0 flushall + R 1 flushall + set slot0_key [slot_key 0 mykey] + set slot1_key [slot_key 1 mykey] + R 1 set $slot0_key "a" + R 1 set $slot1_key "b" + + # we set a delay to write incremental data + R 1 config set rdb-key-save-delay 1000000 + + # Start the slot 0 write load on the R 1 + set load_handle [start_write_load "127.0.0.1" [get_port 1] 100 $slot0_key] + + # Clear all fail points + assert_equal {OK} [R 0 debug asm-failpoint "" ""] + assert_equal {OK} [R 1 debug asm-failpoint "" ""] + + # Start the migration + set task_id [R 0 CLUSTER MIGRATION IMPORT 0 100] + + # Wait for the migration to complete + wait_for_asm_done + + stop_write_load $load_handle + + # Verify the data is migrated, slot 0 and 1 should belong to R 1 + # slot 0 key should be changed by the write load + assert_not_equal "a" [R 0 get $slot0_key] + assert_equal "b" [R 0 get $slot1_key] + R 1 config set rdb-key-save-delay 0 + } + + test "Client output buffer limit is reached on source side" { + R 0 flushall + R 1 flushall + set r1_pid [S 1 process_id] + R 1 debug repl-pause on-streaming-repl-buf + + # Set a small output buffer limit to trigger the error + R 0 config set client-output-buffer-limit "replica 4mb 0 0" + + set task_id [setup_slot_migration_with_delay 0 1 0 100] + + # some write traffic is to have chance to enter streaming buffer state + set slot0_key [slot_key 0 mykey] + R 0 set $slot0_key "a" + + # after 3 second, the slots snapshot (costs 2s to generate) should be transferred, + # then start streaming buffer + after 3000 + + set loglines [count_log_lines 0] + + # Start the slot 0 write load on the R 0 + set load_handle [start_write_load "127.0.0.1" [get_port 0] 100 $slot0_key 1000] + + # verify the metric is accessible, it is transient, will be reset on disconnect + assert {[S 0 mem_cluster_slot_migration_output_buffer] >= 0} + + # After some time, the client output buffer limit should be reached + wait_for_log_messages 0 {"*Client * closed * for overcoming of output buffer limits.*"} $loglines 1000 10 + wait_for_condition 1000 10 { + [string match {*send*stream*} [migration_status 0 $task_id last_error]] + } else { + fail "ASM task did not fail as expected" + } + + stop_write_load $load_handle + + # Reset configurations + R 0 config set client-output-buffer-limit "replica 0 0 0" + R 0 config set rdb-key-save-delay 0 + + # resume server and clear pause point + resume_process $r1_pid + R 1 debug repl-pause clear + + # Wait for the migration to complete + wait_for_asm_done + } + + test "Full sync buffer limit is reached on destination side" { + # Set a small replication buffer limit to trigger the error + R 0 config set replica-full-sync-buffer-limit 1mb + + # start migration from 1 to 0, cost 4s to transfer slots snapshot + set task_id [setup_slot_migration_with_delay 1 0 0 100 2 2000000] + set loglines [count_log_lines 0] + + # Create some traffic on slot 0 + populate_slot 100 -idx 1 -slot 0 -size 100000 + + # After some time, slots sync buffer limit should be reached, but migration would not fail + # since the buffer will be accumulated on source side from now. + wait_for_log_messages 0 {"*Slots sync buffer limit has been reached*"} $loglines 1000 10 + + # verify the peak value, should be greater than 1mb + assert {[S 0 mem_cluster_slot_migration_input_buffer_peak] > 1000000} + # verify the metric is accessible, it is transient, will be reset on disconnect + assert {[S 0 mem_cluster_slot_migration_input_buffer] >= 0} + + wait_for_asm_done + + # Reset configurations + R 0 config set replica-full-sync-buffer-limit 0 + R 1 config set rdb-key-save-delay 0 + R 1 cluster migration import 0 100 + wait_for_asm_done + } + + test "Expired key is not deleted and SCAN/KEYS/RANDOMKEY/CLUSTER GETKEYSINSLOT filter keys in importing slots" { + set slot0_key [slot_key 0 mykey] + set slot1_key [slot_key 1 mykey] + set slot2_key [slot_key 2 mykey] + R 1 flushall + R 0 flushall + + # we set a delay to write incremental data + R 1 config set rdb-key-save-delay 1000000 + + # set expire time 2s. Generating slot snapshot will 3s, so these + # three keys will be expired after slot snapshot is transferred + R 1 setex $slot0_key 2 "a" + R 1 setex $slot1_key 2 "b" + R 1 hset $slot2_key "f1" "1" + R 1 expire $slot2_key 2 + R 1 hexpire $slot2_key 2 FIELDS 1 "f1" + + set task_id [R 0 CLUSTER MIGRATION IMPORT 0 100] + wait_for_condition 2000 10 { + [string match {*send-bulk-and-stream*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not start" + } + + # update expire time during mirgration + R 1 setex $slot0_key 100 "a" + R 1 expire $slot1_key 80 + R 1 expire $slot2_key 60 + R 1 hincrbyfloat $slot2_key "f1" 1 + R 1 hexpire $slot2_key 60 FIELDS 1 "f1" + + # after 2s, at least a key should be transferred, and should not be deleted + # due to expired, neither active nor lazy expiration (SCAN) takes effect, + # Besides SCAN/KEYS/RANDOMKEY/CLUSTER GETKEYSINSLOT command can not find them + after 2000 + R 3 readonly + foreach id {0 3} { ;# 0 is the master, 3 is the replica + assert_equal {0 {}} [R $id scan 0 count 10] + assert_equal {} [R $id keys "*"] + assert_equal {} [R $id keys "{06S}*"] + assert_equal {} [R $id randomkey] + assert_equal {} [R $id cluster getkeysinslot 0 100] + assert_equal [R $id cluster countkeysinslot 0] 0 + assert_equal [R $id dbsize] 0 + + # but we can see the number of keys is increased in INFO KEYSPACE + assert {[scan [regexp -inline {keys\=([\d]*)} [R $id info keyspace]] keys=%d] >= 1} + assert {[scan [regexp -inline {expires\=([\d]*)} [R $id info keyspace]] expires=%d] >= 1} + } + + wait_for_asm_done + + wait_for_ofs_sync [Rn 0] [Rn 3] + + foreach id {0 3} { ;# 0 is the master, 3 is the replica + # verify the keys are valid + assert_range [R $id ttl $slot0_key] 90 100 + assert_range [R $id ttl $slot1_key] 70 80 + assert_range [R $id ttl $slot2_key] 50 60 + assert_range [R $id httl $slot2_key FIELDS 1 "f1"] 50 60 + + # KEYS/SCAN/RANDOMKEY/CLUSTER GETKEYSINSLOT will find the keys after migration + assert_equal [list 0 [list $slot0_key $slot1_key $slot2_key]] [R $id scan 0 count 10] + assert_equal [list $slot0_key $slot1_key $slot2_key] [R $id keys "*"] + assert_equal [list $slot0_key] [R $id keys "{06S}*"] + assert_not_equal {} [R $id randomkey] + assert_equal [list $slot0_key] [R $id cluster getkeysinslot 0 100] + + # INFO KEYSPACE/DBSIZE/CLUSTER COUNTKEYSINSLOT will also reflect the keys + assert_equal 3 [scan [regexp -inline {keys\=([\d]*)} [R $id info keyspace]] keys=%d] + assert_equal 3 [scan [regexp -inline {expires\=([\d]*)} [R $id info keyspace]] expires=%d] + assert_equal 1 [scan [regexp -inline {subexpiry\=([\d]*)} [R $id info keyspace]] subexpiry=%d] + assert_equal 3 [R $id dbsize] + assert_equal 1 [R $id cluster countkeysinslot 0] + } + + # update expire time to 10ms, after some time, the keys should be deleted due to + # active expiration + R 0 pexpire $slot0_key 10 + R 0 pexpire $slot1_key 10 + R 0 hpexpire $slot2_key 10 FIELDS 1 "f1" ;# the last field is expired, the key will be deleted + wait_for_condition 100 50 { + [scan [regexp -inline {keys\=([\d]*)} [R 0 info keyspace]] keys=%d] == {} && + [scan [regexp -inline {keys\=([\d]*)} [R 3 info keyspace]] keys=%d] == {} + } else { + fail "keys did not expire" + } + + R 1 config set rdb-key-save-delay 0 + } + + test "Eviction does not evict keys in importing slots" { + set slot0_key [slot_key 0 mykey] + set slot1_key [slot_key 1 mykey] + set slot2_key [slot_key 2 mykey] + set slot5462_key [slot_key 5462 mykey] + set slot5463_key [slot_key 5463 mykey] + R 1 flushall + R 0 flushall + + # we set a delay to write incremental data + R 0 config set rdb-key-save-delay 1000000 + + set 1k_str [string repeat "a" 1024] + set 1m_str [string repeat "a" 1048576] + + # set two keys to be evicted + R 1 set $slot5462_key $1k_str + R 1 set $slot5463_key $1k_str + + # set maxmemory to 200kb more than current used memory, + # redis should evict some keys if importing some big keys + set r1_mem_used [S 1 used_memory] + set r1_max_mem [expr {$r1_mem_used + 200*1024}] + R 1 config set maxmemory $r1_max_mem + R 1 config set maxmemory-policy allkeys-lru + + # set 3 keys to be migrated + R 0 set $slot0_key $1m_str + R 0 set $slot1_key $1m_str + R 0 set $slot2_key $1m_str + + set task_id [R 1 CLUSTER MIGRATION IMPORT 0 100] + wait_for_condition 2000 10 { + [string match {*send-bulk-and-stream*} [migration_status 0 $task_id state]] + } else { + fail "ASM task did not start" + } + + # after 2.2s, at least two keys should be transferred, they should not be evicted + # but other keys (slot5462_key and slot5463_key) should be evicted + after 2200 + for {set j 0} {$j < 100} {incr j} { R 1 ping } ;# trigger eviction + assert_equal 0 [R 1 exists $slot5462_key] + assert_equal 0 [R 1 exists $slot5463_key] + assert {[scan [regexp -inline {keys\=([\d]*)} [R 1 info keyspace]] keys=%d] >= 2} + + # current used memory should be more than the maxmemory, since the big keys that + # belong importing slots can not be evicted. + set r1_mem_used [S 1 used_memory] + assert {$r1_mem_used > $r1_max_mem + 1024*1024} + + wait_for_asm_done + + # after migration, these big keys should be evicted + for {set j 0} {$j < 100} {incr j} { R 1 ping } ;# trigger eviction + assert_equal {} [scan [regexp -inline {expires\=([\d]*)} [R 1 info keyspace]] expires=%d] + } + + test "Failover will cancel slot migration tasks" { + # migrate slot 0-100 from 1 to 0 + set task_id [setup_slot_migration_with_delay 1 0 0 100] + + # FAILOVER happens on the destination node, instance #3 become master, #0 become slave + failover_and_wait_for_done 3 + + # the old master will cancel the importing task, and the migrating task on + # the source node will be failed + wait_for_condition 1000 50 { + [string match {*canceled*} [migration_status 0 $task_id state]] && + [string match {*failover*} [migration_status 0 $task_id last_error]] && + [string match {*failed*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not cancel" + } + + # We can restart ASM tasks on new master, migrate slot 0-100 from 1 to 3 + R 1 config set rdb-key-save-delay 0 + set task_id [R 3 CLUSTER MIGRATION IMPORT 0 100] + wait_for_asm_done + + # migrate slot 0-100 from 3 to 1 + set task_id [setup_slot_migration_with_delay 3 1 0 100] + + # FAILOVER happens on the source node, instance #3 become slave, #0 become master + failover_and_wait_for_done 0 + + # the old master will cancel the migrating task, but the destination node will + # retry the importing task, and then succeed. + wait_for_condition 1000 50 { + [string match {*canceled*} [migration_status 3 $task_id state]] + } else { + fail "ASM task did not cancel" + } + wait_for_asm_done + } + + test "Flush-like command can cancel slot migration task" { + # flushall, flushdb + foreach flushcmd {flushall flushdb} { + # start slot migration from 1 to 0 + set task_id [setup_slot_migration_with_delay 1 0 0 100] + + if {$::verbose} { puts "Testing flush command: $flushcmd"} + R 0 $flushcmd + + # flush-like will cancel the task + wait_for_condition 1000 50 { + [string match {*canceled*} [migration_status 0 $task_id state]] + } else { + fail "ASM task did not cancel" + } + } + + R 1 config set rdb-key-save-delay 0 + R 0 cluster migration import 0 100 + wait_for_asm_done + } + + test "CLUSTER SETSLOT command when there is a slot migration task" { + # Setup slot migration test from node 0 to node 1 + set task_id [setup_slot_migration_with_delay 0 1 0 100] + + # Cluster SETSLOT command is not allowed when there is a slot migration task + # on the slot. #0 and #1 are having migration task now. + foreach instance {0 1} { + set node_id [R $instance cluster myid] + + catch {R $instance cluster setslot 0 migrating $node_id} err + assert_match {*in an active atomic slot migration*} $err + + catch {R $instance cluster setslot 0 importing $node_id} err + assert_match {*in an active atomic slot migration*} $err + + catch {R $instance cluster setslot 0 stable} err + assert_match {*in an active atomic slot migration*} $err + + catch {R $instance cluster setslot 0 node $node_id} err + assert_match {*in an active atomic slot migration*} $err + } + + # CLUSTER SETSLOT on other node will cancel the migration task, we update + # the owner of slot 0 (that is migrating from #0 to #1) to #2 on #2, we + # bump the config epoch to make sure the change can update #0 and #1 + # slot configuration, so #0 and #1 will cancel the migration task. + # BTW, if config epoch is not bumped, the slot config of #2 may be + # updated by #0 and #1. + R 2 cluster bumpepoch + R 2 cluster setslot 0 node [R 2 cluster myid] + wait_for_condition 1000 50 { + [string match {*canceled*} [migration_status 0 $task_id state]] && + [string match {*slots configuration updated*} [migration_status 0 $task_id last_error]] && + [string match {*canceled*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not cancel" + } + + # set slot 0 back to #0 + R 0 cluster bumpepoch + R 0 cluster setslot 0 node [R 0 cluster myid] + wait_for_cluster_propagation + wait_for_cluster_state "ok" + } + + test "CLUSTER DELSLOTSRANGE command cancels a slot migration task" { + # start slot migration from 0 to 1 + set task_id [setup_slot_migration_with_delay 0 1 0 100] + + R 0 cluster delslotsrange 0 100 + wait_for_condition 1000 50 { + [string match {*canceled*} [migration_status 0 $task_id state]] && + [string match {*slots configuration updated*} [migration_status 0 $task_id last_error]] && + [string match {*failed*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not cancel" + } + R 1 cluster migration cancel id $task_id + + # add the slots back + R 0 cluster addslotsrange 0 100 + wait_for_cluster_propagation + wait_for_cluster_state "ok" + } + + # NOTE: this test needs more than 60s, maybe you can skip when testing + test "CLUSTER FORGET command cancels a slot migration task" { + R 0 config set rdb-key-save-delay 0 + # Migrate all slot on #0 to #1, so we can forget #0 + set task_id [R 1 CLUSTER MIGRATION IMPORT 0 5461] + wait_for_asm_done + + # start slot migration from 1 to 0 + set task_id [setup_slot_migration_with_delay 1 0 0 5461] + + # Forget #0 on #1, the migration task on #1 will be canceled due to node deleted, + # and the importing task on #0 will be failed + R 1 cluster forget [R 0 cluster myid] + wait_for_condition 1000 50 { + [string match {*canceled*} [migration_status 1 $task_id state]] && + [string match {*node deleted*} [migration_status 1 $task_id last_error]] && + [string match {*failed*} [migration_status 0 $task_id state]] + } else { + fail "ASM task did not cancel" + } + + # Add #0 back into cluster + # NOTE: this will cost 60s to let #0 join the cluster since + # other nodes add #0 into black list for 60s after FORGET. + R 1 config set rdb-key-save-delay 0 + R 1 cluster meet "127.0.0.1" [lindex [R 0 config get port] 1] + + # the importing task on #0 will be retried, and eventually succeed + # since now #0 is back in the cluster + wait_for_condition 3000 50 { + [string match {*completed*} [migration_status 0 $task_id state]] && + [string match {*completed*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not finish" + } + + # make sure #0 is completely back to the cluster + wait_for_cluster_propagation + wait_for_cluster_state "ok" + } + + test "CLIENT PAUSE can cancel slot migration task" { + # start slot migration from 0 to 1 + set task_id [setup_slot_migration_with_delay 0 1 0 100] + + # CLIENT PAUSE happens on the destination node, #1 will cancel the importing task + R 1 client pause 100000 write ;# pause 100s + wait_for_condition 1000 50 { + [string match {*canceled*} [migration_status 1 $task_id state]] && + [string match {*client pause*} [migration_status 1 $task_id last_error]] + } else { + fail "ASM task did not cancel" + } + + # start task again + set task_id [R 1 CLUSTER MIGRATION IMPORT 0 100] + after 200 ;# give some time to have chance to schedule the task + # the task should not start since server is paused + assert {[string match {*none*} [migration_status 1 $task_id state]]} + + # unpause the server, the task should start + R 1 client unpause + wait_for_asm_done + + # migrate back to original node #0 + R 0 config set rdb-key-save-delay 0 + R 1 config set rdb-key-save-delay 0 + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + } + + test "Server shutdown can cancel slot migration task, exit with success" { + # start slot migration from 0 to 1 + setup_slot_migration_with_delay 0 1 0 100 + + set loglines [count_log_lines -1] + + # Shutdown the server, it should cancel the migration task + restart_server -1 true false true nosave + + wait_for_log_messages -1 {"*Cancelled due to server shutdown*"} $loglines 100 100 + + wait_for_cluster_propagation + wait_for_cluster_state "ok" + } + + test "Cancel import task when streaming buffer into db" { + # set a delay to have time to cancel import task that is streaming buf to db + R 1 config set key-load-delay 50000 + # start slot migration from 0 to 1 + set task_id [setup_slot_migration_with_delay 0 1 0 100 5] + + # start the slot 0 write load on the node 0 + set slot0_key [slot_key 0 mykey] + set load_handle [start_write_load "127.0.0.1" [get_port 0] 100 $slot0_key 500] + + # wait for entering streaming buffer state + wait_for_condition 1000 10 { + [string match {*streaming-buffer*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not enter streaming buffer state" + } + stop_write_load $load_handle + + # cancel the import task on #1, the destination node works fine + R 1 cluster migration cancel id $task_id + assert_match {*canceled*} [migration_status 1 $task_id state] + + # reset config + R 0 config set key-load-delay 0 + R 1 config set key-load-delay 0 + } + + test "Destination node main channel timeout when waiting stream EOF" { + set task_id [setup_slot_migration_with_delay 0 1 0 100] + R 1 config set repl-timeout 5 + + # pause the source node to make EOF wait timeout. Do not pause + # the child process, so it can deliver slot snapshot to destination + set r0_process_id [S 0 process_id] + pause_process $r0_process_id + + # the destination node will fail after 7s, 5s for EOF wait and 2s for slot snapshot + wait_for_condition 1000 20 { + [string match {*failed*} [migration_status 1 $task_id state]] && + [string match {*Main channel*Connection timeout*wait-stream-eof*} \ + [migration_status 1 $task_id last_error]] + } else { + fail "ASM task did not fail" + } + + # resume the source node + resume_process $r0_process_id + + # After the source node is resumed, the task on source node may receive + # ACKs from destination and consider the task is stream-done. In this case, + # the task on source node will be failed after several seconds + if {[string match {*stream-done*} [migration_status 0 $task_id state]]} { + wait_for_condition 1000 20 { + [string match {*failed*} [migration_status 0 $task_id state]] && + [string match {*Server paused*} [migration_status 0 $task_id last_error]] + } else { + fail "ASM task did not fail" + } + } + + R 1 config set repl-timeout 60 + R 0 cluster migration cancel id $task_id + R 1 cluster migration cancel id $task_id + } + + test "Destination node rdb channel timeout when transferring slots snapshot" { + # cost 10s to transfer each key + set task_id [setup_slot_migration_with_delay 0 1 0 100 2 10000000] + R 1 config set repl-timeout 3 + + # the destination node will fail after 3s + wait_for_condition 1000 20 { + [string match {*failed*} [migration_status 1 $task_id state]] && + [string match {*RDB channel*Connection timeout*rdbchannel-transfer*} \ + [migration_status 1 $task_id last_error]] + } else { + fail "ASM task did not fail" + } + + R 1 config set repl-timeout 60 + R 0 cluster migration cancel id $task_id + R 1 cluster migration cancel id $task_id + } + + test "Source node rdb channel timeout when transferring slots snapshot" { + set r1_pid [S 1 process_id] + R 0 flushall + R 0 config set save "" + # generate several large keys, make sure the memory usage is more than + # socket buffer size, so the rdb channel will block and timeout if + # no data is received by destination. + set val [string repeat "a" 102400] ;# 100kb + for {set i 0} {$i < 1000} {incr i} { + set key [slot_key 0 "key$i"] + R 0 set $key $val + } + R 0 config set repl-timeout 3 ;# 3s for rdb channel timeout + R 0 config set rdb-key-save-delay 10000 ;# 1000 keys cost 10s to save + + # start migration from #0 to #1 + set task_id [R 1 CLUSTER MIGRATION IMPORT 0 100] + wait_for_condition 1000 20 { + [string match {*send-bulk-and-stream*} [migration_status 0 $task_id state]] + } else { + fail "ASM task did not start" + } + + # pause the destination node to make rdb channel timeout + pause_process $r1_pid + + # the source node will fail, the rdb child process can not + # write data to destination, so it will timeout + wait_for_condition 1000 30 { + [string match {*failed*} [migration_status 0 $task_id state]] && + [string match {*RDB channel*Failed to send slots snapshot*} \ + [migration_status 0 $task_id last_error]] + } else { + fail "ASM task did not fail" + } + resume_process $r1_pid + + R 0 config set repl-timeout 60 + R 0 cluster migration cancel id $task_id + R 1 cluster migration cancel id $task_id + } + + test "Source node main channel timeout when sending incremental stream" { + R 0 flushall + R 0 config set repl-timeout 2 ;# 2s for main channel timeout + + set r1_pid [S 1 process_id] + # in order to have time to pause the destination node + R 1 config set key-load-delay 50000 ;# 50ms each 16k data + + # start migration from #0 to #1 + set task_id [setup_slot_migration_with_delay 0 1 0 100] + + # Create 200 keys of 16k size traffic on slot 0, streaming buffer need 10s (200*50ms) + populate_slot 200 -idx 0 -slot 0 -size 16384 + + # wait for streaming buffer state, then pause the destination node + wait_for_condition 1000 20 { + [string match {*streaming-buffer*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not stream buffer, state: [migration_status 1 $task_id state]" + } + pause_process $r1_pid + + # Start the slot 0 write load on the R 0 + set load_handle [start_write_load "127.0.0.1" [get_port 0] 100 [slot_key 0 mykey] 500] + + # the source node will fail after several seconds (including the time + # to fill the socket buffer of source node), the main channel can not + # write data to destination since the destination is paused + wait_for_condition 1000 30 { + [string match {*failed*} [migration_status 0 $task_id state]] && + [string match {*Main channel*Connection timeout*} \ + [migration_status 0 $task_id last_error]] + } else { + fail "ASM task did not fail" + } + stop_write_load $load_handle + resume_process $r1_pid + + R 0 config set repl-timeout 60 + R 1 config set key-load-delay 0 + R 0 cluster migration cancel id $task_id + R 1 cluster migration cancel id $task_id + R 0 flushall + } + + test "Source server paused timeout" { + # set timeout to 0, so the task will fail immediately when checking timeout + R 0 config set cluster-slot-migration-write-pause-timeout 0 + + # start migration from node 0 to 1 + set task_id [setup_slot_migration_with_delay 0 1 0 100] + + # start the slot 0 write load on the node 0 + set slot0_key [slot_key 0 mykey] + set load_handle [start_write_load "127.0.0.1" [get_port 0] 100 $slot0_key] + + # node 0 will fail since server paused timeout + wait_for_condition 2000 10 { + [string match {*failed*} [migration_status 0 $task_id state]] && + [string match {*Server paused timeout*} \ + [migration_status 0 $task_id last_error]] + } else { + fail "ASM task did not fail" + } + + stop_write_load $load_handle + + # reset config + R 0 config set cluster-slot-migration-write-pause-timeout 10000 + R 0 cluster migration cancel id $task_id + R 1 cluster migration cancel id $task_id + } + + test "Sync buffer drain timeout" { + # set a fail point to avoid the source node to enter handoff prep state + # to test the sync buffer drain timeout + R 0 debug asm-failpoint "migrate-main-channel" "handoff-prep" + R 0 config set cluster-slot-migration-sync-buffer-drain-timeout 5000 + + set r1_pid [S 1 process_id] + + # start migration from node 0 to 1 + set task_id [setup_slot_migration_with_delay 0 1 0 100] + + # start the slot 0 write load on the node 0 + set slot0_key [slot_key 0 mykey] + set load_handle [start_write_load "127.0.0.1" [get_port 0] 100 $slot0_key] + + # wait for entering streaming buffer state + wait_for_condition 1000 10 { + [string match {*wait-stream-eof*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not enter wait-stream-eof state" + } + + pause_process $r1_pid ;# avoid the destination to apply commands + + # node 0 will fail since sync buffer drain timeout + wait_for_condition 2000 10 { + [string match {*failed*} [migration_status 0 $task_id state]] && + [string match {*Sync buffer drain timeout*} \ + [migration_status 0 $task_id last_error]] + } else { + fail "ASM task did not fail" + } + + stop_write_load $load_handle + resume_process $r1_pid + + # reset config + R 0 config set cluster-slot-migration-sync-buffer-drain-timeout 60000 + R 0 debug asm-failpoint "" "" + R 0 cluster migration cancel id $task_id + R 1 cluster migration cancel id $task_id + } + + test "Cluster implementation cannot start migrate task temporarily" { + # Inject a fail point to make the source node not ready + R 0 debug asm-failpoint "migrate-main-channel" "none" + + # start migration from node 0 to 1 + set task_id [R 1 CLUSTER MIGRATION IMPORT 0 100] + + # verify source node replies SYNCSLOTS with -NOTREADY + set loglines [count_log_lines -1] + wait_for_log_messages -1 {"*Source node replied to SYNCSLOTS SYNC with -NOTREADY, will retry later*"} $loglines 100 100 + + # clear the fail point and verify the task is completed + R 0 debug asm-failpoint "" "" + wait_for_asm_done + assert_equal "completed" [migration_status 0 $task_id state] + assert_equal "completed" [migration_status 1 $task_id state] + + # cleanup + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + } +} + +start_cluster 3 3 {tags {external:skip cluster} overrides {cluster-node-timeout 60000 cluster-allow-replica-migration no}} { + test "Test bgtrim after a successful migration" { + R 0 debug asm-trim-method bg + R 3 debug asm-trim-method bg + R 0 CONFIG RESETSTAT + R 3 CONFIG RESETSTAT + + R 0 flushall + # Fill slot 0 + populate_slot 1000 -idx 0 -slot 0 + # Fill slot 1 with keys that have TTL + populate_slot 1000 -idx 0 -slot 1 -prefix "expirekey" -expires 100 + # HFE key on slot 2 + set slot2_hfekey [slot_key 2 hfekey] + R 0 HSETEX $slot2_hfekey EX 10 FIELDS 1 f1 v1 + + # Fill slot 101, these keys won't be migrated + populate_slot 1000 -idx 0 -slot 101 + # Fill slot 102 with keys that have TTL + populate_slot 1000 -idx 0 -slot 102 -prefix "expirekey" -expires 100 + # HFE key on slot 103 + set slot103_hfekey [slot_key 103 hfekey] + R 0 HSETEX $slot103_hfekey EX 10 FIELDS 1 f1 v1 + + # migrate slot 0 to node-1 + R 1 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + + # Verify the data is migrated + wait_for_ofs_sync [Rn 0] [Rn 3] + assert_equal 2001 [R 0 dbsize] + assert_equal 2001 [R 3 dbsize] + wait_for_ofs_sync [Rn 1] [Rn 4] + assert_equal 2001 [R 1 dbsize] + assert_equal 2001 [R 4 dbsize] + + # Verify the keys are trimmed lazily + wait_for_condition 1000 10 { + [S 0 lazyfreed_objects] == 2001 && + [S 3 lazyfreed_objects] == 2001 + } else { + puts "lazyfreed_objects: [S 0 lazyfreed_objects] [S 3 lazyfreed_objects]" + fail "Background trim did not happen" + } + + # Cleanup + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + R 0 flushall + R 0 debug asm-trim-method default + R 3 debug asm-trim-method default + } + + test "Test bgtrim after a failed migration" { + R 0 debug asm-trim-method bg + R 3 debug asm-trim-method bg + R 1 CONFIG RESETSTAT + R 4 CONFIG RESETSTAT + + # Fill slot 0 on node-0 and migrate it to node-1 (with some delay) + R 0 flushall + set task_id [setup_slot_migration_with_delay 0 1 0 100 10000 1000] + after 1000 ;# wait some time so that some keys are moved + + # Fail the migration + R 1 CLUSTER MIGRATION CANCEL ID $task_id + wait_for_asm_done + + # Verify the data is not migrated + assert_equal 10000 [R 0 dbsize] + assert_equal 10000 [R 3 dbsize] + + # Verify the keys are trimmed lazily after a failed import on dest side. + wait_for_condition 1000 20 { + [R 1 dbsize] == 0 && + [R 4 dbsize] == 0 && + [S 1 lazyfreed_objects] > 0 && + [S 4 lazyfreed_objects] > 0 + } else { + fail "Background trim did not happen" + } + + # Cleanup + wait_for_asm_done + R 0 flushall + R 0 debug asm-trim-method default + R 3 debug asm-trim-method default + } + + test "Test bgtrim unblocks stream client" { + # Two clients waiting for data on two different streams which are in + # different slots. We are going to migrate one slot, which will unblock + # the client. The other client should still be blocked. + R 0 debug asm-trim-method bg + + set key0 [slot_key 0 mystream] + set key1 [slot_key 1 mystream] + + # First client waits on slot-0 key + R 0 DEL $key0 + R 0 XADD $key0 666 f v + R 0 XGROUP CREATE $key0 mygroup $ + set rd0 [redis_deferring_client] + $rd0 XREADGROUP GROUP mygroup Alice BLOCK 0 STREAMS $key0 ">" + wait_for_blocked_clients_count 1 + + # Second client waits on slot-1 key + R 0 DEL $key1 + R 0 XADD $key1 666 f v + R 0 XGROUP CREATE $key1 mygroup $ + set rd1 [redis_deferring_client] + $rd1 XREADGROUP GROUP mygroup Alice BLOCK 0 STREAMS $key1 ">" + wait_for_blocked_clients_count 2 + + # Migrate slot 0 + R 1 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + + # First client should get MOVED error + assert_error "*MOVED*" {$rd0 read} + $rd0 close + + # Second client should operate normally + R 0 XADD $key1 667 f v + set res [$rd1 read] + assert_equal [lindex $res 0 1 0] {667-0 {f v}} + $rd1 close + + # cleanup + wait_for_asm_done + R 0 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + R 0 flushall + R 0 debug asm-trim-method default + } + + test "Test bgtrim touches watched keys" { + R 0 debug asm-trim-method bg + + # bgtrim should touch watched keys on migrated slots + set key0 [slot_key 0 key] + R 0 set $key0 30 + R 0 watch $key0 + R 1 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + R 0 multi + R 0 ping + assert_equal {} [R 0 exec] + + # bgtrim should not touch watched keys on other slots + set key2 [slot_key 2 key] + R 0 set $key2 30 + R 0 watch $key2 + R 1 CLUSTER MIGRATION IMPORT 1 1 + wait_for_asm_done + R 0 multi + R 0 ping + assert_equal PONG [R 0 exec] + + # cleanup + wait_for_asm_done + R 0 CLUSTER MIGRATION IMPORT 0 1 + wait_for_asm_done + R 0 flushall + R 0 debug asm-trim-method default + } + + test "Test bgtrim after a FAILOVER on destination side" { + R 1 debug asm-trim-method bg + R 4 debug asm-trim-method bg + + set loglines [count_log_lines -4] + + # Fill slot 0 on node-0 and migrate it to node-1 (with some delay) + R 0 flushall + set task_id [setup_slot_migration_with_delay 0 1 0 100 10000 1000] + after 1000 ;# wait some time so that some keys are moved + + # Trigger a failover with force to simulate unreachable master and + # verify unowned keys are trimmed once replica becomes master. + failover_and_wait_for_done 4 force + wait_for_log_messages -4 {"*Detected keys in slots that do not belong*Scheduling trim*"} $loglines 1000 10 + wait_for_condition 1000 10 { + [R 1 dbsize] == 0 && + [R 4 dbsize] == 0 + } else { + fail "Background trim did not happen" + } + + # cleanup + wait_for_cluster_propagation + failover_and_wait_for_done 1 + R 0 config set rdb-key-save-delay 0 + R 1 debug asm-trim-method default + R 4 debug asm-trim-method default + wait_for_asm_done + } + + test "CLUSTER SETSLOT is not allowed if there is a pending trim job" { + R 0 debug asm-trim-method bg + R 3 debug asm-trim-method bg + + # Fill slot 0 on node-0 and migrate it to node-1 (with some delay) + R 0 flushall + set task_id [setup_slot_migration_with_delay 0 1 0 100 10000 1000] + + # Pause will cancel the task and there will be a pending trim job + # until writes are allowed again. + R 1 client pause 100000 write ;# pause 100s + wait_for_asm_done + + # CLUSTER SETSLOT is not allowed if there is a pending trim job. + assert_error {*There is a pending trim job for slot 0*} {R 1 CLUSTER SETSLOT 0 STABLE} + + # Unpause the server, trim will be triggered and SETSLOT will be allowed + R 1 client unpause + R 1 CLUSTER SETSLOT 0 STABLE + } +} + +start_cluster 3 3 {tags {external:skip cluster} overrides {cluster-node-timeout 60000 cluster-allow-replica-migration no save ""}} { + test "Test active trim after a successful migration" { + R 0 debug asm-trim-method active + R 3 debug asm-trim-method active + populate_slot 500 -slot 0 + populate_slot 500 -slot 1 + populate_slot 500 -slot 3 + populate_slot 500 -slot 4 + + # Migrate 1500 keys + R 1 CLUSTER MIGRATION IMPORT 0 1 3 3 + wait_for_asm_done + + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 0 && + [CI 0 cluster_slot_migration_active_trim_current_job_trimmed] == 1500 && + [CI 3 cluster_slot_migration_active_trim_running] == 0 && + [CI 3 cluster_slot_migration_active_trim_current_job_trimmed] == 1500 + } else { + fail "trim failed" + } + + assert_equal 1500 [CI 0 cluster_slot_migration_active_trim_current_job_keys] + assert_equal 1500 [CI 3 cluster_slot_migration_active_trim_current_job_keys] + + assert_equal 500 [R 0 dbsize] + assert_equal 500 [R 3 dbsize] + assert_equal 1500 [R 1 dbsize] + assert_equal 1500 [R 4 dbsize] + assert_equal 0 [R 0 cluster countkeysinslot 0] + assert_equal 0 [R 0 cluster countkeysinslot 1] + assert_equal 0 [R 0 cluster countkeysinslot 3] + assert_equal 500 [R 0 cluster countkeysinslot 4] + + # cleanup + R 0 debug asm-trim-method default + R 3 debug asm-trim-method default + R 0 CLUSTER MIGRATION IMPORT 0 1 3 3 + wait_for_asm_done + R 0 flushall + R 1 flushall + } + + test "Test multiple active trim jobs can be scheduled" { + # Active trim will be scheduled but it won't run + R 0 debug asm-trim-method active -1 + R 3 debug asm-trim-method active -1 + + populate_slot 500 -slot 0 + populate_slot 500 -slot 1 + populate_slot 500 -slot 3 + populate_slot 500 -slot 4 + + # Migrate 1500 keys + R 1 CLUSTER MIGRATION IMPORT 0 1 + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 1 && + [CI 3 cluster_slot_migration_active_trim_running] == 1 + } else { + fail "migrate failed" + } + + # Migrate another slot and verify there are two trim tasks on the source + R 1 CLUSTER MIGRATION IMPORT 3 3 + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 2 && + [CI 3 cluster_slot_migration_active_trim_running] == 2 + } else { + fail "migrate failed" + } + + # Enabled active trim and wait until it is completed. + R 0 debug asm-trim-method active 0 + R 3 debug asm-trim-method active 0 + wait_for_asm_done + + assert_equal 500 [R 0 dbsize] + assert_equal 500 [R 3 dbsize] + assert_equal 0 [R 0 cluster countkeysinslot 0] + assert_equal 0 [R 0 cluster countkeysinslot 1] + assert_equal 0 [R 0 cluster countkeysinslot 3] + assert_equal 500 [R 0 cluster countkeysinslot 4] + + # cleanup + R 0 debug asm-trim-method default + R 3 debug asm-trim-method default + R 0 CLUSTER MIGRATION IMPORT 0 1 3 3 + wait_for_asm_done + R 0 flushall + R 1 flushall + } + + test "Test active-trim clears partially imported keys on cancel" { + R 1 debug asm-trim-method active + R 4 debug asm-trim-method active + + # Rdb delivery will take 10 seconds + R 0 config set rdb-key-save-delay 10000 + populate_slot 250 -slot 0 + populate_slot 250 -slot 1 + populate_slot 250 -slot 3 + populate_slot 250 -slot 4 + + R 1 CLUSTER MIGRATION IMPORT 0 100 + after 2000 + R 1 CLUSTER MIGRATION CANCEL ALL + wait_for_asm_done + + assert_morethan [CI 1 cluster_slot_migration_active_trim_current_job_keys] 0 + assert_morethan [CI 4 cluster_slot_migration_active_trim_current_job_trimmed] 0 + + assert_equal 1000 [R 0 dbsize] + assert_equal 1000 [R 3 dbsize] + assert_equal 0 [R 1 dbsize] + assert_equal 0 [R 4 dbsize] + + # Cleanup + R 1 debug asm-trim-method default + R 4 debug asm-trim-method default + R 0 config set rdb-key-save-delay 0 + } + + test "Test active-trim clears partially imported keys on failover" { + R 1 debug asm-trim-method active + R 4 debug asm-trim-method active + + # Rdb delivery will take 10 seconds + R 0 config set rdb-key-save-delay 10000 + + populate_slot 250 -slot 0 + populate_slot 250 -slot 1 + populate_slot 250 -slot 3 + populate_slot 250 -slot 4 + + set prev_trim_started_1 [CI 1 cluster_slot_migration_stats_active_trim_started] + set prev_trim_started_4 [CI 4 cluster_slot_migration_stats_active_trim_started] + + R 1 CLUSTER MIGRATION IMPORT 0 100 + after 2000 + failover_and_wait_for_done 4 + wait_for_asm_done + + # Verify there is at least one trim job started + assert_morethan [CI 1 cluster_slot_migration_stats_active_trim_started] $prev_trim_started_1 + assert_morethan [CI 4 cluster_slot_migration_stats_active_trim_started] $prev_trim_started_4 + + assert_equal 1000 [R 0 dbsize] + assert_equal 1000 [R 3 dbsize] + assert_equal 0 [R 1 dbsize] + assert_equal 0 [R 4 dbsize] + + # Cleanup + failover_and_wait_for_done 1 + R 1 debug asm-trim-method default + R 4 debug asm-trim-method default + R 0 config set rdb-key-save-delay 0 + R 0 flushall + R 1 flushall + } + + test "Test import task does not start if active trim is in progress for the same slots" { + # Active trim will be scheduled but it won't run + R 0 flushall + R 1 flushall + R 0 debug asm-trim-method active -1 + + populate_slot 500 -slot 0 + populate_slot 500 -slot 1 + + # Migrate 1000 keys + R 1 CLUSTER MIGRATION IMPORT 0 1 + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 1 + } else { + fail "migrate failed" + } + + # Try to migrate slots back + R 0 CLUSTER MIGRATION IMPORT 0 1 + wait_for_log_messages 0 {"*Can not start import task*trim in progress for some of the slots*"} 0 1000 10 + + # Enabled active trim and verify slots are imported back + R 0 debug asm-trim-method active 0 + wait_for_asm_done + + assert_equal 1000 [R 0 dbsize] + assert_equal 500 [R 0 cluster countkeysinslot 0] + assert_equal 500 [R 0 cluster countkeysinslot 1] + + # cleanup + R 0 debug asm-trim-method default + R 0 flushall + } + + test "Rdb save during active trim should skip keys in trimmed slots" { + # Insert some delay to activate trim + R 0 debug asm-trim-method active 1000 + R 0 config set repl-diskless-sync-delay 0 + R 0 flushall + + populate_slot 5000 -idx 0 -slot 0 + populate_slot 5000 -idx 0 -slot 1 + populate_slot 5000 -idx 0 -slot 2 + + # Start migration and wait until trim is in progress + R 1 CLUSTER MIGRATION IMPORT 0 1 + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 1 && + [S 0 rdb_bgsave_in_progress] == 0 + } else { + puts "[CI 0 cluster_slot_migration_active_tasks]" + puts "[CI 0 cluster_slot_migration_active_trim_running]" + fail "trim failed" + } + + # Trigger save during active trim + R 0 save + # Wait until the log contains a "keys skipped" message with a non-zero value + wait_for_log_messages 0 {"*BGSAVE done, 5000 keys saved, [1-9]* keys skipped*"} 0 1000 10 + + restart_server 0 yes no yes nosave + assert_equal 5000 [R 0 dbsize] + assert_equal 0 [R 0 cluster countkeysinslot 0] + assert_equal 0 [R 0 cluster countkeysinslot 1] + assert_equal 5000 [R 0 cluster countkeysinslot 2] + + # Cleanup + wait_for_cluster_propagation + wait_for_cluster_state "ok" + R 0 flushall + R 1 flushall + R 0 save + R 0 CLUSTER MIGRATION IMPORT 0 1 + wait_for_asm_done + } + + test "AOF rewrite during active trim should skip keys in trimmed slots" { + R 0 debug asm-trim-method active 1000 + R 0 config set repl-diskless-sync-delay 0 + R 0 config set aof-use-rdb-preamble no + R 0 config set appendonly yes + R 0 config rewrite + R 0 flushall + populate_slot 5000 -idx 0 -slot 0 + populate_slot 5000 -idx 0 -slot 1 + populate_slot 5000 -idx 0 -slot 2 + + R 1 CLUSTER MIGRATION IMPORT 0 1 + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 1 + } else { + puts "[CI 0 cluster_slot_migration_active_tasks]" + puts "[CI 0 cluster_slot_migration_active_trim_running]" + fail "trim failed" + } + + wait_for_condition 50 100 { + [S 0 rdb_bgsave_in_progress] == 0 + } else { + fail "bgsave is in progress" + } + + R 0 bgrewriteaof + # Wait until the log contains a "keys skipped" message with a non-zero value + wait_for_log_messages 0 {"*AOF rewrite done, [1-9]* keys saved, [1-9]* keys skipped*"} 0 1000 10 + + restart_server 0 yes no yes nosave + assert_equal 5000 [R 0 dbsize] + assert_equal 0 [R 0 cluster countkeysinslot 0] + assert_equal 0 [R 0 cluster countkeysinslot 1] + assert_equal 5000 [R 0 cluster countkeysinslot 2] + + # cleanup + R 0 config set appendonly no + R 0 config rewrite + restart_server 0 yes no yes nosave + wait_for_cluster_propagation + wait_for_cluster_state "ok" + R 0 flushall + R 1 flushall + R 0 save + R 0 CLUSTER MIGRATION IMPORT 0 1 + wait_for_asm_done + } + + test "Pause actions will stop active trimming" { + R 0 debug asm-trim-method active 1000 + R 0 config set repl-diskless-sync-delay 0 + R 0 flushall + populate_slot 10000 -idx 0 -slot 0 + + R 1 CLUSTER MIGRATION IMPORT 0 100 + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 1 + } else { + puts "[CI 0 cluster_slot_migration_active_tasks]" + puts "[CI 0 cluster_slot_migration_active_trim_running]" + fail "trim failed" + } + + # Pause the server and verify no keys are trimmed + R 0 client pause 100000 write ;# pause 100s + set prev [CI 0 cluster_slot_migration_active_trim_current_job_trimmed] + after 1000 ; # wait some time to see if any keys are trimmed + set curr [CI 0 cluster_slot_migration_active_trim_current_job_trimmed] + assert_equal $prev $curr + + R 0 client unpause + R 0 debug asm-trim-method default + wait_for_asm_done + assert_equal 0 [R 0 dbsize] + + # revert + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + assert_equal 10000 [R 0 dbsize] + } + + foreach diskless_load {"disabled" "swapdb" "on-empty-db"} { + test "Test fullsync cancels active trim (repl-diskless-load $diskless_load)" { + R 3 debug asm-trim-method active -10 + R 3 config set repl-diskless-load $diskless_load + R 0 flushall + + R 0 config set repl-diskless-sync-delay 0 + populate_slot 10000 -idx 0 -slot 0 + + R 1 CLUSTER MIGRATION IMPORT 0 0 + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 0 && + [CI 3 cluster_slot_migration_active_trim_running] == 1 + } else { + puts "[CI 0 cluster_slot_migration_active_tasks]" + puts "[CI 0 cluster_slot_migration_active_trim_running]" + puts "[CI 3 cluster_slot_migration_active_trim_running]" + fail "trim failed" + } + + set prev_cancelled [CI 3 cluster_slot_migration_stats_active_trim_cancelled] + R 0 config set client-output-buffer-limit "replica 1024 0 0" + + # Trigger a fullsync + populate_slot 1 -idx 0 -size 2000000 -slot 2 + + wait_for_condition 1000 10 { + [CI 3 cluster_slot_migration_active_trim_running] == 0 && + [CI 3 cluster_slot_migration_stats_active_trim_cancelled] == $prev_cancelled + 1 + } else { + puts "[CI 3 cluster_slot_migration_active_trim_running]" + puts "[CI 3 cluster_slot_migration_stats_active_trim_cancelled]" + fail "trim failed" + } + + R 3 debug asm-trim-method active 0 + R 3 config set repl-diskless-load disabled + R 0 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + wait_for_ofs_sync [Rn 0] [Rn 3] + assert_equal 10001 [R 0 dbsize] + assert_equal 10001 [R 3 dbsize] + assert_equal 0 [R 1 dbsize] + assert_equal 0 [R 4 dbsize] + R 0 flushall + } + } + + test "Test importing slots while active-trim is in progress for the same slots on replica" { + R 3 debug asm-trim-method active 10000 + R 0 flushall + populate_slot 10000 -slot 0 + wait_for_ofs_sync [Rn 0] [Rn 3] + + # Wait until active trim is in progress on replica + R 1 CLUSTER MIGRATION IMPORT 0 100 + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 0 && + [CI 3 cluster_slot_migration_active_trim_running] == 1 + } else { + puts "[CI 0 cluster_slot_migration_active_tasks]" + puts "[CI 0 cluster_slot_migration_active_trim_running]" + puts "[CI 3 cluster_slot_migration_active_trim_running]" + fail "trim failed" + } + + set loglines [count_log_lines -3] + + # Get slots back + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_condition 1000 20 { + [CI 0 cluster_slot_migration_active_tasks] == 1 && + [CI 0 cluster_slot_migration_active_trim_running] == 0 && + [CI 3 cluster_slot_migration_active_trim_running] == 1 + } else { + fail "trim failed" + } + + # Verify replica blocks master until trim is done + wait_for_log_messages -3 {"*Blocking master client until trim job is done*"} $loglines 1000 30 + R 3 debug asm-trim-method active 0 + wait_for_log_messages -3 {"*Unblocking master client after active trim*"} $loglines 1000 30 + + wait_for_asm_done + wait_for_ofs_sync [Rn 0] [Rn 3] + assert_equal 10000 [R 0 dbsize] + assert_equal 10000 [R 3 dbsize] + assert_equal 0 [R 1 dbsize] + assert_equal 0 [R 4 dbsize] + } + + test "TRIMSLOTS should not trim slots that this node is serving" { + assert_error {*the slot 0 is served by this node*} {R 0 trimslots ranges 1 0 0} + assert_error {*READONLY*} {R 3 trimslots ranges 1 0 100} + assert_equal {OK} [R 0 trimslots ranges 1 16383 16383] + assert_error {*READONLY*} {R 3 trimslots ranges 1 16383 16383} + } + + test "Trigger multiple active trim jobs at the same time" { + R 1 debug asm-trim-method active 0 + R 1 flushall + + set prev_trim_done [CI 1 cluster_slot_migration_stats_active_trim_completed] + + R 1 debug populate 1000 [slot_prefix 0] 100 + R 1 debug populate 1000 [slot_prefix 1] 100 + R 1 debug populate 1000 [slot_prefix 2] 100 + + R 1 multi + R 1 trimslots ranges 1 0 0 + R 1 trimslots ranges 1 1 1 + R 1 trimslots ranges 1 2 2 + R 1 exec + + wait_for_condition 1000 10 { + [CI 1 cluster_slot_migration_stats_active_trim_completed] == $prev_trim_done + 3 + } else { + fail "active trim failed" + } + + R 1 flushall + R 1 debug asm-trim-method default + } + + test "Restart will clean up unowned slot keys" { + R 1 flushall + + # generate 1000 keys belonging to slot 0 + R 1 debug populate 1000 [slot_prefix 0] 100 + assert {[scan [regexp -inline {keys\=([\d]*)} [R 1 info keyspace]] keys=%d] >= 1000} + + # restart node-1 + restart_server -1 true false true save + wait_for_cluster_propagation + wait_for_cluster_state "ok" + + # Node-1 has no keys since unowned slot 0 keys were cleaned up during restart + assert {[scan [regexp -inline {keys\=([\d]*)} [R 1 info keyspace]] keys=%d] == {}} + + R 1 flushall + } + + test "Test active trim is used when client tracking is used" { + R 0 flushall + R 1 flushall + R 0 debug asm-trim-method default + R 1 debug asm-trim-method default + + set prev_active_trim [CI 0 cluster_slot_migration_stats_active_trim_completed] + + # Setup a tracking client that is redirected to a pubsub client + set rd_redirection [redis_deferring_client] + $rd_redirection client id + set redir_id [$rd_redirection read] + $rd_redirection subscribe __redis__:invalidate + $rd_redirection read ; # Consume the SUBSCRIBE reply. + + # setup tracking + set key0 [slot_key 0 key] + R 0 CLIENT TRACKING on REDIRECT $redir_id + R 0 SET $key0 1 + R 0 GET $key0 + R 1 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_stats_active_trim_completed] == [expr $prev_active_trim + 1] + } else { + fail "active trim did not happen" + } + + # Verify the tracking client received the invalidation message + set msg [$rd_redirection read] + set head [lindex $msg 0] + + if {$head eq "message"} { + # RESP 2 + set got_key [lindex [lindex $msg 2] 0] + } elseif {$head eq "invalidate"} { + # RESP 3 + set got_key [lindex $msg 1 0] + } else { + fail "unexpected invalidation message: $msg" + } + assert_equal $got_key $key0 + + # cleanup + $rd_redirection close + wait_for_asm_done + R 0 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + R 0 flushall + } +} + +set testmodule [file normalize tests/modules/atomicslotmigration.so] + +start_cluster 3 6 [list tags {external:skip cluster modules} config_lines [list loadmodule $testmodule cluster-node-timeout 60000 cluster-allow-replica-migration no]] { + test "Module api sanity" { + R 0 asm.sanity ;# on master + R 3 asm.sanity ;# on replica + } + + test "Module replicate cross slot command" { + set task_id [setup_slot_migration_with_delay 0 1 0 100] + set listkey [slot_key 0 "asmlist"] + # replicate cross slot command during migrating + R 0 asm.lpush_replicate_crossslot_command $listkey "item1" + + # node 0 will fail due to cross slot + wait_for_condition 2000 10 { + [string match {*canceled*} [migration_status 0 $task_id state]] && + [string match {*cross slot*} [migration_status 0 $task_id last_error]] + } else { + fail "ASM task did not fail" + } + R 1 CLUSTER MIGRATION CANCEL ID $task_id + + # sanity check if lpush replicated correctly to the replica + wait_for_ofs_sync [Rn 0] [Rn 3] + assert_equal {item1} [R 0 lrange $listkey 0 -1] + R 3 readonly + assert_equal {item1} [R 3 lrange $listkey 0 -1] + } + + test "Test RM_ClusterCanAccessKeysInSlot" { + # Test invalid slots + assert_equal 0 [R 0 asm.cluster_can_access_keys_in_slot -1] + assert_equal 0 [R 0 asm.cluster_can_access_keys_in_slot 20000] + assert_equal 0 [R 2 asm.cluster_can_access_keys_in_slot 16384] + assert_equal 0 [R 5 asm.cluster_can_access_keys_in_slot 16384] + + # Test on a master-replica pair + assert_equal 1 [R 0 asm.cluster_can_access_keys_in_slot 0] + assert_equal 1 [R 0 asm.cluster_can_access_keys_in_slot 100] + assert_equal 1 [R 3 asm.cluster_can_access_keys_in_slot 0] + assert_equal 1 [R 3 asm.cluster_can_access_keys_in_slot 100] + + # Test on a master-replica pair + assert_equal 1 [R 2 asm.cluster_can_access_keys_in_slot 16383] + assert_equal 1 [R 5 asm.cluster_can_access_keys_in_slot 16383] + } + + test "Test RM_ClusterCanAccessKeysInSlot returns false for unowned slots" { + # Active trim will be scheduled but it won't run + R 0 debug asm-trim-method active -1 + R 3 debug asm-trim-method active -1 + + setup_slot_migration_with_delay 0 1 0 100 3 1000000 + + # Verify importing slots are not local + assert_equal 0 [R 1 asm.cluster_can_access_keys_in_slot 0] + assert_equal 0 [R 1 asm.cluster_can_access_keys_in_slot 100] + assert_equal 0 [R 4 asm.cluster_can_access_keys_in_slot 0] + assert_equal 0 [R 4 asm.cluster_can_access_keys_in_slot 100] + + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 1 && + [CI 3 cluster_slot_migration_active_trim_running] == 1 + } else { + fail "migrate failed" + } + + # Wait for config propagation before checking the slot ownership on replica + wait_for_cluster_propagation + + # Verify slots that are being trimmed are not local + assert_equal 0 [R 0 asm.cluster_can_access_keys_in_slot 0] + assert_equal 0 [R 0 asm.cluster_can_access_keys_in_slot 100] + assert_equal 0 [R 3 asm.cluster_can_access_keys_in_slot 0] + assert_equal 0 [R 3 asm.cluster_can_access_keys_in_slot 100] + + # Enabled active trim and wait until it is completed. + R 0 debug asm-trim-method active 0 + R 3 debug asm-trim-method active 0 + wait_for_asm_done + wait_for_ofs_sync [Rn 0] [Rn 3] + + # Verify slots are local after migration + assert_equal 1 [R 1 asm.cluster_can_access_keys_in_slot 0] + assert_equal 1 [R 1 asm.cluster_can_access_keys_in_slot 100] + assert_equal 1 [R 4 asm.cluster_can_access_keys_in_slot 0] + assert_equal 1 [R 4 asm.cluster_can_access_keys_in_slot 100] + + # cleanup + R 0 debug asm-trim-method default + R 3 debug asm-trim-method default + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + R 0 flushall + R 1 flushall + } + + foreach trim_method {"active" "bg"} { + test "Test cluster module notifications on a successful migration ($trim_method-trim)" { + clear_module_event_log + R 0 debug asm-trim-method $trim_method + R 3 debug asm-trim-method $trim_method + R 6 debug asm-trim-method $trim_method + + # Set a key in the slot range + set key [slot_key 0 mykey] + R 0 set $key "value" + + # Migrate the slot ranges + set task_id [R 1 CLUSTER MIGRATION IMPORT 0 100 200 300] + wait_for_asm_done + + set src_id [R 0 cluster myid] + set dest_id [R 1 cluster myid] + + # Verify the events on source, both master and replica + set migrate_event_log [list \ + "sub: cluster-slot-migration-migrate-started, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100,200-300" \ + "sub: cluster-slot-migration-migrate-completed, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100,200-300" \ + ] + assert_equal [R 0 asm.get_cluster_event_log] $migrate_event_log + assert_equal [R 3 asm.get_cluster_event_log] {} + assert_equal [R 6 asm.get_cluster_event_log] {} + + # Verify the events on destination, both master and replica + set import_event_log [list \ + "sub: cluster-slot-migration-import-started, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100,200-300" \ + "sub: cluster-slot-migration-import-completed, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100,200-300" \ + ] + wait_for_condition 500 20 { + [R 1 asm.get_cluster_event_log] eq $import_event_log && + [R 4 asm.get_cluster_event_log] eq $import_event_log && + [R 7 asm.get_cluster_event_log] eq $import_event_log + } else { + puts "R1: [R 1 asm.get_cluster_event_log]" + puts "R4: [R 4 asm.get_cluster_event_log]" + puts "R7: [R 7 asm.get_cluster_event_log]" + fail "ASM import event not received" + } + + # Verify the trim events + if {$trim_method eq "active"} { + set trim_event_log [list \ + "sub: cluster-slot-migration-trim-started, slots:0-100,200-300" \ + "keyspace: key_trimmed, key: $key" \ + "sub: cluster-slot-migration-trim-completed, slots:0-100,200-300" \ + ] + } else { + set trim_event_log [list \ + "sub: cluster-slot-migration-trim-background, slots:0-100,200-300" \ + ] + } + wait_for_condition 500 10 { + [R 0 asm.get_cluster_trim_event_log] eq $trim_event_log && + [R 3 asm.get_cluster_trim_event_log] eq $trim_event_log && + [R 6 asm.get_cluster_trim_event_log] eq $trim_event_log + } else { + fail "ASM source trim event not received" + } + + # cleanup + R 0 CLUSTER MIGRATION IMPORT 0 100 200 300 + wait_for_asm_done + clear_module_event_log + reset_default_trim_method + R 0 flushall + R 1 flushall + } + + test "Test cluster module notifications on a failed migration ($trim_method-trim)" { + clear_module_event_log + R 1 debug asm-trim-method $trim_method + R 4 debug asm-trim-method $trim_method + R 7 debug asm-trim-method $trim_method + + # Set a key in the slot range + set key [slot_key 0 mykey] + R 0 set $key "value" + + # Start migration and cancel it + set task_id [setup_slot_migration_with_delay 0 1 0 100 0 2000000] + # Wait until at least one key is moved to destination + wait_for_condition 1000 10 { + [scan [regexp -inline {keys\=([\d]*)} [R 1 info keyspace]] keys=%d] >= 1 + } else { + fail "Key not moved to destination" + } + R 1 CLUSTER MIGRATION CANCEL ID $task_id + wait_for_asm_done + + set src_id [R 0 cluster myid] + set dest_id [R 1 cluster myid] + + # Verify the events on source, both master and replica + set migrate_event_log [list \ + "sub: cluster-slot-migration-migrate-started, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + "sub: cluster-slot-migration-migrate-failed, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + ] + assert_equal [R 0 asm.get_cluster_event_log] $migrate_event_log + assert_equal [R 3 asm.get_cluster_event_log] {} + assert_equal [R 6 asm.get_cluster_event_log] {} + + # Verify the events on destination, both master and replica + set import_event_log [list \ + "sub: cluster-slot-migration-import-started, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + "sub: cluster-slot-migration-import-failed, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + ] + wait_for_condition 500 10 { + [R 1 asm.get_cluster_event_log] eq $import_event_log && + [R 4 asm.get_cluster_event_log] eq $import_event_log && + [R 7 asm.get_cluster_event_log] eq $import_event_log + } else { + fail "ASM import event not received" + } + + # Verify the trim events on destination (partially imported keys are trimmed) + if {$trim_method eq "active"} { + set trim_event_log [list \ + "sub: cluster-slot-migration-trim-started, slots:0-100" \ + "keyspace: key_trimmed, key: $key" \ + "sub: cluster-slot-migration-trim-completed, slots:0-100" \ + ] + } else { + set trim_event_log [list \ + "sub: cluster-slot-migration-trim-background, slots:0-100" \ + ] + } + wait_for_condition 500 10 { + [R 1 asm.get_cluster_trim_event_log] eq $trim_event_log && + [R 4 asm.get_cluster_trim_event_log] eq $trim_event_log && + [R 7 asm.get_cluster_trim_event_log] eq $trim_event_log + } else { + fail "ASM destination trim event not received" + } + + # cleanup + clear_module_event_log + reset_default_trim_method + wait_for_asm_done + R 0 flushall + R 1 flushall + } + + test "Test cluster module notifications on failover ($trim_method-trim)" { + # NOTE: cluster legacy may have a bug, multiple manual failover will fail, + # so only perform one round of failover test, fix it later + if {$trim_method eq "bg"} { + clear_module_event_log + R 1 debug asm-trim-method $trim_method + R 4 debug asm-trim-method $trim_method + R 7 debug asm-trim-method $trim_method + + # Set a key in the slot range + set key [slot_key 0 mykey] + R 0 set $key "value" + + # Start migration + set task_id [setup_slot_migration_with_delay 0 1 0 100 0 2000000] + # Wait until at least one key is moved to destination + wait_for_condition 1000 10 { + [scan [regexp -inline {keys\=([\d]*)} [R 1 info keyspace]] keys=%d] >= 1 + } else { + fail "Key not moved to destination" + } + + failover_and_wait_for_done 4 + wait_for_asm_done + + set src_id [R 0 cluster myid] + set dest_id [R 1 cluster myid] + + # Verify the events on source, both master and replica + set migrate_event_log [list \ + "sub: cluster-slot-migration-migrate-started, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + "sub: cluster-slot-migration-migrate-failed, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + ] + assert_equal [R 0 asm.get_cluster_event_log] $migrate_event_log + assert_equal [R 3 asm.get_cluster_event_log] {} + assert_equal [R 6 asm.get_cluster_event_log] {} + + # Verify the events on destination, both master and replica + set import_event_log [list \ + "sub: cluster-slot-migration-import-started, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + "sub: cluster-slot-migration-import-failed, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + ] + wait_for_condition 500 20 { + [R 1 asm.get_cluster_event_log] eq $import_event_log && + [R 4 asm.get_cluster_event_log] eq $import_event_log && + [R 7 asm.get_cluster_event_log] eq $import_event_log + } else { + puts "R1: [R 1 asm.get_cluster_event_log]" + puts "R4: [R 4 asm.get_cluster_event_log]" + puts "R7: [R 7 asm.get_cluster_event_log]" + fail "ASM import event not received" + } + + # Verify the trim events on destination (partially imported keys are trimmed) + # NOTE: after failover, the new master will initiate the slot trimming, + # and only slot 0 has data, so only slot 0 is trimmed + if {$trim_method eq "active"} { + set trim_event_log [list \ + "sub: cluster-slot-migration-trim-started, slots:0-0" \ + "keyspace: key_trimmed, key: $key" \ + "sub: cluster-slot-migration-trim-completed, slots:0-0" \ + ] + } else { + set trim_event_log [list \ + "sub: cluster-slot-migration-trim-background, slots:0-0" \ + ] + } + wait_for_condition 500 20 { + [R 1 asm.get_cluster_trim_event_log] eq $trim_event_log && + [R 4 asm.get_cluster_trim_event_log] eq $trim_event_log && + [R 7 asm.get_cluster_trim_event_log] eq $trim_event_log + } else { + puts "R1: [R 1 asm.get_cluster_trim_event_log]" + puts "R4: [R 4 asm.get_cluster_trim_event_log]" + puts "R7: [R 7 asm.get_cluster_trim_event_log]" + fail "ASM destination trim event not received" + } + + # cleanup + failover_and_wait_for_done 1 + clear_module_event_log + reset_default_trim_method + R 0 flushall + R 1 flushall + } + } + } + + foreach with_rdb {"with" "without"} { + test "Test cluster module notifications when replica restart $with_rdb RDB during importing" { + clear_module_event_log + R 1 debug asm-trim-method $trim_method + R 4 debug asm-trim-method $trim_method + R 7 debug asm-trim-method $trim_method + R 4 config set save "" + + set src_id [R 0 cluster myid] + set dest_id [R 1 cluster myid] + + # Set a key in the slot range + set key [slot_key 0 mykey] + R 0 set $key "value" + + # Start migration, 2s delay + set task_id [setup_slot_migration_with_delay 0 1 0 100 0 2000000] + # Wait until at least one key is moved to destination + wait_for_condition 1000 10 { + [scan [regexp -inline {keys\=([\d]*)} [R 1 info keyspace]] keys=%d] >= 1 + } else { + fail "Key not moved to destination" + } + wait_for_ofs_sync [Rn 1] [Rn 4] + + # restart node 4 + if {$with_rdb eq "with"} { + restart_server -4 true false true save ;# rdb save + } else { + restart_server -4 true false true nosave ;# no rdb saved + } + wait_for_cluster_propagation + + wait_for_asm_done + + # started and completed are paired, and not duplicated + set import_event_log [list \ + "sub: cluster-slot-migration-import-started, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + "sub: cluster-slot-migration-import-completed, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + ] + wait_for_condition 500 10 { + [R 1 asm.get_cluster_event_log] eq $import_event_log && + [R 4 asm.get_cluster_event_log] eq $import_event_log && + [R 7 asm.get_cluster_event_log] eq $import_event_log + } else { + fail "ASM import event not received" + } + + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + R 4 save ;# save an empty rdb to override previous one + clear_module_event_log + reset_default_trim_method + R 0 flushall + R 1 flushall + } + } + + test "Test cluster module notifications when replica is disconnected and full resync after importing" { + clear_module_event_log + R 1 debug asm-trim-method $trim_method + R 4 debug asm-trim-method $trim_method + R 7 debug asm-trim-method $trim_method + + set src_id [R 0 cluster myid] + set dest_id [R 1 cluster myid] + + # Set a key in the slot range + set key [slot_key 0 mykey] + R 0 set $key "value" + + # Start migration, 2s delay + set task_id [setup_slot_migration_with_delay 0 1 0 100 0 2000000] + # Wait until at least one key is moved to destination + wait_for_condition 1000 10 { + [scan [regexp -inline {keys\=([\d]*)} [R 1 info keyspace]] keys=%d] >= 1 + } else { + fail "Key not moved to destination" + } + wait_for_ofs_sync [Rn 1] [Rn 4] + + # puase node-4 + set r4_pid [S 4 process_id] + pause_process $r4_pid + + # set a small repl-backlog-size and write some commands to make node-4 + # full resync when reconnecting after waking up + set r1_full_sync [S 1 sync_full] + R 1 config set repl-backlog-size 16kb + R 1 client kill type replica + set 1k_str [string repeat "a" 1024] + for {set i 0} {$i < 2000} {incr i} { + R 1 set [slot_key 6000] $1k_str + } + + # after ASM task is completed, wake up node-4 + wait_for_condition 1000 10 { + [CI 1 cluster_slot_migration_active_tasks] == 0 && + [CI 1 cluster_slot_migration_active_trim_running] == 0 + } else { + fail "ASM tasks did not completed" + } + resume_process $r4_pid + + # make sure full resync happens + wait_for_sync [Rn 4] + wait_for_ofs_sync [Rn 1] [Rn 4] + assert_morethan [S 1 sync_full] $r1_full_sync + + # started and completed are paired, and not duplicated + set import_event_log [list \ + "sub: cluster-slot-migration-import-started, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + "sub: cluster-slot-migration-import-completed, source_node_id:$src_id, destination_node_id:$dest_id, task_id:$task_id, slots:0-100" \ + ] + wait_for_condition 500 10 { + [R 1 asm.get_cluster_event_log] eq $import_event_log && + [R 4 asm.get_cluster_event_log] eq $import_event_log && + [R 7 asm.get_cluster_event_log] eq $import_event_log + } else { + fail "ASM import event not received" + } + + # since ASM task is completed on node-1 before node-4 reconnects, + # no trim event should be received on node-4 + assert_equal {} [R 4 asm.get_cluster_trim_event_log] + + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + clear_module_event_log + reset_default_trim_method + R 0 flushall + R 1 flushall + } + + test "Test new master can trim slots when migration is completed and failover occurs on source side" { + R 0 asm.disable_trim ;# can not start slot trimming on source side + set slot0_key [slot_key 0 mykey] + R 0 set $slot0_key "value" + + # migrate slot 0 from #0 to #1, and wait it completed, but not allow to trim slots + # on source node + set task_id [R 1 CLUSTER MIGRATION IMPORT 0 0] + wait_for_condition 1000 10 { + [string match {*completed*} [migration_status 0 $task_id state]] && + [string match {*completed*} [migration_status 1 $task_id state]] + } else { + fail "ASM task did not complete" + } + # verify trim is not allowed on source node, and replica node doesn't have trim job either + wait_for_ofs_sync [Rn 0] [Rn 3] + assert_equal 1 [R 0 asm.trim_in_progress] + assert_equal "value" [R 0 asm.read_pending_trim_key $slot0_key] + assert_equal 0 [R 3 asm.trim_in_progress] + assert_equal "value" [R 3 asm.read_pending_trim_key $slot0_key] + + set loglines [count_log_lines 0] + + # failover happens on source node, instance #3 become slave, #0 become master + failover_and_wait_for_done 3 + R 0 asm.enable_trim ;# enable trim on old master + + # old master should cancel the pending trim job + wait_for_log_messages 0 {"*Cancelling the pending trim job*"} $loglines 1000 10 + + wait_for_ofs_sync [Rn 3] [Rn 0] + # verify trim is allowed on new master, and the key is trimmed + wait_for_condition 1000 10 { + [R 3 asm.trim_in_progress] == 0 && + [R 3 asm.read_pending_trim_key $slot0_key] eq "" && + [R 0 asm.trim_in_progress] == 0 && + [R 0 asm.read_pending_trim_key $slot0_key] eq "" + } else { + fail "Trim did not complete" + } + + # verify the trim events, use active trim since module is subscribed to trimmed event + set trim_event_log [list \ + "sub: cluster-slot-migration-trim-started, slots:0-0" \ + "keyspace: key_trimmed, key: $slot0_key" \ + "sub: cluster-slot-migration-trim-completed, slots:0-0" \ + ] + wait_for_condition 500 20 { + [R 0 asm.get_cluster_trim_event_log] eq $trim_event_log && + [R 3 asm.get_cluster_trim_event_log] eq $trim_event_log && + [R 6 asm.get_cluster_trim_event_log] eq $trim_event_log + } else { + fail "ASM destination trim event not received" + } + + # cleanup + failover_and_wait_for_done 0 + R 0 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + clear_module_event_log + reset_default_trim_method + R 0 flushall + R 1 flushall + } + + test "Test module replicates commands at the beginning of slot migration " { + R 0 flushall + R 1 flushall + + # Sanity check + assert_equal 0 [R 1 asm.read_keyless_cmd_val] + assert_equal 0 [R 4 asm.read_keyless_cmd_val] + + # Enable module command replication and set a key to be replicated + # Module will replicate two commands: + # 1- A keyless command: asm.keyless_cmd + # 2- SET command for the given key and value + set keyname [slot_key 0 modulekey] + R 0 asm.replicate_module_command 1 $keyname "value" + + setup_slot_migration_with_delay 0 1 0 100 + wait_for_asm_done + wait_for_ofs_sync [Rn 1] [Rn 4] + + # Verify the commands are replicated + assert_equal 1 [R 1 asm.read_keyless_cmd_val] + assert_equal value [R 1 get $keyname] + + # Verify the commands are replicated to replica + R 4 readonly + assert_equal 1 [R 4 asm.read_keyless_cmd_val] + assert_equal value [R 4 get $keyname] + + # cleanup + R 0 asm.replicate_module_command 0 "" "" + R 0 CLUSTER MIGRATION IMPORT 0 100 + wait_for_asm_done + R 0 flushall + R 1 flushall + } + + test "Test subcommand propagation during slot migration" { + R 0 flushall + R 1 flushall + set task_id [setup_slot_migration_with_delay 0 1 0 100] + + set key [slot_key 0 mykey] + R 0 asm.parent set $key "value" ;# execute a module subcommand + wait_for_asm_done + assert_equal "value" [R 1 GET $key] + + # cleanup + R 0 cluster migration import 0 100 + wait_for_asm_done + } + + test "Test trim method selection based on module keyspace subscription" { + R 0 debug asm-trim-method default + R 1 debug asm-trim-method default + + R 0 flushall + R 1 flushall + + populate_slot 10 -idx 0 -slot 0 + + # Make sure module is subscribed to NOTIFY_KEY_TRIMMED event. In this + # case, active trim must be used. + R 0 asm.subscribe_trimmed_event 1 + set loglines [count_log_lines 0] + R 1 CLUSTER MIGRATION IMPORT 0 15 + wait_for_asm_done + wait_for_log_messages 0 {"*Active trim scheduled for slots: 0-15*"} $loglines 1000 10 + + # Move slots back to node-0. Make sure module is not subscribed to + # NOTIFY_KEY_TRIMMED event. In this case, background trim must be used. + R 1 asm.subscribe_trimmed_event 0 + set loglines [count_log_lines -1] + R 0 CLUSTER MIGRATION IMPORT 0 15 + wait_for_asm_done + wait_for_log_messages -1 {"*Background trim started for slots: 0-15*"} $loglines 1000 10 + + # cleanup + wait_for_asm_done + R 0 asm.subscribe_trimmed_event 1 + R 1 asm.subscribe_trimmed_event 1 + R 0 flushall + R 1 flushall + } + + test "Verify trimmed key value can be read in the server event callback" { + R 0 flushall + set key [slot_key 0] + set value "value123random" + R 0 set $key $value + + R 1 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + wait_for_condition 1000 10 { + [R 0 asm.get_last_deleted_key] eq "keyevent: key: $key, value: $value" + } else { + fail "Last deleted key event not received" + } + + # cleanup + R 0 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + } + + test "Verify module cannot open a key in a slot that is being trimmed" { + R 0 flushall + R 0 debug asm-trim-method active -1 ;# disable active trim + + set key [slot_key 0] + R 0 set $key value + + R 1 CLUSTER MIGRATION IMPORT 0 0 + wait_for_condition 1000 10 { + [CI 0 cluster_slot_migration_active_tasks] == 0 && + [CI 1 cluster_slot_migration_active_tasks] == 0 && + [CI 0 cluster_slot_migration_active_trim_running] == 1 + } else { + fail "migrate failed" + } + + # We cannot open the key since it is in a slot being trimmed + assert_equal {} [R 0 asm.get $key] + + # cleanup + R 0 debug asm-trim-method default + R 0 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + } + + test "Test RM_ClusterGetLocalSlotRanges" { + assert_equal [R 0 asm.cluster_get_local_slot_ranges] {{0 5461}} + assert_equal [R 3 asm.cluster_get_local_slot_ranges] {{0 5461}} + + R 0 cluster migration import 5463 6000 + wait_for_asm_done + wait_for_cluster_propagation + assert_equal [R 0 asm.cluster_get_local_slot_ranges] {{0 5461} {5463 6000}} + assert_equal [R 3 asm.cluster_get_local_slot_ranges] {{0 5461} {5463 6000}} + + R 0 cluster migration import 5462 5462 6001 10922 + wait_for_asm_done + wait_for_cluster_propagation + assert_equal [R 0 asm.cluster_get_local_slot_ranges] {{0 10922}} + assert_equal [R 3 asm.cluster_get_local_slot_ranges] {{0 10922}} + assert_equal [R 1 asm.cluster_get_local_slot_ranges] {} + assert_equal [R 4 asm.cluster_get_local_slot_ranges] {} + } +} + +set testmodule [file normalize tests/modules/atomicslotmigration.so] + +start_cluster 2 0 [list tags {external:skip cluster modules} config_lines [list loadmodule $testmodule cluster-node-timeout 60000 cluster-allow-replica-migration no appendonly yes]] { + test "TRIMSLOTS in AOF will work synchronously on restart" { + # When TRIMSLOTS is replayed from AOF during restart, it must execute + # synchronously rather than using active trim. This prevents race + # conditions where subsequent AOF commands might operate on keys + # that should have been trimmed. + + # Subscribe to key trimmed event to force active trim + R 0 asm.subscribe_trimmed_event 1 + populate_slot 1000 -slot 0 + populate_slot 1000 -slot 1 + R 1 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + + # verify active trim is used + assert_equal 1 [CI 0 cluster_slot_migration_stats_active_trim_completed] + + # restart server and verify aof is loaded + restart_server 0 yes no yes nosave + assert {[scan [regexp -inline {aof_current_size:([\d]*)} [R 0 info persistence]] aof_current_size=%d] > 0} + wait_for_cluster_state "ok" + + # verify TRIMSLOTS in AOF is executed synchronously + assert_equal 0 [CI 0 cluster_slot_migration_stats_active_trim_completed] + assert_equal 1000 [R 0 dbsize] + + # cleanup + R 0 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + assert_equal 2000 [R 0 dbsize] + R 0 flushall + R 1 flushall + clear_module_event_log + + } + + test "Test trim is disabled when module requests it" { + R 0 asm.disable_trim + + set slot0_key [slot_key 0 mykey] + R 0 set $slot0_key "value" + set task_id [R 1 CLUSTER MIGRATION IMPORT 0 0] + wait_for_condition 1000 10 { + [string match {*completed*} [migration_status 0 $task_id state]] + } else { + fail "ASM task did not complete" + } + # since we disable trim, the key should still exist on source, + # we can read it with REDISMODULE_OPEN_KEY_ACCESS_TRIMMED flag + assert_equal "value" [R 0 asm.read_pending_trim_key $slot0_key] + assert_equal 1 [R 0 asm.trim_in_progress] + + # enable trim and verify the key is trimmed + R 0 asm.enable_trim + wait_for_condition 1000 10 { + [R 0 asm.read_pending_trim_key $slot0_key] eq "" && + [R 0 asm.trim_in_progress] == 0 + } else { + fail "Trim did not complete" + } + wait_for_asm_done + R 0 CLUSTER MIGRATION IMPORT 0 0 + wait_for_asm_done + clear_module_event_log + } + + test "Can not start new asm task when trim is not allowed" { + # start a migration task, wait it completed but not allow to trim slots + R 0 asm.disable_trim + set task_id [R 1 CLUSTER MIGRATION IMPORT 0 0] + wait_for_condition 1000 10 { + [string match {*completed*} [migration_status 0 $task_id state]] + } else { + fail "ASM task did not complete" + } + # Can not start new migrating task since trim is disabled + set task_id [R 1 CLUSTER MIGRATION IMPORT 1 1] + wait_for_condition 1000 10 { + [string match {*fail*} [migration_status 1 $task_id state]] && + [string match {*Trim is disabled by module*} [migration_status 1 $task_id last_error]] + } else { + fail "ASM task did not fail" + } + R 0 asm.enable_trim + wait_for_asm_done + + # start a migration task, wait it completed but not allow to trim slots + R 0 asm.disable_trim + set task_id [R 1 CLUSTER MIGRATION IMPORT 2 2] + wait_for_condition 1000 10 { + [string match {*completed*} [migration_status 0 $task_id state]] + } else { + fail "ASM task did not complete" + } + set logline [count_log_lines 0] + # Can not start new importing task since trim is disabled + set task_id [R 0 CLUSTER MIGRATION IMPORT 0 1] + wait_for_log_messages 0 {"*Can not start import task*trim is disabled by module*"} $logline 1000 10 + R 0 asm.enable_trim + wait_for_asm_done + } +} + +start_server {tags "cluster external:skip"} { + test "Test RM_ClusterGetLocalSlotRanges without cluster" { + r module load $testmodule + assert_equal [r asm.cluster_get_local_slot_ranges] {{0 16383}} + } +} +} |