Emergency Reparent: Take Union of Executed and Retrieved for Consideration

go/mysql/flavor_mysql.go

@@ -121,6 +121,15 @@ func (mysqlFlavor) status(c *Conn) (SlaveStatus, error) {
 	if err != nil {
 		return SlaveStatus{}, vterrors.Wrapf(err, "SlaveStatus can't parse MySQL 5.6 GTID (Executed_Gtid_Set: %#v)", resultMap["Executed_Gtid_Set"])
 	}
+	relayLogGTIDSet, err := parseMysql56GTIDSet(resultMap["Retrieved_Gtid_Set"])
+	if err != nil {
+		return SlaveStatus{}, vterrors.Wrapf(err, "SlaveStatus can't parse MySQL 5.6 GTID (Retrieved_Gtid_Set: %#v)", resultMap["Retrieved_Gtid_Set"])
+	}
+	// We take the union of the executed and retrieved gtidset, because the retrieved gtidset only represents GTIDs since
+	// the relay log has been reset. To get the full Position, we need to take a union of executed GTIDSets, since these would
+	// have been in the relay log's GTIDSet in the past, prior to a reset.
+	status.RelayLogPosition.GTIDSet = status.Position.GTIDSet.Union(relayLogGTIDSet)
+
 	return status, nil
 }
 

go/mysql/slave_status.go

