Skip to content

Commit

Permalink
Fix bug in SwitchTraffic that wasn't respecting --dry_run for rea…
Browse files Browse the repository at this point in the history 
…donly and replica tablets during a resharding event (vitessio#12992)

* use switcher struct when switching shard reads during a reshard event

Signed-off-by: austenLacy <[email protected]>

* Create failing test for bug reported in vitessio#12992, where a TrafficSwitch dry run for reads during resharding tries to actually switch reads and fails

Signed-off-by: Rohit Nayak <[email protected]>

---------

Signed-off-by: austenLacy <[email protected]>
Signed-off-by: Rohit Nayak <[email protected]>
Co-authored-by: Rohit Nayak <[email protected]>
  • Loading branch information
@austenLacy @rohit-nayak-ps @maksimov
2 people authored and maksimov committed Sep 9, 2024
1 parent 183309c commit 0e637c0
Show file tree
Hide file tree
Showing 4 changed files with 37 additions and 52 deletions.
40 changes: 29 additions & 11 deletions go/test/endtoend/vreplication/vreplication_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -744,7 +744,8 @@ func reshardCustomer2to4Split(t *testing.T, cells []*Cell, sourceCellOrAlias str
t.Run("reshardCustomer2to4Split", func(t *testing.T) {
ksName := "customer"
counts := map[string]int{"zone1-600": 4, "zone1-700": 5, "zone1-800": 6, "zone1-900": 5}
reshard(t, ksName, "customer", "c2c4", "-80,80-", "-40,40-80,80-c0,c0-", 600, counts, nil, cells, sourceCellOrAlias)
reshard(t, ksName, "customer", "c2c4", "-80,80-", "-40,40-80,80-c0,c0-",
600, counts, nil, nil, cells, sourceCellOrAlias, 1)
waitForRowCount(t, vtgateConn, ksName, "customer", 20)
query := "insert into customer (name) values('yoko')"
execVtgateQuery(t, vtgateConn, ksName, query)
Expand All @@ -756,7 +757,8 @@ func reshardMerchant2to3SplitMerge(t *testing.T) {
t.Run("reshardMerchant2to3SplitMerge", func(t *testing.T) {
ksName := merchantKeyspace
counts := map[string]int{"zone1-1600": 0, "zone1-1700": 2, "zone1-1800": 0}
reshard(t, ksName, "merchant", "m2m3", "-80,80-", "-40,40-c0,c0-", 1600, counts, dryRunResultsSwitchWritesM2m3, nil, "")
reshard(t, ksName, "merchant", "m2m3", "-80,80-", "-40,40-c0,c0-",
1600, counts, dryRunResultsSwitchReadM2m3, dryRunResultsSwitchWritesM2m3, nil, "", 1)
waitForRowCount(t, vtgateConn, ksName, "merchant", 2)
query := "insert into merchant (mname, category) values('amazon', 'electronics')"
execVtgateQuery(t, vtgateConn, ksName, query)
Expand Down Expand Up @@ -802,7 +804,8 @@ func reshardMerchant3to1Merge(t *testing.T) {
t.Run("reshardMerchant3to1Merge", func(t *testing.T) {
ksName := merchantKeyspace
counts := map[string]int{"zone1-2000": 3}
reshard(t, ksName, "merchant", "m3m1", "-40,40-c0,c0-", "0", 2000, counts, nil, nil, "")
reshard(t, ksName, "merchant", "m3m1", "-40,40-c0,c0-", "0",
2000, counts, nil, nil, nil, "", 1)
waitForRowCount(t, vtgateConn, ksName, "merchant", 3)
query := "insert into merchant (mname, category) values('flipkart', 'electronics')"
execVtgateQuery(t, vtgateConn, ksName, query)
Expand All @@ -814,19 +817,23 @@ func reshardCustomer3to2SplitMerge(t *testing.T) { //-40,40-80,80-c0 => merge/sp
t.Run("reshardCustomer3to2SplitMerge", func(t *testing.T) {
ksName := "customer"
counts := map[string]int{"zone1-1000": 8, "zone1-1100": 8, "zone1-1200": 5}
reshard(t, ksName, "customer", "c4c3", "-40,40-80,80-c0", "-60,60-c0", 1000, counts, nil, nil, "")
reshard(t, ksName, "customer", "c4c3", "-40,40-80,80-c0", "-60,60-c0",
1000, counts, nil, nil, nil, "", 1)
})
}

func reshardCustomer3to1Merge(t *testing.T) { //to unsharded
t.Run("reshardCustomer3to1Merge", func(t *testing.T) {
ksName := "customer"
counts := map[string]int{"zone1-1500": 21}
reshard(t, ksName, "customer", "c3c1", "-60,60-c0,c0-", "0", 1500, counts, nil, nil, "")
reshard(t, ksName, "customer", "c3c1", "-60,60-c0,c0-", "0",
1500, counts, nil, nil, nil, "", 3)
})
}

func reshard(t *testing.T, ksName string, tableName string, workflow string, sourceShards string, targetShards string, tabletIDBase int, counts map[string]int, dryRunResultSwitchWrites []string, cells []*Cell, sourceCellOrAlias string) {
func reshard(t *testing.T, ksName string, tableName string, workflow string, sourceShards string, targetShards string,
tabletIDBase int, counts map[string]int, dryRunResultSwitchReads, dryRunResultSwitchWrites []string, cells []*Cell, sourceCellOrAlias string,
autoIncrementStep int) {
t.Run("reshard", func(t *testing.T) {
if cells == nil {
cells = []*Cell{defaultCell}
Expand Down Expand Up @@ -859,7 +866,10 @@ func reshard(t *testing.T, ksName string, tableName string, workflow string, sou
}
}
vdiff1(t, ksWorkflow, "")
switchReads(t, allCellNames, ksWorkflow)
if dryRunResultSwitchReads != nil {
switchReadsDryRun(t, workflowType, allCellNames, ksWorkflow, dryRunResultSwitchReads)

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Static Code Checks Etc 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_multicell) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_multicell) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_stoponreshard_false) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_stoponreshard_false) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_failover) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_failover) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_v2) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_v2) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_stoponreshard_true) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_stoponreshard_true) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_cellalias) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_cellalias) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_across_db_versions) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_across_db_versions) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_basic) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_basic) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_migrate_vdiff2_convert_tz) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_migrate_vdiff2_convert_tz) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_with_keyspaces_to_watch) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_with_keyspaces_to_watch) 

