[vtctld] Migrate ShardReplicationPositions

go/cmd/vtctldclient/cli/shards.go

@@ -17,8 +17,13 @@ limitations under the License.
 package cli
 
 import (
+	"sort"
+
+	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/vt/topo/topoproto"
 
+	replicationdatapb "vitess.io/vitess/go/vt/proto/replicationdata"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
 )
 
@@ -41,3 +46,78 @@ func ParseKeyspaceShards(args []string) ([]*vtctldatapb.Shard, error) {
 
 	return shards, nil
 }
+
+// ReplicatingTablet is a struct to group a Tablet together with its replication
+// Status.
+type ReplicatingTablet struct {
+	Status *replicationdatapb.Status
+	Tablet *topodatapb.Tablet
+}
+
+type rTablets []*ReplicatingTablet
+
+func (rts rTablets) Len() int      { return len(rts) }
+func (rts rTablets) Swap(i, j int) { rts[i], rts[j] = rts[j], rts[i] }
+func (rts rTablets) Less(i, j int) bool {
+	l, r := rts[i], rts[j]
+
+	// l or r ReplicationStatus would be nil if we failed to get
+	// the position (put them at the beginning of the list)
+	if l.Status == nil {
+		return r.Status != nil
+	}
+
+	if r.Status == nil {
+		return false
+	}
+
+	// the type proto has MASTER first, so sort by that. Will show
+	// the MASTER first, then each replica type sorted by
+	// replication position.
+	if l.Tablet.Type < r.Tablet.Type {
+		return true
+	}
+
+	if l.Tablet.Type > r.Tablet.Type {
+		return false
+	}
+
+	// then compare replication positions
+	lpos, err := mysql.DecodePosition(l.Status.Position)
+	if err != nil {
+		return true
+	}
+
+	rpos, err := mysql.DecodePosition(r.Status.Position)
+	if err != nil {
+		return false
+	}
+
+	return !lpos.AtLeast(rpos)
+}
+
+// SortReplicatingTablets returns a sorted list of replicating tablets (which is
+// a struct grouping a Tablet together with its replication Status).
+//
+// The sorting order is:
+// 1. Tablets that do not have a replication Status.
+// 2. Any tablets of type MASTER.
+// 3. Remaining tablets sorted by comparing replication positions.
+func SortReplicatingTablets(tabletMap map[string]*topodatapb.Tablet, replicationStatuses map[string]*replicationdatapb.Status) []*ReplicatingTablet {
+	rtablets := make([]*ReplicatingTablet, 0, len(tabletMap))
+
+	for alias, tablet := range tabletMap {
+		if status, ok := replicationStatuses[alias]; ok {
+			rtablets = append(rtablets, &ReplicatingTablet{
+				Status: status,
+				Tablet: tablet,
+			})
+		} else {
+			rtablets = append(rtablets, &ReplicatingTablet{Tablet: tablet})
+		}
+	}
+
+	sort.Sort(rTablets(rtablets))
+
+	return rtablets
+}

go/cmd/vtctldclient/internal/command/shards.go

@@ -52,6 +52,16 @@ var (
 		Args: cobra.ExactArgs(2),
 		RunE: commandRemoveShardCell,
 	}
+	// ShardReplicationPositions makes a ShardReplicationPositions gRPC request
+	// to a vtctld.
+	ShardReplicationPositions = &cobra.Command{
+		Use: "ShardReplicationPositions <keyspace/shard>",
+		Long: `Shows the replication status of each tablet in the shard graph.
+Output is sorted by tablet type, then replication position.
+Use ctrl-C to interrupt the command and see partial results if needed.`,
+		Args: cobra.ExactArgs(1),
+		RunE: commandShardReplicationPositions,
+	}
 )
 
 var createShardOptions = struct {
@@ -173,6 +183,38 @@ func commandRemoveShardCell(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
+func commandShardReplicationPositions(cmd *cobra.Command, args []string) error {
+	keyspace, shard, err := topoproto.ParseKeyspaceShard(cmd.Flags().Arg(0))
+	if err != nil {
+		return err
+	}
+
+	cli.FinishedParsing(cmd)
+
+	resp, err := client.ShardReplicationPositions(commandCtx, &vtctldatapb.ShardReplicationPositionsRequest{
+		Keyspace: keyspace,
+		Shard:    shard,
+	})
+	if err != nil {
+		return err
+	}
+
+	for _, rt := range cli.SortReplicatingTablets(resp.TabletMap, resp.ReplicationStatuses) {
+		var line string
+
+		switch rt.Status {
+		case nil:
+			line = cli.MarshalTabletAWK(rt.Tablet) + "<err> <err> <err>"
+		default:
+			line = cli.MarshalTabletAWK(rt.Tablet) + fmt.Sprintf(" %v %v", rt.Status.Position, rt.Status.SecondsBehindMaster)
+		}
+
+		fmt.Println(line)
+	}
+
+	return nil
+}
+
 func init() {
 	CreateShard.Flags().BoolVarP(&createShardOptions.Force, "force", "f", false, "")
 	CreateShard.Flags().BoolVarP(&createShardOptions.IncludeParent, "include-parent", "p", false, "")
@@ -187,4 +229,6 @@ func init() {
 	RemoveShardCell.Flags().BoolVarP(&removeShardCellOptions.Force, "force", "f", false, "Proceed even if the cell's topology server cannot be reached. The assumption is that you turned down the entire cell, and just need to update the global topo data.")
 	RemoveShardCell.Flags().BoolVarP(&removeShardCellOptions.Recursive, "recursive", "r", false, "Also delete all tablets in that cell beloning to the specified shard.")
 	Root.AddCommand(RemoveShardCell)
+
+	Root.AddCommand(ShardReplicationPositions)
 }