Skip to content

Commit

Permalink
Improve force handling
Browse files Browse the repository at this point in the history
Signed-off-by: Matt Lord <mattalord@gmail.com>
  • Loading branch information
mattlord committed Sep 7, 2024
1 parent d73f546 commit 740ef56
Show file tree
Hide file tree
Showing 2 changed files with 13 additions and 12 deletions.
23 changes: 12 additions & 11 deletions go/vt/vtctl/workflow/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -2983,7 +2983,7 @@ func (s *Server) updateShardRecords(ctx context.Context, keyspace string, shards
}

// refreshPrimaryTablets will just RPC-ping all the primary tablets with RefreshState
func (s *Server) refreshPrimaryTablets(ctx context.Context, shards []*topo.ShardInfo) error {
func (s *Server) refreshPrimaryTablets(ctx context.Context, shards []*topo.ShardInfo, force bool) error {
wg := sync.WaitGroup{}
rec := concurrency.AllErrorRecorder{}
for _, si := range shards {
Expand All @@ -2996,10 +2996,10 @@ func (s *Server) refreshPrimaryTablets(ctx context.Context, shards []*topo.Shard
return
}

if err := s.tmc.RefreshState(ctx, ti.Tablet); err != nil {
if err := s.tmc.RefreshState(ctx, ti.Tablet); err != nil && !force {
rec.RecordError(err)
} else {
s.Logger().Infof("%v responded", topoproto.TabletAliasString(si.PrimaryAlias))
s.Logger().Infof("%v responded (err: %v)", topoproto.TabletAliasString(si.PrimaryAlias), err)
}
}(si)
}
Expand Down Expand Up @@ -3089,13 +3089,6 @@ func (s *Server) WorkflowSwitchTraffic(ctx context.Context, req *vtctldatapb.Wor
return nil, err
}

if req.GetForce() {
if ts.options == nil {
ts.options = &vtctldatapb.WorkflowOptions{}
}
ts.options.WarnOnPartialTabletRefresh = true
}

if startState.WorkflowType == TypeMigrate {
return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "invalid action for Migrate workflow: SwitchTraffic")
}
Expand All @@ -3121,6 +3114,14 @@ func (s *Server) WorkflowSwitchTraffic(ctx context.Context, req *vtctldatapb.Wor
return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "cannot reverse traffic for multi-tenant migrations")
}
}

if req.GetForce() {
if ts.options == nil {
ts.options = &vtctldatapb.WorkflowOptions{}
}
ts.options.WarnOnPartialTabletRefresh = true
}

reason, err := s.canSwitch(ctx, ts, startState, direction, int64(maxReplicationLagAllowed.Seconds()), req.GetShards(), req.GetForce())
if err != nil {
return nil, err
Expand Down Expand Up @@ -3275,14 +3276,14 @@ func (s *Server) switchReads(ctx context.Context, req *vtctldatapb.WorkflowSwitc
}
}

// If journals exist notify user and fail.
journalsExist, _, err := ts.checkJournals(ctx)
if err != nil && !req.GetForce() {
return defaultErrorHandler(ts.Logger(), fmt.Sprintf("failed to read journal in the %s keyspace", ts.SourceKeyspaceName()), err)
}
if journalsExist {
s.Logger().Infof("Found a previous journal entry for %d", ts.id)
}

var sw iswitcher
if req.DryRun {
sw = &switcherDryRun{ts: ts, drLog: NewLogRecorder()}
Expand Down
2 changes: 1 addition & 1 deletion go/vt/vtctl/workflow/traffic_switcher.go
Original file line number Diff line number Diff line change
Expand Up @@ -727,7 +727,7 @@ func (ts *trafficSwitcher) changeShardsAccess(ctx context.Context, keyspace stri
if err := ts.TopoServer().UpdateDisableQueryService(ctx, keyspace, shards, topodatapb.TabletType_PRIMARY, nil, access == disallowWrites /* disable */); err != nil {
return err
}
return ts.ws.refreshPrimaryTablets(ctx, shards)
return ts.ws.refreshPrimaryTablets(ctx, shards, ts.options != nil && ts.options.WarnOnPartialTabletRefresh)
}

func (ts *trafficSwitcher) allowTargetWrites(ctx context.Context) error {
Expand Down

0 comments on commit 740ef56

Please sign in to comment.