@@ -23,7 +23,12 @@ import (
 
 // SlaveStatus holds replication information from SHOW SLAVE STATUS.
 type SlaveStatus struct {
-	Position            Position
+	Position Position
+	// RelayLogPosition is the Position that the replica would be at if it
+	// were to finish executing everything that's currently in its relay log.
+	// However, some MySQL flavors don't expose this information,
+	// in which case RelayLogPosition.IsZero() will be true.
+	RelayLogPosition    Position
 	SlaveIORunning      bool
 	SlaveSQLRunning     bool
 	SecondsBehindMaster uint
@@ -42,6 +47,7 @@ func (s *SlaveStatus) SlaveRunning() bool {
 func SlaveStatusToProto(s SlaveStatus) *replicationdatapb.Status {
 	return &replicationdatapb.Status{
 		Position:            EncodePosition(s.Position),
+		RelayLogPosition:    EncodePosition(s.RelayLogPosition),
 		SlaveIoRunning:      s.SlaveIORunning,
 		SlaveSqlRunning:     s.SlaveSQLRunning,
 		SecondsBehindMaster: uint32(s.SecondsBehindMaster),
@@ -57,8 +63,13 @@ func ProtoToSlaveStatus(s *replicationdatapb.Status) SlaveStatus {
 	if err != nil {
 		panic(vterrors.Wrapf(err, "cannot decode Position"))
 	}
+	relayPos, err := DecodePosition(s.RelayLogPosition)
+	if err != nil {
+		panic(vterrors.Wrapf(err, "cannot decode RelayLogPosition"))
+	}
 	return SlaveStatus{
 		Position:            pos,
+		RelayLogPosition:    relayPos,
 		SlaveIORunning:      s.SlaveIoRunning,
 		SlaveSQLRunning:     s.SlaveSqlRunning,
 		SecondsBehindMaster: uint(s.SecondsBehindMaster),

go/vt/wrangler/reparent.go

@@ -23,6 +23,7 @@ This file handles the reparenting operations.
 import (
 	"context"
 	"fmt"
+	"sort"
 	"sync"
 	"time"
 
@@ -757,47 +758,75 @@ func (wr *Wrangler) findCurrentMaster(tabletMap map[string]*topo.TabletInfo) *to
 	return currentMaster
 }
 
-// maxReplPosSearch is a struct helping to search for a tablet with the largest replication
-// position querying status from all tablets in parallel.
-type maxReplPosSearch struct {
+// tabletReplInfoAggregator is a struct that can be used to get all tablets replication positions tied to their aliases concurrently.
+// It allows for a sorting of the retrieved replication infos, based on position, such that the earliest members
+// of the sorted list have the largest replication position.
+type tabletReplInfoAggregator struct {
 	wrangler            *Wrangler
 	ctx                 context.Context
 	waitReplicasTimeout time.Duration
 	waitGroup           sync.WaitGroup
-	maxPosLock          sync.Mutex
-	maxPos              mysql.Position
-	maxPosTablet        *topodatapb.Tablet
+	statusLock          sync.Mutex
+	tabletReplInfos     []*replicationInfo
 }
 
-func (maxPosSearch *maxReplPosSearch) processTablet(tablet *topodatapb.Tablet) {
-	defer maxPosSearch.waitGroup.Done()
-	maxPosSearch.wrangler.logger.Infof("getting replication position from %v", topoproto.TabletAliasString(tablet.Alias))
+type replicationInfo struct {
+	alias    *topodatapb.TabletAlias
+	position mysql.Position
+}
+
+func (t *tabletReplInfoAggregator) processTablet(tablet *topodatapb.Tablet) {
+	defer t.waitGroup.Done()
+	t.wrangler.logger.Infof("getting replication position from %v", topoproto.TabletAliasString(tablet.Alias))
 
-	slaveStatusCtx, cancelSlaveStatus := context.WithTimeout(maxPosSearch.ctx, maxPosSearch.waitReplicasTimeout)
+	slaveStatusCtx, cancelSlaveStatus := context.WithTimeout(t.ctx, t.waitReplicasTimeout)
 	defer cancelSlaveStatus()
 
-	status, err := maxPosSearch.wrangler.tmc.SlaveStatus(slaveStatusCtx, tablet)
+	status, err := t.wrangler.tmc.SlaveStatus(slaveStatusCtx, tablet)
 	if err != nil {
-		maxPosSearch.wrangler.logger.Warningf("failed to get replication status from %v, ignoring tablet: %v", topoproto.TabletAliasString(tablet.Alias), err)
+		t.wrangler.logger.Warningf("failed to get replication status from %v, ignoring tablet: %v", topoproto.TabletAliasString(tablet.Alias), err)
 		return
 	}
-	replPos, err := mysql.DecodePosition(status.Position)
+	replPos, err := decodePosition(status)
 	if err != nil {
-		maxPosSearch.wrangler.logger.Warningf("cannot decode slave %v position %v: %v", topoproto.TabletAliasString(tablet.Alias), status.Position, err)
+		t.wrangler.logger.Warningf("cannot decode slave %v position %v: %v", topoproto.TabletAliasString(tablet.Alias), status.Position, err)
 		return
 	}
 
-	maxPosSearch.maxPosLock.Lock()
-	if maxPosSearch.maxPosTablet == nil || !maxPosSearch.maxPos.AtLeast(replPos) {
-		maxPosSearch.maxPos = replPos
-		maxPosSearch.maxPosTablet = tablet
+	t.statusLock.Lock()
+	t.tabletReplInfos = append(t.tabletReplInfos, &replicationInfo{
+		alias:    tablet.Alias,
+		position: replPos,
+	})
+	t.statusLock.Unlock()
+}
+
+// unpackTabletAliases will turn a []*replicationInfo from the tabletReplInfos field into an []*TabletAlias.
+func (t *tabletReplInfoAggregator) unpackTabletAliases() []*topodatapb.TabletAlias {
+	t.statusLock.Lock()
+	tabletAliases := make([]*topodatapb.TabletAlias, 0, len(t.tabletReplInfos))
+
+	for i := range t.tabletReplInfos {
+		tabletAliases = append(tabletAliases, t.tabletReplInfos[i].alias)
 	}
-	maxPosSearch.maxPosLock.Unlock()
+	t.statusLock.Unlock()
+
+	return tabletAliases
+}
+
+func (t *tabletReplInfoAggregator) Len() int {
+	return len(t.tabletReplInfos)
+}
+func (t *tabletReplInfoAggregator) Swap(i, j int) {
+	t.tabletReplInfos[i], t.tabletReplInfos[j] = t.tabletReplInfos[j], t.tabletReplInfos[i]
+}
+func (t *tabletReplInfoAggregator) Less(i, j int) bool {
+	return !t.tabletReplInfos[i].position.AtLeast(t.tabletReplInfos[j].position)
 }
 
 // chooseNewMaster finds a tablet that is going to become master after reparent. The criteria
-// for the new master-elect are (preferably) to be in the same cell as the current master, and
-// to be different from avoidMasterTabletAlias. The tablet with the largest replication
+// for the new master-elect are to be in the same cell as the current master, and
+// to be different from avoidMasterTabletAlias, if supplied. The tablet with the largest replication
 // position is chosen to minimize the time of catching up with the master. Note that the search
 // for largest replication position will race with transactions being executed on the master at
 // the same time, so when all tablets are roughly at the same position then the choice of the
@@ -808,37 +837,71 @@ func (wr *Wrangler) chooseNewMaster(
 	tabletMap map[string]*topo.TabletInfo,
 	avoidMasterTabletAlias *topodatapb.TabletAlias,
 	waitReplicasTimeout time.Duration) (*topodatapb.TabletAlias, error) {
-
-	if avoidMasterTabletAlias == nil {
-		return nil, fmt.Errorf("tablet to avoid for reparent is not provided, cannot choose new master")
+	candidates, err := wr.FindMasterCandidates(ctx, tabletMap, waitReplicasTimeout)
+	if err != nil {
+		return nil, err
 	}
+
 	var masterCell string
 	if shardInfo.MasterAlias != nil {
 		masterCell = shardInfo.MasterAlias.Cell
 	}
 
-	maxPosSearch := maxReplPosSearch{
+	// Filter out avoidMasterTabletAlias, if it exists in the candidates list, and also filter out
+	// any candidates that aren't in the master tablets cell.
+	filteredCandidates := make([]*topodatapb.TabletAlias, 0, len(candidates))
+	for i := range candidates {
+		if candidates[i] != avoidMasterTabletAlias && candidates[i].Cell == masterCell {
+			filteredCandidates = append(filteredCandidates, candidates[i])
+		}
+	}
+
+	if len(filteredCandidates) == 0 {
+		return nil, nil
+	}
+	return filteredCandidates[0], nil
+}
+
+// FindMasterCandidates will look at all of the tablets in the supplied tabletMap, and return a list of tabletAliases that
+// are all caught up. This means that all of the returned TabletAliases will have the same replication position.
+func (wr *Wrangler) FindMasterCandidates(
+	ctx context.Context,
+	tabletMap map[string]*topo.TabletInfo,
+	waitReplicasTimeout time.Duration) ([]*topodatapb.TabletAlias, error) {
+	aggregator := &tabletReplInfoAggregator{
 		wrangler:            wr,
 		ctx:                 ctx,
 		waitReplicasTimeout: waitReplicasTimeout,
 		waitGroup:           sync.WaitGroup{},
-		maxPosLock:          sync.Mutex{},
+		statusLock:          sync.Mutex{},
+		tabletReplInfos:     make([]*replicationInfo, 0, len(tabletMap)),
 	}
 	for _, tabletInfo := range tabletMap {
-		if (masterCell != "" && tabletInfo.Alias.Cell != masterCell) ||
-			topoproto.TabletAliasEqual(tabletInfo.Alias, avoidMasterTabletAlias) ||
-			tabletInfo.Tablet.Type != topodatapb.TabletType_REPLICA {
+		if tabletInfo.Tablet.Type != topodatapb.TabletType_REPLICA {
 			continue
 		}
-		maxPosSearch.waitGroup.Add(1)
-		go maxPosSearch.processTablet(tabletInfo.Tablet)
+		aggregator.waitGroup.Add(1)
+		go aggregator.processTablet(tabletInfo.Tablet)
 	}
-	maxPosSearch.waitGroup.Wait()
+	aggregator.waitGroup.Wait()
 
-	if maxPosSearch.maxPosTablet == nil {
-		return nil, nil
+	sort.Sort(aggregator)
+
+	if len(aggregator.tabletReplInfos) < 2 {
+		// Either 1 result, or 0. Regardless, we can return the list at this point as we don't need to
+		// filter out tablets that aren't fully caught up.
+		return aggregator.unpackTabletAliases(), nil
 	}
-	return maxPosSearch.maxPosTablet.Alias, nil
+
+	for i := 1; i < len(aggregator.tabletReplInfos); i++ {
+		if aggregator.Less(i, i-1) {
+			// We found the first one that isn't fully caught up. Remove this and all further list items.
+			aggregator.tabletReplInfos = aggregator.tabletReplInfos[0:i]
+			break
+		}
+	}
+
+	return aggregator.unpackTabletAliases(), nil
 }
 
 // EmergencyReparentShard will make the provided tablet the master for
@@ -947,26 +1010,9 @@ func (wr *Wrangler) emergencyReparentShardLocked(ctx context.Context, ev *events
 		return fmt.Errorf("lost topology lock, aborting: %v", err)
 	}
 
-	// Verify masterElect is alive and has the most advanced position
-	masterElectStatus, ok := statusMap[masterElectTabletAliasStr]
-	if !ok {
-		return fmt.Errorf("couldn't get master elect %v replication position", topoproto.TabletAliasString(masterElectTabletAlias))
-	}
-	masterElectPos, err := mysql.DecodePosition(masterElectStatus.Position)
-	if err != nil {
-		return fmt.Errorf("cannot decode master elect position %v: %v", masterElectStatus.Position, err)
-	}
-	for alias, status := range statusMap {
-		if alias == masterElectTabletAliasStr {
-			continue
-		}
-		pos, err := mysql.DecodePosition(status.Position)
-		if err != nil {
-			return fmt.Errorf("cannot decode replica %v position %v: %v", alias, status.Position, err)
-		}
-		if !masterElectPos.AtLeast(pos) {
-			return fmt.Errorf("tablet %v is more advanced than master elect tablet %v: %v > %v", alias, masterElectTabletAliasStr, status.Position, masterElectStatus.Position)
-		}
+	// Bail out if the master tablet candidate is not the farthest ahead.
+	if err := wr.CheckTabletIsFarthestAhead(masterElectTabletAliasStr, statusMap); err != nil {
+		return err
 	}
 
 	// Promote the masterElect
@@ -1043,6 +1089,52 @@ func (wr *Wrangler) emergencyReparentShardLocked(ctx context.Context, ev *events
 	return nil
 }
 
+// CheckTabletIsFarthestAhead will take a tablet alias string, along with a statusMap of tablet alias strings to tablet Status objects,
+// and return an error if the tablet is not the farthest ahead, or otherwise return nil if it is the farthest ahead.
+func (wr *Wrangler) CheckTabletIsFarthestAhead(tabletAliasStr string, statusMap map[string]*replicationdatapb.Status) error {
+	tabletStatus, ok := statusMap[tabletAliasStr]
+	if !ok {
+		return fmt.Errorf("couldn't get tablet %v replication position", tabletAliasStr)
+	}
+	candidatePos, err := decodePosition(tabletStatus)
+	if err != nil {
+		return fmt.Errorf("cannot decode tablet position %v: %v", tabletStatus.Position, err)
+	}
+	for alias, status := range statusMap {
+		if alias == tabletAliasStr {
+			continue
+		}
+		pos, err := decodePosition(status)
+		if err != nil {
+			return fmt.Errorf("cannot decode and merge replica %v executed and retrieved positions %v: %v", alias, status.Position, err)
+		}
+		if !candidatePos.AtLeast(pos) {
+			return fmt.Errorf("tablet %v is more advanced than master elect tablet %v: %v > %v", alias, tabletAliasStr, status.Position, tabletStatus.Position)
+		}
+	}
+
+	return nil
+}
+
+// decodePosition is a helper that will decode the RelayLogPosition, if it's non-empty, and otherwise fall back
+// to the executed position.
+func decodePosition(status *replicationdatapb.Status) (mysql.Position, error) {
+	relayPos, err := mysql.DecodePosition(status.RelayLogPosition)
+	if err != nil {
+		log.Infof("Decode relay log position failed, err: %v", err)
+		return mysql.Position{}, err
+	}
+	if !relayPos.IsZero() {
+		return relayPos, nil
+	}
+	executedPos, err := mysql.DecodePosition(status.Position)
+	if err != nil {
+		log.Infof("Decode position failed, err: %v", err)
+		return mysql.Position{}, err
+	}
+	return executedPos, nil
+}
+
 // TabletExternallyReparented changes the type of new master for this shard to MASTER
 // and updates it's tablet record in the topo. Updating the shard record is handled
 // by the new master tablet