mirror of
https://github.com/hanzoai/redis.git
synced 2026-10-30 18:25:12 +00:00
ca6e3dafb2
Add new module API `RM_GetClusterNodeSlotRanges` that allows modules to query slot ranges for any cluster node by its node ID, not just the local node: ```c RedisModuleSlotRangeArray *RM_GetClusterNodeSlotRanges(RedisModuleCtx *ctx, const char *nodeid) ``` --------- Co-authored-by: Ozan Tezcan <[email protected]> Co-authored-by: Yuan Wang <[email protected]>
3095 lines
119 KiB
Tcl
3095 lines
119 KiB
Tcl
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 3000 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 (ignore_error_reply=1 tolerates MOVED/ASK
|
|
# replies while slots are being migrated).
|
|
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 1]
|
|
|
|
# Start write traffic on node-1 (ignore_error_reply=1 for migration redirects).
|
|
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 1]
|
|
|
|
# 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 slot 0 write load on R1. ignore_error_reply=1 tolerates MOVED/ASK
|
|
# replies that can appear while slot 0 is being migrated.
|
|
set load_handle [start_write_load "127.0.0.1" [get_port 1] 100 $slot0_key 0 0 1]
|
|
|
|
# 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 {*Write pause timeout*} [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 write pause 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
|
|
R 1 debug asm-failpoint "import-main-channel" "takeover"
|
|
|
|
# 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 due to write pause timeout
|
|
wait_for_condition 2000 10 {
|
|
[string match {*failed*} [migration_status 0 $task_id state]] &&
|
|
[string match {*Write pause 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
|
|
R 1 debug asm-failpoint "" ""
|
|
}
|
|
|
|
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] {}
|
|
}
|
|
|
|
test "Test RM_GetClusterNodeSlotRanges for local node" {
|
|
set local_id [R 0 cluster myid]
|
|
set ranges [R 0 asm.get_cluster_node_slot_ranges $local_id]
|
|
set local_ranges [R 0 asm.cluster_get_local_slot_ranges]
|
|
assert_equal $ranges $local_ranges
|
|
}
|
|
|
|
test "Test RM_GetClusterNodeSlotRanges for remote node" {
|
|
set node2_id [R 2 cluster myid]
|
|
set ranges [R 0 asm.get_cluster_node_slot_ranges $node2_id]
|
|
set remote_ranges [R 2 asm.cluster_get_local_slot_ranges]
|
|
assert_equal $ranges $remote_ranges
|
|
}
|
|
|
|
test "Test RM_GetClusterNodeSlotRanges for non-existent node" {
|
|
set ranges [R 0 asm.get_cluster_node_slot_ranges "0000000000000000000000000000000000000000"]
|
|
assert_equal $ranges {}
|
|
}
|
|
|
|
test "Test RM_GetClusterNodeSlotRanges for replica returns master slots" {
|
|
set replica3_id [R 3 cluster myid]
|
|
set ranges [R 0 asm.get_cluster_node_slot_ranges $replica3_id]
|
|
set master_ranges [R 0 asm.cluster_get_local_slot_ranges]
|
|
assert_equal $ranges $master_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}}
|
|
}
|
|
}
|
|
|
|
start_server {tags "cluster external:skip"} {
|
|
test "Test RM_GetClusterNodeSlotRanges without cluster" {
|
|
r module load $testmodule
|
|
set local_id "nonexistent-node-id"
|
|
set ranges [r asm.get_cluster_node_slot_ranges $local_id]
|
|
assert_equal $ranges {}
|
|
}
|
|
}
|
|
}
|