PRS: Measure replication progress instead of lag.

Signed-off-by: Anthony Yeh <enisoc@planetscale.com>

go/vt/schemamanager/schemaswap/schema_swap.go

@@ -1300,7 +1300,6 @@ func (shardSwap *shardSchemaSwap) reparentFromMaster(masterTablet *topodatapb.Ta
 			shardSwap.shardName,
 			nil,                /* masterElectTabletAlias */
 			masterTablet.Alias, /* avoidMasterAlias */
-			*reparentTimeout,
 			*reparentTimeout)
 		if err != nil {
 			return err

go/vt/vtcombo/tablet_map.go

@@ -694,7 +694,7 @@ func (itmc *internalTabletManagerClient) SlaveWasPromoted(ctx context.Context, t
 	return fmt.Errorf("not implemented in vtcombo")
 }
 
-func (itmc *internalTabletManagerClient) SetMaster(ctx context.Context, tablet *topodatapb.Tablet, parent *topodatapb.TabletAlias, timeCreatedNS int64, forceStartSlave bool) error {
+func (itmc *internalTabletManagerClient) SetMaster(ctx context.Context, tablet *topodatapb.Tablet, parent *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartSlave bool) error {
 	return fmt.Errorf("not implemented in vtcombo")
 }
 

go/vt/vtctl/reparent.go

@@ -45,7 +45,7 @@ func init() {
 	addCommand("Shards", command{
 		"PlannedReparentShard",
 		commandPlannedReparentShard,
-		"-keyspace_shard=<keyspace/shard> [-new_master=<tablet alias>] [-avoid_master=<tablet alias>] [-wait_slave_timeout=<duration>] [-lag_threshold=<duration>]",
+		"-keyspace_shard=<keyspace/shard> [-new_master=<tablet alias>] [-avoid_master=<tablet alias>] [-wait_slave_timeout=<duration>]",
 		"Reparents the shard to the new master, or away from old master. Both old and new master need to be up and running."})
 	addCommand("Shards", command{
 		"EmergencyReparentShard",
@@ -107,8 +107,7 @@ func commandPlannedReparentShard(ctx context.Context, wr *wrangler.Wrangler, sub
 		return fmt.Errorf("active reparent commands disabled (unset the -disable_active_reparents flag to enable)")
 	}
 
-	waitSlaveTimeout := subFlags.Duration("wait_slave_timeout", *topo.RemoteOperationTimeout, "time to wait for replicas to catch up after reparenting")
-	lagThreshold := subFlags.Duration("lag_threshold", *topo.RemoteOperationTimeout, "require the new master's replication lag to be within this threshold before attempting a reparent; this avoids disrupting the current master if the new master is not likely to catch up in time. (0 means always attempt, regardless of lag)")
+	waitSlaveTimeout := subFlags.Duration("wait_slave_timeout", *topo.RemoteOperationTimeout, "time to wait for replicas to catch up on replication before and after reparenting")
 	keyspaceShard := subFlags.String("keyspace_shard", "", "keyspace/shard of the shard that needs to be reparented")
 	newMaster := subFlags.String("new_master", "", "alias of a tablet that should be the new master")
 	avoidMaster := subFlags.String("avoid_master", "", "alias of a tablet that should not be the master, i.e. reparent to any other tablet if this one is the master")
@@ -143,7 +142,7 @@ func commandPlannedReparentShard(ctx context.Context, wr *wrangler.Wrangler, sub
 			return err
 		}
 	}
-	return wr.PlannedReparentShard(ctx, keyspace, shard, newMasterAlias, avoidMasterAlias, *waitSlaveTimeout, *lagThreshold)
+	return wr.PlannedReparentShard(ctx, keyspace, shard, newMasterAlias, avoidMasterAlias, *waitSlaveTimeout)
 }
 
 func commandEmergencyReparentShard(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag.FlagSet, args []string) error {

go/vt/vttablet/agentrpctest/test_agent_rpc.go

@@ -940,6 +940,7 @@ func agentRPCTestInitMasterPanic(ctx context.Context, t *testing.T, client tmcli
 
 var testPopulateReparentJournalCalled = false
 var testTimeCreatedNS int64 = 4569900
+var testWaitPosition string = "test wait position"
 var testActionName = "TestActionName"
 var testMasterAlias = &topodatapb.TabletAlias{
 	Cell: "ce",
@@ -1071,24 +1072,25 @@ func agentRPCTestSlaveWasPromotedPanic(ctx context.Context, t *testing.T, client
 var testSetMasterCalled = false
 var testForceStartSlave = true
 
-func (fra *fakeRPCAgent) SetMaster(ctx context.Context, parent *topodatapb.TabletAlias, timeCreatedNS int64, forceStartSlave bool) error {
+func (fra *fakeRPCAgent) SetMaster(ctx context.Context, parent *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartSlave bool) error {
 	if fra.panics {
 		panic(fmt.Errorf("test-triggered panic"))
 	}
 	compare(fra.t, "SetMaster parent", parent, testMasterAlias)
 	compare(fra.t, "SetMaster timeCreatedNS", timeCreatedNS, testTimeCreatedNS)
+	compare(fra.t, "SetMaster waitPosition", waitPosition, testWaitPosition)
 	compare(fra.t, "SetMaster forceStartSlave", forceStartSlave, testForceStartSlave)
 	testSetMasterCalled = true
 	return nil
 }
 
 func agentRPCTestSetMaster(ctx context.Context, t *testing.T, client tmclient.TabletManagerClient, tablet *topodatapb.Tablet) {
-	err := client.SetMaster(ctx, tablet, testMasterAlias, testTimeCreatedNS, testForceStartSlave)
+	err := client.SetMaster(ctx, tablet, testMasterAlias, testTimeCreatedNS, testWaitPosition, testForceStartSlave)
 	compareError(t, "SetMaster", err, true, testSetMasterCalled)
 }
 
 func agentRPCTestSetMasterPanic(ctx context.Context, t *testing.T, client tmclient.TabletManagerClient, tablet *topodatapb.Tablet) {
-	err := client.SetMaster(ctx, tablet, testMasterAlias, testTimeCreatedNS, testForceStartSlave)
+	err := client.SetMaster(ctx, tablet, testMasterAlias, testTimeCreatedNS, testWaitPosition, testForceStartSlave)
 	expectHandleRPCPanic(t, "SetMaster", true /*verbose*/, err)
 }
 

go/vt/vttablet/faketmclient/fake_client.go

@@ -271,7 +271,7 @@ func (client *FakeTabletManagerClient) SlaveWasPromoted(ctx context.Context, tab
 }
 
 // SetMaster is part of the tmclient.TabletManagerClient interface.
-func (client *FakeTabletManagerClient) SetMaster(ctx context.Context, tablet *topodatapb.Tablet, parent *topodatapb.TabletAlias, timeCreatedNS int64, forceStartSlave bool) error {
+func (client *FakeTabletManagerClient) SetMaster(ctx context.Context, tablet *topodatapb.Tablet, parent *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartSlave bool) error {
 	return nil
 }
 

go/vt/vttablet/grpctmclient/client.go

@@ -702,7 +702,7 @@ func (client *Client) SlaveWasPromoted(ctx context.Context, tablet *topodatapb.T
 }
 
 // SetMaster is part of the tmclient.TabletManagerClient interface.
-func (client *Client) SetMaster(ctx context.Context, tablet *topodatapb.Tablet, parent *topodatapb.TabletAlias, timeCreatedNS int64, forceStartSlave bool) error {
+func (client *Client) SetMaster(ctx context.Context, tablet *topodatapb.Tablet, parent *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartSlave bool) error {
 	cc, c, err := client.dial(tablet)
 	if err != nil {
 		return err
@@ -711,6 +711,7 @@ func (client *Client) SetMaster(ctx context.Context, tablet *topodatapb.Tablet,
 	_, err = c.SetMaster(ctx, &tabletmanagerdatapb.SetMasterRequest{
 		Parent:          parent,
 		TimeCreatedNs:   timeCreatedNS,
+		WaitPosition:    waitPosition,
 		ForceStartSlave: forceStartSlave,
 	})
 	return err

go/vt/vttablet/grpctmserver/server.go

@@ -402,7 +402,7 @@ func (s *server) SetMaster(ctx context.Context, request *tabletmanagerdatapb.Set
 	defer s.agent.HandleRPCPanic(ctx, "SetMaster", request, response, true /*verbose*/, &err)
 	ctx = callinfo.GRPCCallInfo(ctx)
 	response = &tabletmanagerdatapb.SetMasterResponse{}
-	return response, s.agent.SetMaster(ctx, request.Parent, request.TimeCreatedNs, request.ForceStartSlave)
+	return response, s.agent.SetMaster(ctx, request.Parent, request.TimeCreatedNs, request.WaitPosition, request.ForceStartSlave)
 }
 
 func (s *server) SlaveWasRestarted(ctx context.Context, request *tabletmanagerdatapb.SlaveWasRestartedRequest) (response *tabletmanagerdatapb.SlaveWasRestartedResponse, err error) {

go/vt/vttablet/tabletmanager/replication_reporter.go

@@ -153,7 +153,7 @@ func repairReplication(ctx context.Context, agent *ActionAgent) error {
 		}
 	}
 
-	return agent.setMasterRepairReplication(ctx, si.MasterAlias, 0, true)
+	return agent.setMasterRepairReplication(ctx, si.MasterAlias, 0, "", true)
 }
 
 func registerReplicationReporter(agent *ActionAgent) {

go/vt/vttablet/tabletmanager/rpc_agent.go

@@ -116,7 +116,7 @@ type RPCAgent interface {
 
 	SlaveWasPromoted(ctx context.Context) error
 
-	SetMaster(ctx context.Context, parent *topodatapb.TabletAlias, timeCreatedNS int64, forceStartSlave bool) error
+	SetMaster(ctx context.Context, parent *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartSlave bool) error
 
 	SlaveWasRestarted(ctx context.Context, parent *topodatapb.TabletAlias) error
 

go/vt/vttablet/tabletmanager/rpc_replication.go

@@ -522,13 +522,13 @@ func (agent *ActionAgent) SlaveWasPromoted(ctx context.Context) error {
 
 // SetMaster sets replication master, and waits for the
 // reparent_journal table entry up to context timeout
-func (agent *ActionAgent) SetMaster(ctx context.Context, parentAlias *topodatapb.TabletAlias, timeCreatedNS int64, forceStartSlave bool) error {
+func (agent *ActionAgent) SetMaster(ctx context.Context, parentAlias *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartSlave bool) error {
 	if err := agent.lock(ctx); err != nil {
 		return err
 	}
 	defer agent.unlock()
 
-	if err := agent.setMasterLocked(ctx, parentAlias, timeCreatedNS, forceStartSlave); err != nil {
+	if err := agent.setMasterLocked(ctx, parentAlias, timeCreatedNS, waitPosition, forceStartSlave); err != nil {
 		return err
 	}
 
@@ -543,7 +543,7 @@ func (agent *ActionAgent) SetMaster(ctx context.Context, parentAlias *topodatapb
 	return nil
 }
 
-func (agent *ActionAgent) setMasterRepairReplication(ctx context.Context, parentAlias *topodatapb.TabletAlias, timeCreatedNS int64, forceStartSlave bool) (err error) {
+func (agent *ActionAgent) setMasterRepairReplication(ctx context.Context, parentAlias *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartSlave bool) (err error) {
 	parent, err := agent.TopoServer.GetTablet(ctx, parentAlias)
 	if err != nil {
 		return err
@@ -556,10 +556,10 @@ func (agent *ActionAgent) setMasterRepairReplication(ctx context.Context, parent
 
 	defer unlock(&err)
 
-	return agent.setMasterLocked(ctx, parentAlias, timeCreatedNS, forceStartSlave)
+	return agent.setMasterLocked(ctx, parentAlias, timeCreatedNS, waitPosition, forceStartSlave)
 }
 
-func (agent *ActionAgent) setMasterLocked(ctx context.Context, parentAlias *topodatapb.TabletAlias, timeCreatedNS int64, forceStartSlave bool) (err error) {
+func (agent *ActionAgent) setMasterLocked(ctx context.Context, parentAlias *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartSlave bool) (err error) {
 	// End orchestrator maintenance at the end of fixing replication.
 	// This is a best effort operation, so it should happen in a goroutine
 	defer func() {
@@ -648,11 +648,24 @@ func (agent *ActionAgent) setMasterLocked(ctx context.Context, parentAlias *topo
 		}
 	}
 
-	// If needed, wait until we replicate the specified row,
-	// or our context times out.
-	if shouldbeReplicating && timeCreatedNS != 0 {
-		if err := agent.MysqlDaemon.WaitForReparentJournal(ctx, timeCreatedNS); err != nil {
-			return err
+	// If needed, wait until we replicate to the specified point, or our context
+	// times out. Callers can specify the point to wait for as either a
+	// GTID-based replication position or a Vitess reparent journal entry,
+	// or both.
+	if shouldbeReplicating {
+		if waitPosition != "" {
+			pos, err := mysql.DecodePosition(waitPosition)
+			if err != nil {
+				return err
+			}
+			if err := agent.MysqlDaemon.WaitMasterPos(ctx, pos); err != nil {
+				return err
+			}
+		}
+		if timeCreatedNS != 0 {
+			if err := agent.MysqlDaemon.WaitForReparentJournal(ctx, timeCreatedNS); err != nil {
+				return err
+			}
 		}
 	}
 

go/vt/vttablet/tabletmanager/shard_sync.go

@@ -219,7 +219,7 @@ func (agent *ActionAgent) abortMasterTerm(ctx context.Context, masterAlias *topo
 	setMasterCtx, cancelSetMaster := context.WithTimeout(ctx, *topo.RemoteOperationTimeout)
 	defer cancelSetMaster()
 	log.Infof("Attempting to reparent self to new master %v.", masterAliasStr)
-	if err := agent.SetMaster(setMasterCtx, masterAlias, 0, true); err != nil {
+	if err := agent.SetMaster(setMasterCtx, masterAlias, 0, "", true); err != nil {
 		return vterrors.Wrap(err, "failed to reparent self to new master")
 	}
 	return nil

go/vt/vttablet/tmclient/rpc_client_api.go

@@ -184,7 +184,7 @@ type TabletManagerClient interface {
 	// SetMaster tells a tablet to make itself a slave to the
 	// passed in master tablet alias, and wait for the row in the
 	// reparent_journal table (if timeCreatedNS is non-zero).
-	SetMaster(ctx context.Context, tablet *topodatapb.Tablet, parent *topodatapb.TabletAlias, timeCreatedNS int64, forceStartSlave bool) error
+	SetMaster(ctx context.Context, tablet *topodatapb.Tablet, parent *topodatapb.TabletAlias, timeCreatedNS int64, waitPosition string, forceStartSlave bool) error
 
 	// SlaveWasRestarted tells the remote tablet its master has changed
 	SlaveWasRestarted(ctx context.Context, tablet *topodatapb.Tablet, parent *topodatapb.TabletAlias) error

go/vt/wrangler/reparent.go

@@ -130,7 +130,7 @@ func (wr *Wrangler) ReparentTablet(ctx context.Context, tabletAlias *topodatapb.
 	}
 
 	// and do the remote command
-	return wr.tmc.SetMaster(ctx, ti.Tablet, shardInfo.MasterAlias, 0, false)
+	return wr.tmc.SetMaster(ctx, ti.Tablet, shardInfo.MasterAlias, 0, "", false)
 }
 
 // InitShardMaster will make the provided tablet the master for the shard.
@@ -331,7 +331,7 @@ func (wr *Wrangler) initShardMasterLocked(ctx context.Context, ev *events.Repare
 
 // PlannedReparentShard will make the provided tablet the master for the shard,
 // when both the current and new master are reachable and in good shape.
-func (wr *Wrangler) PlannedReparentShard(ctx context.Context, keyspace, shard string, masterElectTabletAlias, avoidMasterAlias *topodatapb.TabletAlias, waitReplicasTimeout, masterElectLagThreshold time.Duration) (err error) {
+func (wr *Wrangler) PlannedReparentShard(ctx context.Context, keyspace, shard string, masterElectTabletAlias, avoidMasterAlias *topodatapb.TabletAlias, waitReplicasTimeout time.Duration) (err error) {
 	// lock the shard
 	lockAction := fmt.Sprintf(
 		"PlannedReparentShard(%v, avoid_master=%v)",
@@ -356,7 +356,7 @@ func (wr *Wrangler) PlannedReparentShard(ctx context.Context, keyspace, shard st
 	}
 
 	// do the work
-	err = wr.plannedReparentShardLocked(ctx, ev, keyspace, shard, masterElectTabletAlias, avoidMasterAlias, waitReplicasTimeout, masterElectLagThreshold)
+	err = wr.plannedReparentShardLocked(ctx, ev, keyspace, shard, masterElectTabletAlias, avoidMasterAlias, waitReplicasTimeout)
 	if err != nil {
 		event.DispatchUpdate(ev, "failed PlannedReparentShard: "+err.Error())
 	} else {
@@ -365,7 +365,7 @@ func (wr *Wrangler) PlannedReparentShard(ctx context.Context, keyspace, shard st
 	return err
 }
 
-func (wr *Wrangler) plannedReparentShardLocked(ctx context.Context, ev *events.Reparent, keyspace, shard string, masterElectTabletAlias, avoidMasterTabletAlias *topodatapb.TabletAlias, waitReplicasTimeout, masterElectLagThreshold time.Duration) error {
+func (wr *Wrangler) plannedReparentShardLocked(ctx context.Context, ev *events.Reparent, keyspace, shard string, masterElectTabletAlias, avoidMasterTabletAlias *topodatapb.TabletAlias, waitReplicasTimeout time.Duration) error {
 	shardInfo, err := wr.ts.GetShard(ctx, keyspace, shard)
 	if err != nil {
 		return err
@@ -544,20 +544,36 @@ func (wr *Wrangler) plannedReparentShardLocked(ctx context.Context, ev *events.R
 		oldMasterTabletInfo := currentMaster
 		ev.OldMaster = *oldMasterTabletInfo.Tablet
 
-		// Before demoting the old master, check replication on the candidate
-		// master to make sure it has a chance of catching up.
-		statusCtx, statusCancel := context.WithTimeout(ctx, *topo.RemoteOperationTimeout)
-		defer statusCancel()
-		status, err := wr.tmc.SlaveStatus(statusCtx, masterElectTabletInfo.Tablet)
+		// Before demoting the old master, first make sure replication is
+		// working from the old master to the candidate master. If it's not
+		// working, we can't do a planned reparent because the candidate won't
+		// catch up.
+		wr.logger.Infof("Checking replication on master-elect %v", masterElectTabletAliasStr)
+
+		// First we find the position of the current master. Note that this is
+		// just a snapshot of the position since we let it keep accepting new
+		// writes until we're sure we're going to proceed.
+		snapshotCtx, snapshotCancel := context.WithTimeout(ctx, *topo.RemoteOperationTimeout)
+		defer snapshotCancel()
+
+		snapshotPos, err := wr.tmc.MasterPosition(snapshotCtx, currentMaster.Tablet)
 		if err != nil {
-			return vterrors.Wrapf(err, "can't get replication status on master-elect %v", masterElectTabletAliasStr)
+			return vterrors.Wrapf(err, "can't get replication position on current master %v; current master must be healthy to perform planned reparent", currentMaster.AliasString())
 		}
-		if !status.SlaveIoRunning || !status.SlaveSqlRunning {
-			return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "replication not running on master-elect %v; replication must be healthy to perform planned reparent", masterElectTabletAliasStr)
-		}
-		// Check if it's behind by a small enough amount.
-		if float64(status.SecondsBehindMaster) > masterElectLagThreshold.Seconds() {
-			return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "replication lag on master-elect %v (%v seconds) is greater than the specified lag threshold (%v); let replication catch up first or try again with a higher threshold", masterElectTabletAliasStr, status.SecondsBehindMaster, masterElectLagThreshold)
+
+		// Now wait for the master-elect to catch up to that snapshot point.
+		// If it catches up to that point within the waitReplicasTimeout,
+		// we can be fairly confident it will catch up on everything that's
+		// happened in the meantime once we demote the master to stop writes.
+		//
+		// We do this as an idempotent SetMaster to make sure the replica knows
+		// who the current master is.
+		setMasterCtx, setMasterCancel := context.WithTimeout(ctx, waitReplicasTimeout)
+		defer setMasterCancel()
+
+		err = wr.tmc.SetMaster(setMasterCtx, masterElectTabletInfo.Tablet, currentMaster.Alias, 0, snapshotPos, true)
+		if err != nil {
+			return vterrors.Wrapf(err, "replication on master-elect %v did not catch up in time; replication must be healthy to perform planned reparent", masterElectTabletAliasStr)
 		}
 
 		// Check we still have the topology lock.
@@ -642,7 +658,7 @@ func (wr *Wrangler) plannedReparentShardLocked(ctx context.Context, ev *events.R
 			// to start replication after being converted to a replica.
 			forceStartReplication := false
 
-			if err := wr.tmc.SetMaster(replCtx, tabletInfo.Tablet, masterElectTabletAlias, reparentJournalTimestamp, forceStartReplication); err != nil {
+			if err := wr.tmc.SetMaster(replCtx, tabletInfo.Tablet, masterElectTabletAlias, reparentJournalTimestamp, "", forceStartReplication); err != nil {
 				rec.RecordError(fmt.Errorf("tablet %v SetMaster failed: %v", alias, err))
 				return
 			}
@@ -976,7 +992,7 @@ func (wr *Wrangler) emergencyReparentShardLocked(ctx context.Context, ev *events
 				if status, ok := statusMap[alias]; ok {
 					forceStartSlave = status.SlaveIoRunning || status.SlaveSqlRunning
 				}
-				if err := wr.tmc.SetMaster(replCtx, tabletInfo.Tablet, masterElectTabletAlias, now, forceStartSlave); err != nil {
+				if err := wr.tmc.SetMaster(replCtx, tabletInfo.Tablet, masterElectTabletAlias, now, "", forceStartSlave); err != nil {
 					rec.RecordError(fmt.Errorf("tablet %v SetMaster failed: %v", alias, err))
 				}
 			}(alias, tabletInfo)