Skip to content

Commit 3b24612

Browse files
mattlordmattlord
committed
Allow complete when keyspace is completely gone
Signed-off-by: Matt Lord <mattalord@gmail.com>
1 parent 59658e4 commit 3b24612

File tree

4 files changed

+117
-32
lines changed

4 files changed

+117
-32
lines changed

‎go/vt/vtctl/workflow/server.go

Lines changed: 73 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -428,8 +428,12 @@ func (s *Server) GetWorkflowState(ctx context.Context, targetKeyspace, workflowN
428428
return s.getWorkflowState(ctx, targetKeyspace, workflowName)
429429
}
430430

431-
func (s *Server) getWorkflowState(ctx context.Context, targetKeyspace, workflowName string) (*trafficSwitcher, *State, error) {
432-
ts, err := s.buildTrafficSwitcher(ctx, targetKeyspace, workflowName)
431+
func (s *Server) getWorkflowState(ctx context.Context, targetKeyspace, workflowName string, opts ...WorkflowOption) (*trafficSwitcher, *State, error) {
432+
var options workflowOptions
433+
for _, o := range opts {
434+
o.apply(&options)
435+
}
436+
ts, err := s.buildTrafficSwitcher(ctx, targetKeyspace, workflowName, opts...)
433437
if err != nil {
434438
s.Logger().Errorf("buildTrafficSwitcher failed: %v", err)
435439
return nil, nil, err
@@ -442,6 +446,13 @@ func (s *Server) getWorkflowState(ctx context.Context, targetKeyspace, workflowN
442446
IsPartialMigration: ts.isPartialMigration,
443447
}
444448

449+
if ts.workflowType == binlogdatapb.VReplicationWorkflowType_MoveTables && options.ignoreSourceKeyspace {
450+
if err := s.updateTablesTrafficState(ctx, state, ts.tables); err != nil {
451+
return nil, nil, err
452+
}
453+
return ts, state, nil
454+
}
455+
445456
if ts.workflowType == binlogdatapb.VReplicationWorkflowType_CreateLookupIndex {
446457
// Nothing left to do.
447458
return ts, state, nil
@@ -471,7 +482,6 @@ func (s *Server) getWorkflowState(ctx context.Context, targetKeyspace, workflowN
471482
if len(ts.Tables()) == 0 {
472483
return nil, nil, vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "no tables in workflow %s.%s", targetKeyspace, workflowName)
473484
}
474-
table := ts.Tables()[0]
475485

476486
if ts.IsMultiTenantMigration() {
477487
// Deduce which traffic has been switched by looking at the current keyspace routing rules.
@@ -497,29 +507,9 @@ func (s *Server) getWorkflowState(ctx context.Context, targetKeyspace, workflowN
497507
}
498508
}
499509
} else {
500-
state.RdonlyCellsSwitched, state.RdonlyCellsNotSwitched, err = s.GetCellsWithTableReadsSwitched(ctx, sourceKeyspace, targetKeyspace, table, topodatapb.TabletType_RDONLY)
501-
if err != nil {
502-
return nil, nil, err
503-
}
504-
505-
state.ReplicaCellsSwitched, state.ReplicaCellsNotSwitched, err = s.GetCellsWithTableReadsSwitched(ctx, sourceKeyspace, targetKeyspace, table, topodatapb.TabletType_REPLICA)
506-
if err != nil {
510+
if err := s.updateTablesTrafficState(ctx, state, ts.tables); err != nil {
507511
return nil, nil, err
508512
}
509-
globalRules, err := topotools.GetRoutingRules(ctx, ts.TopoServer())
510-
if err != nil {
511-
return nil, nil, err
512-
}
513-
for _, table := range ts.Tables() {
514-
// If a rule for the primary tablet type exists for any table and points to the target keyspace,
515-
// then writes have been switched.
516-
ruleKey := fmt.Sprintf("%s.%s", sourceKeyspace, table)
517-
rr := globalRules[ruleKey]
518-
if len(rr) > 0 && rr[0] != ruleKey {
519-
state.WritesSwitched = true
520-
break
521-
}
522-
}
523513
}
524514
} else {
525515
state.WorkflowType = TypeReshard
@@ -1076,7 +1066,7 @@ func (s *Server) moveTablesCreate(ctx context.Context, req *vtctldatapb.MoveTabl
10761066
err = vterrors.Wrapf(err, "failed to cleanup denied table entries: %v", cerr)
10771067
}
10781068
}
1079-
if cerr := s.dropArtifacts(ctx, false, false, &switcher{s: s, ts: ts}); cerr != nil {
1069+
if cerr := s.dropArtifacts(ctx, false, &switcher{s: s, ts: ts}); cerr != nil {
10801070
err = vterrors.Wrapf(err, "failed to cleanup workflow artifacts: %v", cerr)
10811071
}
10821072
if origVSchema == nil { // There's no previous version to restore
@@ -1249,7 +1239,11 @@ func (s *Server) MoveTablesComplete(ctx context.Context, req *vtctldatapb.MoveTa
12491239
span, ctx := trace.NewSpan(ctx, "workflow.Server.MoveTablesComplete")
12501240
defer span.Finish()
12511241

1252-
ts, state, err := s.getWorkflowState(ctx, req.GetTargetKeyspace(), req.GetWorkflow())
1242+
opts := []WorkflowOption{}
1243+
if req.IgnoreSourceKeyspace {
1244+
opts = append(opts, IgnoreSourceKeyspace())
1245+
}
1246+
ts, state, err := s.getWorkflowState(ctx, req.GetTargetKeyspace(), req.GetWorkflow(), opts...)
12531247
if err != nil {
12541248
return nil, err
12551249
}
@@ -1291,7 +1285,7 @@ func (s *Server) MoveTablesComplete(ctx context.Context, req *vtctldatapb.MoveTa
12911285
}
12921286

12931287
if req.IgnoreSourceKeyspace {
1294-
if err := s.dropArtifacts(ctx, req.KeepRoutingRules, true, &switcher{s: s, ts: ts}); err != nil {
1288+
if err := s.dropArtifacts(ctx, req.KeepRoutingRules, &switcher{s: s, ts: ts}, opts...); err != nil {
12951289
return nil, vterrors.Wrapf(err, "failed to cleanup workflow artifacts")
12961290
}
12971291
if err := ts.TopoServer().RebuildSrvVSchema(ctx, nil); err != nil {
@@ -2053,7 +2047,11 @@ func (s *Server) deleteTenantData(ctx context.Context, ts *trafficSwitcher, batc
20532047
})
20542048
}
20552049

2056-
func (s *Server) buildTrafficSwitcher(ctx context.Context, targetKeyspace, workflowName string) (*trafficSwitcher, error) {
2050+
func (s *Server) buildTrafficSwitcher(ctx context.Context, targetKeyspace, workflowName string, opts ...WorkflowOption) (*trafficSwitcher, error) {
2051+
var options workflowOptions
2052+
for _, o := range opts {
2053+
o.apply(&options)
2054+
}
20572055
tgtInfo, err := BuildTargets(ctx, s.ts, s.tmc, targetKeyspace, workflowName)
20582056
if err != nil {
20592057
s.Logger().Infof("Error building targets: %s", err)
@@ -2120,6 +2118,9 @@ func (s *Server) buildTrafficSwitcher(ctx context.Context, targetKeyspace, workf
21202118
}
21212119
}
21222120

2121+
if options.ignoreSourceKeyspace {
2122+
continue
2123+
}
21232124
if _, ok := ts.sources[bls.Shard]; ok {
21242125
continue
21252126
}
@@ -2152,6 +2153,10 @@ func (s *Server) buildTrafficSwitcher(ctx context.Context, targetKeyspace, workf
21522153
}
21532154
}
21542155
}
2156+
if ts.workflowType == binlogdatapb.VReplicationWorkflowType_MoveTables && options.ignoreSourceKeyspace {
2157+
log.Errorf("DEBUG: Ignoring source keyspace for MoveTables workflow with source Keyspace %s", ts.sourceKeyspace)
2158+
return ts, nil
2159+
}
21552160
vs, err := sourceTopo.GetVSchema(ctx, ts.sourceKeyspace)
21562161
if err != nil {
21572162
return nil, err
@@ -2247,7 +2252,7 @@ func (s *Server) dropSources(ctx context.Context, ts *trafficSwitcher, removalTy
22472252
}
22482253
}
22492254
}
2250-
if err := s.dropArtifacts(ctx, keepRoutingRules, false, sw); err != nil {
2255+
if err := s.dropArtifacts(ctx, keepRoutingRules, sw); err != nil {
22512256
return nil, err
22522257
}
22532258
if err := ts.TopoServer().RebuildSrvVSchema(ctx, nil); err != nil {
@@ -2257,8 +2262,12 @@ func (s *Server) dropSources(ctx context.Context, ts *trafficSwitcher, removalTy
22572262
return sw.logs(), nil
22582263
}
22592264

2260-
func (s *Server) dropArtifacts(ctx context.Context, keepRoutingRules, ignoreSourceKeyspace bool, sw iswitcher) error {
2261-
if !ignoreSourceKeyspace {
2265+
func (s *Server) dropArtifacts(ctx context.Context, keepRoutingRules bool, sw iswitcher, opts ...WorkflowOption) error {
2266+
var options workflowOptions
2267+
for _, o := range opts {
2268+
o.apply(&options)
2269+
}
2270+
if !options.ignoreSourceKeyspace {
22622271
if err := sw.dropSourceReverseVReplicationStreams(ctx); err != nil {
22632272
return err
22642273
}
@@ -3456,6 +3465,39 @@ func (s *Server) validateShardsHaveVReplicationPermissions(ctx context.Context,
34563465
return nil
34573466
}
34583467

3468+
func (s *Server) updateTablesTrafficState(ctx context.Context, state *State, tables []string) error {
3469+
// We assume a consistent state, so only choose routing rule for one table.
3470+
if len(tables) == 0 {
3471+
return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "no tables in workflow %s.%s", state.TargetKeyspace, state.Workflow)
3472+
}
3473+
3474+
var err error
3475+
state.RdonlyCellsSwitched, state.RdonlyCellsNotSwitched, err = s.GetCellsWithTableReadsSwitched(ctx, state.SourceKeyspace, state.TargetKeyspace, tables[0], topodatapb.TabletType_RDONLY)
3476+
if err != nil {
3477+
return err
3478+
}
3479+
state.ReplicaCellsSwitched, state.ReplicaCellsNotSwitched, err = s.GetCellsWithTableReadsSwitched(ctx, state.SourceKeyspace, state.TargetKeyspace, tables[0], topodatapb.TabletType_REPLICA)
3480+
if err != nil {
3481+
return err
3482+
}
3483+
globalRules, err := topotools.GetRoutingRules(ctx, s.ts)
3484+
if err != nil {
3485+
return err
3486+
}
3487+
for _, table := range tables {
3488+
// If a rule for the primary tablet type exists for any table and points to the target keyspace,
3489+
// then writes have been switched.
3490+
ruleKey := fmt.Sprintf("%s.%s", state.SourceKeyspace, table)
3491+
rr := globalRules[ruleKey]
3492+
if len(rr) > 0 && rr[0] != ruleKey {
3493+
state.WritesSwitched = true
3494+
break
3495+
}
3496+
}
3497+
log.Errorf("DEBUG: state: %+v", state)
3498+
return err
3499+
}
3500+
34593501
func (s *Server) Logger() logutil.Logger {
34603502
if s.options.logger == nil {
34613503
s.options.logger = logutil.NewConsoleLogger() // Use default system logger

‎go/vt/vtctl/workflow/server_options.go

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,3 +54,38 @@ func WithLogger(l logutil.Logger) ServerOption {
5454
o.logger = l
5555
})
5656
}
57+
58+
// workflowOptions configure a workflow's optional behavior when
59+
// performing actions in the worfklow server.
60+
// workflowOptions are set by the WorkflowOption values passed
61+
// to the server functions.
62+
type workflowOptions struct {
63+
ignoreSourceKeyspace bool
64+
}
65+
66+
// WorkflowOption alters how we perform the certain workflow operations.
67+
type WorkflowOption interface {
68+
apply(*workflowOptions)
69+
}
70+
71+
// funcWorkflowOption wraps a function that modifies workflowOptions into
72+
// an implementation of the WorkflowOption interface.
73+
type funcWorkflowOption struct {
74+
f func(*workflowOptions)
75+
}
76+
77+
func (fwo *funcWorkflowOption) apply(wo *workflowOptions) {
78+
fwo.f(wo)
79+
}
80+
81+
func newFuncWorkflowOption(f func(*workflowOptions)) *funcWorkflowOption {
82+
return &funcWorkflowOption{
83+
f: f,
84+
}
85+
}
86+
87+
func IgnoreSourceKeyspace() WorkflowOption {
88+
return newFuncWorkflowOption(func(o *workflowOptions) {
89+
o.ignoreSourceKeyspace = true
90+
})
91+
}

‎go/vt/vtctl/workflow/server_test.go

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -387,6 +387,14 @@ func TestMoveTablesComplete(t *testing.T) {
387387
Workflow: workflowName,
388388
IgnoreSourceKeyspace: true,
389389
},
390+
preFunc: func(t *testing.T, env *testEnv) {
391+
err := env.ts.DeleteKeyspace(ctx, sourceKeyspaceName)
392+
require.NoError(t, err)
393+
},
394+
postFunc: func(t *testing.T, env *testEnv) {
395+
err := env.ts.CreateKeyspace(ctx, sourceKeyspaceName, &topodatapb.Keyspace{})
396+
require.NoError(t, err)
397+
},
390398
expectedTargetQueries: []*queryResult{
391399
{
392400
query: fmt.Sprintf("delete from _vt.vreplication where db_name = 'vt_%s' and workflow = '%s'",

‎go/vt/vtctl/workflow/traffic_switcher.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -268,7 +268,7 @@ func (ts *trafficSwitcher) ExternalTopo() *topo.Server { ret
268268
func (ts *trafficSwitcher) MigrationType() binlogdatapb.MigrationType { return ts.migrationType }
269269
func (ts *trafficSwitcher) IsPartialMigration() bool { return ts.isPartialMigration }
270270
func (ts *trafficSwitcher) ReverseWorkflowName() string { return ts.reverseWorkflow }
271-
func (ts *trafficSwitcher) SourceKeyspaceName() string { return ts.sourceKSSchema.Keyspace.Name }
271+
func (ts *trafficSwitcher) SourceKeyspaceName() string { return ts.sourceKeyspace }
272272
func (ts *trafficSwitcher) SourceKeyspaceSchema() *vindexes.KeyspaceSchema { return ts.sourceKSSchema }
273273
func (ts *trafficSwitcher) Sources() map[string]*MigrationSource { return ts.sources }
274274
func (ts *trafficSwitcher) Tables() []string { return ts.tables }

0 commit comments

Comments
 (0)