PlannedReparentShard: Fix more known-recoverable problems.

go/vt/schemamanager/schemaswap/schema_swap.go

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

go/vt/topo/shard.go

@@ -24,8 +24,10 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"time"
 
 	"golang.org/x/net/context"
+	"vitess.io/vitess/go/vt/logutil"
 	"vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/vterrors"
 
@@ -174,6 +176,16 @@ func (si *ShardInfo) HasMaster() bool {
 	return !topoproto.TabletAliasIsZero(si.Shard.MasterAlias)
 }
 
+// GetMasterTermStartTime returns the shard's master term start time as a Time value.
+func (si *ShardInfo) GetMasterTermStartTime() time.Time {
+	return logutil.ProtoToTime(si.Shard.MasterTermStartTime)
+}
+
+// SetMasterTermStartTime sets the shard's master term start time as a Time value.
+func (si *ShardInfo) SetMasterTermStartTime(t time.Time) {
+	si.Shard.MasterTermStartTime = logutil.TimeToProto(t)
+}
+
 // GetShard is a high level function to read shard data.
 // It generates trace spans.
 func (ts *Server) GetShard(ctx context.Context, keyspace, shard string) (*ShardInfo, error) {

go/vt/topo/tablet.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"path"
 	"sync"
+	"time"
 
 	"golang.org/x/net/context"
 	"vitess.io/vitess/go/vt/proto/vtrpc"
@@ -30,6 +31,7 @@ import (
 	"vitess.io/vitess/go/netutil"
 	"vitess.io/vitess/go/trace"
 	"vitess.io/vitess/go/vt/log"
+	"vitess.io/vitess/go/vt/logutil"
 
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	"vitess.io/vitess/go/vt/topo/events"
@@ -207,6 +209,16 @@ func (ti *TabletInfo) IsSlaveType() bool {
 	return IsSlaveType(ti.Type)
 }
 
+// GetMasterTermStartTime returns the tablet's master term start time as a Time value.
+func (ti *TabletInfo) GetMasterTermStartTime() time.Time {
+	return logutil.ProtoToTime(ti.Tablet.MasterTermStartTime)
+}
+
+// SetMasterTermStartTime sets the tablet's master term start time as a Time value.
+func (ti *TabletInfo) SetMasterTermStartTime(t time.Time) {
+	ti.Tablet.MasterTermStartTime = logutil.TimeToProto(t)
+}
+
 // NewTabletInfo returns a TabletInfo basing on tablet with the
 // version set. This function should be only used by Server
 // implementations.

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>]",
+		"-keyspace_shard=<keyspace/shard> [-new_master=<tablet alias>] [-avoid_master=<tablet alias>] [-wait_slave_timeout=<duration>] [-lag_threshold=<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,7 +107,8 @@ 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 slaves to catch up in reparenting")
+	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)")
 	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")
@@ -142,7 +143,7 @@ func commandPlannedReparentShard(ctx context.Context, wr *wrangler.Wrangler, sub
 			return err
 		}
 	}
-	return wr.PlannedReparentShard(ctx, keyspace, shard, newMasterAlias, avoidMasterAlias, *waitSlaveTimeout)
+	return wr.PlannedReparentShard(ctx, keyspace, shard, newMasterAlias, avoidMasterAlias, *waitSlaveTimeout, *lagThreshold)
 }
 
 func commandEmergencyReparentShard(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag.FlagSet, args []string) error {

go/vt/vttablet/tabletmanager/init_tablet.go

@@ -120,7 +120,7 @@ func (agent *ActionAgent) InitTablet(port, gRPCPort int32) error {
 				// Read the master term start time from tablet.
 				// If it is nil, it might mean that we are upgrading, so use current time instead
 				if oldTablet.MasterTermStartTime != nil {
-					agent.setMasterTermStartTime(logutil.ProtoToTime(oldTablet.MasterTermStartTime))
+					agent.setMasterTermStartTime(oldTablet.GetMasterTermStartTime())
 				} else {
 					agent.setMasterTermStartTime(time.Now())
 				}
@@ -137,8 +137,8 @@ func (agent *ActionAgent) InitTablet(port, gRPCPort int32) error {
 			if oldTablet.Type == topodatapb.TabletType_MASTER {
 				// Our existing tablet type is master, but the shard record does not agree.
 				// Only take over if our master_term_start_time is after what is in the shard record
-				oldMasterTermStartTime := logutil.ProtoToTime(oldTablet.MasterTermStartTime)
-				currentShardTime := logutil.ProtoToTime(si.MasterTermStartTime)
+				oldMasterTermStartTime := oldTablet.GetMasterTermStartTime()
+				currentShardTime := si.GetMasterTermStartTime()
 				if oldMasterTermStartTime.After(currentShardTime) {
 					tabletType = topodatapb.TabletType_MASTER
 					// read the master term start time from tablet

go/vt/vttablet/tabletmanager/rpc_backup.go

@@ -120,7 +120,7 @@ func (agent *ActionAgent) Backup(ctx context.Context, concurrency int, logger lo
 
 		// Change our type back to the original value.
 		// Original type could be master so pass in a real value for masterTermStartTime
-		_, err = topotools.ChangeType(bgCtx, agent.TopoServer, tablet.Alias, originalType, tablet.MasterTermStartTime)
+		_, err = topotools.ChangeType(bgCtx, agent.TopoServer, tablet.Alias, originalType, tablet.Tablet.MasterTermStartTime)
 		if err != nil {
 			// failure in changing the topology type is probably worse,
 			// so returning that (we logged the snapshot error anyway)

go/vt/vttablet/tabletmanager/shard_sync.go

@@ -154,7 +154,7 @@ func syncShardMaster(ctx context.Context, ts *topo.Server, tablet *topodatapb.Ta
 
 	var shardInfo *topo.ShardInfo
 	_, err = ts.UpdateShardFields(ctx, tablet.Keyspace, tablet.Shard, func(si *topo.ShardInfo) error {
-		lastTerm := logutil.ProtoToTime(si.MasterTermStartTime)
+		lastTerm := si.GetMasterTermStartTime()
 
 		// Save the ShardInfo so we can check it afterward.
 		// We can't use the return value of UpdateShardFields because it might be nil.