undefined: workflowType

Check failure on line 870 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Static Code Checks Etc 

undefined: workflowType
}
switchReads(t, workflowType, allCellNames, ksWorkflow, false)

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Static Code Checks Etc 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Static Code Checks Etc 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_multicell) 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_multicell) 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_stoponreshard_false) 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_stoponreshard_false) 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_failover) 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_failover) 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_v2) 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_v2) 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_stoponreshard_true) 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_stoponreshard_true) 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_cellalias) 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_cellalias) 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_across_db_versions) 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_across_db_versions) 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_basic) 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_basic) 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_migrate_vdiff2_convert_tz) 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vreplication_migrate_vdiff2_convert_tz) 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_with_keyspaces_to_watch) 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Run endtoend tests on Cluster (vstream_with_keyspaces_to_watch) 

too many arguments in call to switchReads

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Static Code Checks Etc 

undefined: workflowType

Check failure on line 872 in go/test/endtoend/vreplication/vreplication_test.go

View workflow job for this annotation

GitHub Actions / Static Code Checks Etc 

too many arguments in call to switchReads
if dryRunResultSwitchWrites != nil {
switchWritesDryRun(t, ksWorkflow, dryRunResultSwitchWrites)
}
Expand Down Expand Up @@ -1193,10 +1203,18 @@ func applyVSchema(t *testing.T, vschema, keyspace string) {
require.NoError(t, err)
}

