[reparentutil / wrangler] Extract PlannedReparentShard logic from wrangler to PlannedReparenter struct

go/test/endtoend/reparent/reparent_test.go

@@ -204,7 +204,7 @@ func TestReparentReplicaOffline(t *testing.T) {
  // Perform a graceful reparent operation.
  out, err := prsWithTimeout(t, tab2, false, "", "31s")
  require.Error(t, err)
- assert.Contains(t, out, fmt.Sprintf("tablet %s SetMaster failed", tab4.Alias))
+ assert.Contains(t, out, fmt.Sprintf("tablet %s failed to SetMaster", tab4.Alias))
  checkMasterTablet(t, tab2)
 }
 
@@ -345,7 +345,7 @@ func TestReparentWithDownReplica(t *testing.T) {
  // Perform a graceful reparent operation. It will fail as one tablet is down.
  out, err := prs(t, tab2)
  require.Error(t, err)
- assert.Contains(t, out, fmt.Sprintf("tablet %s SetMaster failed", tab3.Alias))
+ assert.Contains(t, out, fmt.Sprintf("tablet %s failed to SetMaster", tab3.Alias))
 
  // insert data into the new master, check the connected replica work
  confirmReplication(t, tab2, []*cluster.Vttablet{tab1, tab4})
@@ -441,5 +441,5 @@ func TestReparentDoesntHangIfMasterFails(t *testing.T) {
  // insert. The replicas should then abort right away.
  out, err := prs(t, tab2)
  require.Error(t, err)
- assert.Contains(t, out, "master failed to PopulateReparentJournal")
+ assert.Contains(t, out, "primary failed to PopulateReparentJournal")
 }

go/vt/vtctl/grpcvtctldserver/testutil/util.go

@@ -20,6 +20,7 @@ package testutil
 
 import (
  "context"
+ "errors"
  "fmt"
  "testing"
 
@@ -87,6 +88,10 @@ type AddTabletOptions struct {
  // update the shard record to make that tablet the primary, and fail the
  // test if the shard record has a serving primary already.
  AlsoSetShardMaster bool
+ // ForceSetShardMaster, when combined with AlsoSetShardMaster, will ignore
+ // any existing primary in the shard, making the current tablet the serving
+ // primary (given it is type MASTER), and log that it has done so.
+ ForceSetShardMaster bool
  // SkipShardCreation, when set, makes AddTablet never attempt to create a
  // shard record in the topo under any circumstances.
  SkipShardCreation bool
@@ -136,7 +141,13 @@ func AddTablet(ctx context.Context, t *testing.T, ts *topo.Server, tablet *topod
  if tablet.Type == topodatapb.TabletType_MASTER && opts.AlsoSetShardMaster {
  _, err := ts.UpdateShardFields(ctx, tablet.Keyspace, tablet.Shard, func(si *topo.ShardInfo) error {
  if si.IsMasterServing && si.MasterAlias != nil {
- return fmt.Errorf("shard %v/%v already has a serving master (%v)", tablet.Keyspace, tablet.Shard, topoproto.TabletAliasString(si.MasterAlias))
+ msg := fmt.Sprintf("shard %v/%v already has a serving master (%v)", tablet.Keyspace, tablet.Shard, topoproto.TabletAliasString(si.MasterAlias))
+
+ if !opts.ForceSetShardMaster {
+ return errors.New(msg)
+ }
+
+ t.Logf("%s; replacing with %v because ForceSetShardMaster = true", msg, topoproto.TabletAliasString(tablet.Alias))
  }
 
  si.MasterAlias = tablet.Alias

go/vt/vtctl/reparentutil/emergency_reparenter_test.go

@@ -1650,18 +1650,47 @@ func TestEmergencyReparenter_waitForAllRelayLogsToApply(t *testing.T) {
 
 type emergencyReparenterTestTMClient struct {
  tmclient.TabletManagerClient
+ // keyed by tablet alias.
+ DemoteMasterDelays map[string]time.Duration
+ // keyed by tablet alias.
+ DemoteMasterResults map[string]struct {
+ Status *replicationdatapb.MasterStatus
+ Error error
+ }
+ // keyed by tablet alias.
+ MasterPositionDelays map[string]time.Duration
+ // keyed by tablet alias.
+ MasterPositionResults map[string]struct {
+ Position string
+ Error error
+ }
+ // keyed by tablet alias.
+ PopulateReparentJournalDelays map[string]time.Duration
  // keyed by tablet alias
  PopulateReparentJournalResults map[string]error
  // keyed by tablet alias.
+ PromoteReplicaDelays map[string]time.Duration
+ // keyed by tablet alias. injects a sleep to the end of the function
+ // regardless of parent context timeout or error result.
+ PromoteReplicaPostDelays map[string]time.Duration
+ // keyed by tablet alias.
  PromoteReplicaResults map[string]struct {
  Result string
  Error error
  }
+ ReplicationStatusResults map[string]struct {
+ Position *replicationdatapb.Status
+ Error error
+ }
  // keyed by tablet alias.
  SetMasterDelays map[string]time.Duration
  // keyed by tablet alias.
  SetMasterResults map[string]error
  // keyed by tablet alias.
+ SetReadWriteDelays map[string]time.Duration
+ // keyed by tablet alias.
+ SetReadWriteResults map[string]error
+ // keyed by tablet alias.
  StopReplicationAndGetStatusDelays map[string]time.Duration
  // keyed by tablet alias.
  StopReplicationAndGetStatusResults map[string]struct {
@@ -1671,9 +1700,74 @@ type emergencyReparenterTestTMClient struct {
  }
  // keyed by tablet alias.
  WaitForPositionDelays map[string]time.Duration
+ // keyed by tablet alias. injects a sleep to the end of the function
+ // regardless of parent context timeout or error result.
+ WaitForPositionPostDelays map[string]time.Duration
  // WaitForPosition(tablet *topodatapb.Tablet, position string) error, so we
  // key by tablet alias and then by position.
  WaitForPositionResults map[string]map[string]error
+ // keyed by tablet alias.
+ UndoDemoteMasterDelays map[string]time.Duration
+ // keyed by tablet alias
+ UndoDemoteMasterResults map[string]error
+}
+
+func (fake *emergencyReparenterTestTMClient) DemoteMaster(ctx context.Context, tablet *topodatapb.Tablet) (*replicationdatapb.MasterStatus, error) {
+ if fake.DemoteMasterResults == nil {
+ return nil, assert.AnError
+ }
+
+ if tablet.Alias == nil {
+ return nil, assert.AnError
+ }
+
+ key := topoproto.TabletAliasString(tablet.Alias)
+
+ if fake.DemoteMasterDelays != nil {
+ if delay, ok := fake.DemoteMasterDelays[key]; ok {
+ select {
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-time.After(delay):
+ // proceed to results
+ }
+ }
+ }
+
+ if result, ok := fake.DemoteMasterResults[key]; ok {
+ return result.Status, result.Error
+ }
+
+ return nil, assert.AnError
+}
+
+func (fake *emergencyReparenterTestTMClient) MasterPosition(ctx context.Context, tablet *topodatapb.Tablet) (string, error) {
+ if fake.MasterPositionResults == nil {
+ return "", assert.AnError
+ }
+
+ if tablet.Alias == nil {
+ return "", assert.AnError
+ }
+
+ key := topoproto.TabletAliasString(tablet.Alias)
+
+ if fake.MasterPositionDelays != nil {
+ if delay, ok := fake.MasterPositionDelays[key]; ok {
+ select {
+ case <-ctx.Done():
+ return "", ctx.Err()
+ case <-time.After(delay):
+ // proceed to results
+ }
+ }
+ }
+
+ if result, ok := fake.MasterPositionResults[key]; ok {
+ return result.Position, result.Error
+ }
+
+ return "", assert.AnError
 }
 
 func (fake *emergencyReparenterTestTMClient) PopulateReparentJournal(ctx context.Context, tablet *topodatapb.Tablet, timeCreatedNS int64, actionName string, primaryAlias *topodatapb.TabletAlias, pos string) error {
@@ -1682,6 +1776,17 @@ func (fake *emergencyReparenterTestTMClient) PopulateReparentJournal(ctx context
  }
 
  key := topoproto.TabletAliasString(tablet.Alias)
+
+ if fake.PopulateReparentJournalDelays != nil {
+ if delay, ok := fake.PopulateReparentJournalDelays[key]; ok {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-time.After(delay):
+ // proceed to results
+ }
+ }
+ }
  if result, ok := fake.PopulateReparentJournalResults[key]; ok {
  return result
  }
@@ -1695,13 +1800,49 @@ func (fake *emergencyReparenterTestTMClient) PromoteReplica(ctx context.Context,
  }
 
  key := topoproto.TabletAliasString(tablet.Alias)
+
+ defer func() {
+ if fake.PromoteReplicaPostDelays == nil {
+ return
+ }
+
+ if delay, ok := fake.PromoteReplicaPostDelays[key]; ok {
+ time.Sleep(delay)
+ }
+ }()
+
+ if fake.PromoteReplicaDelays != nil {
+ if delay, ok := fake.PromoteReplicaDelays[key]; ok {
+ select {
+ case <-ctx.Done():
+ return "", ctx.Err()
+ case <-time.After(delay):
+ // proceed to results
+ }
+ }
+ }
+
  if result, ok := fake.PromoteReplicaResults[key]; ok {
  return result.Result, result.Error
  }
 
  return "", assert.AnError
 }
 
+func (fake *emergencyReparenterTestTMClient) ReplicationStatus(ctx context.Context, tablet *topodatapb.Tablet) (*replicationdatapb.Status, error) {
+ if fake.ReplicationStatusResults == nil {
+ return nil, assert.AnError
+ }
+
+ key := topoproto.TabletAliasString(tablet.Alias)
+
+ if result, ok := fake.ReplicationStatusResults[key]; ok {
+ return result.Position, result.Error
+ }
+
+ return nil, assert.AnError
+}
+
 func (fake *emergencyReparenterTestTMClient) SetMaster(ctx context.Context, tablet *topodatapb.Tablet, parent *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartReplication bool) error {
  if fake.SetMasterResults == nil {
  return assert.AnError
@@ -1727,6 +1868,35 @@ func (fake *emergencyReparenterTestTMClient) SetMaster(ctx context.Context, tabl
  return assert.AnError
 }
 
+func (fake *emergencyReparenterTestTMClient) SetReadWrite(ctx context.Context, tablet *topodatapb.Tablet) error {
+ if fake.SetReadWriteResults == nil {
+ return assert.AnError
+ }
+
+ if tablet.Alias == nil {
+ return assert.AnError
+ }
+
+ key := topoproto.TabletAliasString(tablet.Alias)
+
+ if fake.SetReadWriteDelays != nil {
+ if delay, ok := fake.SetReadWriteDelays[key]; ok {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-time.After(delay):
+ // proceed to results
+ }
+ }
+ }
+
+ if err, ok := fake.SetReadWriteResults[key]; ok {
+ return err
+ }
+
+ return assert.AnError
+}
+
 func (fake *emergencyReparenterTestTMClient) StopReplicationAndGetStatus(ctx context.Context, tablet *topodatapb.Tablet, mode replicationdatapb.StopReplicationMode) (*replicationdatapb.Status, *replicationdatapb.StopReplicationStatus, error) {
  if fake.StopReplicationAndGetStatusResults == nil {
  return nil, nil, assert.AnError
@@ -1759,6 +1929,16 @@ func (fake *emergencyReparenterTestTMClient) StopReplicationAndGetStatus(ctx con
 func (fake *emergencyReparenterTestTMClient) WaitForPosition(ctx context.Context, tablet *topodatapb.Tablet, position string) error {
  tabletKey := topoproto.TabletAliasString(tablet.Alias)
 
+ defer func() {
+ if fake.WaitForPositionPostDelays == nil {
+ return
+ }
+
+ if delay, ok := fake.WaitForPositionPostDelays[tabletKey]; ok {
+ time.Sleep(delay)
+ }
+ }()
+
  if fake.WaitForPositionDelays != nil {
  if delay, ok := fake.WaitForPositionDelays[tabletKey]; ok {
  select {
@@ -1786,3 +1966,32 @@ func (fake *emergencyReparenterTestTMClient) WaitForPosition(ctx context.Context
 
  return result
 }
+
+func (fake *emergencyReparenterTestTMClient) UndoDemoteMaster(ctx context.Context, tablet *topodatapb.Tablet) error {
+ if fake.UndoDemoteMasterResults == nil {
+ return assert.AnError
+ }
+
+ if tablet.Alias == nil {
+ return assert.AnError
+ }
+
+ key := topoproto.TabletAliasString(tablet.Alias)
+
+ if fake.UndoDemoteMasterDelays != nil {
+ if delay, ok := fake.UndoDemoteMasterDelays[key]; ok {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-time.After(delay):
+ // proceed to results
+ }
+ }
+ }
+
+ if result, ok := fake.UndoDemoteMasterResults[key]; ok {
+ return result
+ }
+
+ return assert.AnError
+}