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, 0 insertions, 3063 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 deleted file mode 100644 index f04257f..0000000 --- a/examples/redis-unstable/tests/unit/cluster/atomic-slot-migration.tcl +++ /dev/null @@ -1,3063 +0,0 @@ -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}} - } -} -} |