func switchReadsDryRun(t *testing.T, cells, ksWorkflow string, dryRunResults []string) {
output, err := vc.VtctlClient.ExecuteCommandWithOutput("SwitchReads", "--", "--cells="+cells, "--tablet_types=replica", "--dry_run", ksWorkflow)
require.NoError(t, err, fmt.Sprintf("SwitchReads DryRun Error: %s: %s", err, output))
validateDryRunResults(t, output, dryRunResults)
func switchReadsDryRun(t *testing.T, workflowType, cells, ksWorkflow string, dryRunResults []string) {
if workflowType != binlogdatapb.VReplicationWorkflowType_name[int32(binlogdatapb.VReplicationWorkflowType_MoveTables)] &&
workflowType != binlogdatapb.VReplicationWorkflowType_name[int32(binlogdatapb.VReplicationWorkflowType_Reshard)] {
require.FailNowf(t, "Invalid workflow type for SwitchTraffic, must be MoveTables or Reshard",
"workflow type specified: %s", workflowType)
}
output, err := vc.VtctlClient.ExecuteCommandWithOutput(workflowType, "--", "--cells="+cells, "--tablet_types=rdonly,replica",
"--dry_run", "SwitchTraffic", ksWorkflow)
require.NoError(t, err, fmt.Sprintf("Switching Reads DryRun Error: %s: %s", err, output))
if dryRunResults != nil {
validateDryRunResults(t, output, dryRunResults)
}
}

func switchReads(t *testing.T, cells, ksWorkflow string) {
Expand Down
42 changes: 4 additions & 38 deletions go/test/endtoend/vreplication/vreplication_test_env.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -86,42 +86,8 @@ var dryRunResultsSwitchWritesM2m3 = []string{
"Unlock keyspace merchant-type",
}

var dryRunResultsDropSourcesDropCustomerShard = []string{
"Lock keyspace product",
"Lock keyspace customer",
"Dropping these tables from the database and removing them from the vschema for keyspace product:",
" Keyspace product Shard 0 DbName vt_product Tablet 100 Table Lead",
" Keyspace product Shard 0 DbName vt_product Tablet 100 Table Lead-1",
" Keyspace product Shard 0 DbName vt_product Tablet 100 Table customer",
" Keyspace product Shard 0 DbName vt_product Tablet 100 Table db_order_test",
"Denied tables [Lead,Lead-1,customer,db_order_test] will be removed from:",
" Keyspace product Shard 0 Tablet 100",
"Delete reverse vreplication streams on source:",
" Keyspace product Shard 0 Workflow p2c_reverse DbName vt_product Tablet 100",
"Delete vreplication streams on target:",
" Keyspace customer Shard -80 Workflow p2c DbName vt_customer Tablet 200",
" Keyspace customer Shard 80- Workflow p2c DbName vt_customer Tablet 300",
"Routing rules for participating tables will be deleted",
"Unlock keyspace customer",
"Unlock keyspace product",
}

var dryRunResultsDropSourcesRenameCustomerShard = []string{
"Lock keyspace product",
"Lock keyspace customer",
"Renaming these tables from the database and removing them from the vschema for keyspace product:",
" Keyspace product Shard 0 DbName vt_product Tablet 100 Table Lead",
" Keyspace product Shard 0 DbName vt_product Tablet 100 Table Lead-1",
" Keyspace product Shard 0 DbName vt_product Tablet 100 Table customer",
" Keyspace product Shard 0 DbName vt_product Tablet 100 Table db_order_test",
"Denied tables [Lead,Lead-1,customer,db_order_test] will be removed from:",
" Keyspace product Shard 0 Tablet 100",
"Delete reverse vreplication streams on source:",
" Keyspace product Shard 0 Workflow p2c_reverse DbName vt_product Tablet 100",
"Delete vreplication streams on target:",
" Keyspace customer Shard -80 Workflow p2c DbName vt_customer Tablet 200",
" Keyspace customer Shard 80- Workflow p2c DbName vt_customer Tablet 300",
"Routing rules for participating tables will be deleted",
"Unlock keyspace customer",
"Unlock keyspace product",
var dryRunResultsSwitchReadM2m3 = []string{
"Lock keyspace merchant-type",
"Switch reads from keyspace merchant-type to keyspace merchant-type for shards -80,80- to shards -40,40-c0,c0-",
"Unlock keyspace merchant-type",
}
5 changes: 3 additions & 2 deletions go/test/endtoend/vreplication/vstream_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -369,7 +369,8 @@ func testVStreamStopOnReshardFlag(t *testing.T, stopOnReshard bool, baseTabletID
tickCount++
switch tickCount {
case 1:
reshard(t, "sharded", "customer", "vstreamStopOnReshard", "-80,80-", "-40,40-", baseTabletID+400, nil, nil, nil, defaultCellName)
reshard(t, "sharded", "customer", "vstreamStopOnReshard", "-80,80-",
"-40,40-", baseTabletID+400, nil, nil, nil, nil, defaultCellName, 1)
case 60:
done = true
}
Expand Down Expand Up @@ -499,7 +500,7 @@ func testVStreamCopyMultiKeyspaceReshard(t *testing.T, baseTabletID int) numEven
tickCount++
switch tickCount {
case 1:
reshard(t, "sharded", "customer", "vstreamCopyMultiKeyspaceReshard", "-80,80-", "-40,40-", baseTabletID+400, nil, nil, nil, defaultCellName)
reshard(t, "sharded", "customer", "vstreamCopyMultiKeyspaceReshard", "-80,80-", "-40,40-", baseTabletID+400, nil, nil, nil, nil, defaultCellName, 1)
reshardDone = true
case 60:
done = true
Expand Down
2 changes: 1 addition & 1 deletion go/vt/wrangler/traffic_switcher.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -397,7 +397,7 @@ func (wr *Wrangler) SwitchReads(ctx context.Context, targetKeyspace, workflowNam
return sw.logs(), nil
}
wr.Logger().Infof("About to switchShardReads: %+v, %+v, %+v", cells, servedTypes, direction)
if err := ts.switchShardReads(ctx, cells, servedTypes, direction); err != nil {
if err := sw.switchShardReads(ctx, cells, servedTypes, direction); err != nil {
ts.Logger().Errorf("switchShardReads failed: %v", err)
return nil, err
}
Expand Down

0 comments on commit 0e637c0

Please sign in to comment.