vtctl: Add command "RefreshStateByShard".

This command allows to run "RefreshState" on all tablets for a given shard.

This is useful in the following cases:

* when tablets must be refreshed after a blacklist rule was removed
* manually refreshing tablets e.g. when filtered replication was enabled or disabled

B=31752251

doc/vtctlReference.md

@@ -1824,6 +1824,7 @@ Blocks until the specified shard has caught up with the filtered replication of
 * [InitTablet](#inittablet)
 * [Ping](#ping)
 * [RefreshState](#refreshstate)
+* [RefreshStateByShard](#refreshstatebyshard)
 * [ReparentTablet](#reparenttablet)
 * [RestoreFromBackup](#restorefrombackup)
 * [RunHealthCheck](#runhealthcheck)
@@ -2104,6 +2105,30 @@ Reloads the tablet record on the specified tablet.
 * The <code>&lt;tablet alias&gt;</code> argument is required for the <code>&lt;RefreshState&gt;</code> command. This error occurs if the command is not called with exactly one argument.
 
 
+### RefreshStateByShard
+
+Runs 'RefreshState' on all tablets in the given shard.
+
+#### Example
+
+<pre class="command-example">RefreshStateByShard [-cells=c1,c2,...] &lt;keyspace/shard&gt;</pre>
+
+#### Flags
+
+| Name | Type | Definition |
+| :-------- | :--------- | :--------- |
+| cells | string | Specifies a comma-separated list of cells whose tablets are included. If empty, all cells are considered. |
+
+
+#### Arguments
+
+* <code>&lt;keyspace/shard&gt;</code> &ndash; Required. The name of a sharded database that contains one or more tables as well as the shard associated with the command. The keyspace must be identified by a string that does not contain whitepace, while the shard is typically identified by a string in the format <code>&lt;range start&gt;-&lt;range end&gt;</code>.
+
+#### Errors
+
+* The <code>&lt;keyspace/shard&gt;</code> argument is required for the <code>&lt;RefreshStateByShard&gt;</code> command. This error occurs if the command is not called with exactly one argument.
+
+
 ### ReparentTablet
 
 Reparent a tablet to the current master in the shard. This only works if the current slave position matches the last known reparent action.

go/vt/vtctl/vtctl.go

@@ -174,6 +174,9 @@ var commands = []commandGroup{
 			{"RefreshState", commandRefreshState,
 				"<tablet alias>",
 				"Reloads the tablet record on the specified tablet."},
+			{"RefreshStateByShard", commandRefreshStateByShard,
+				"[-cells=c1,c2,...] <keyspace/shard>",
+				"Runs 'RefreshState' on all tablets in the given shard."},
 			{"RunHealthCheck", commandRunHealthCheck,
 				"<tablet alias>",
 				"Runs a health check on a remote tablet."},
@@ -871,6 +874,31 @@ func commandRefreshState(ctx context.Context, wr *wrangler.Wrangler, subFlags *f
 	return wr.TabletManagerClient().RefreshState(ctx, tabletInfo.Tablet)
 }
 
+func commandRefreshStateByShard(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag.FlagSet, args []string) error {
+	cellsStr := subFlags.String("cells", "", "Specifies a comma-separated list of cells whose tablets are included. If empty, all cells are considered.")
+	if err := subFlags.Parse(args); err != nil {
+		return err
+	}
+	if subFlags.NArg() != 1 {
+		return fmt.Errorf("The <keyspace/shard> argument is required for the RefreshStateByShard command.")
+	}
+
+	keyspace, shard, err := topoproto.ParseKeyspaceShard(subFlags.Arg(0))
+	if err != nil {
+		return err
+	}
+	si, err := wr.TopoServer().GetShard(ctx, keyspace, shard)
+	if err != nil {
+		return err
+	}
+
+	var cells []string
+	if *cellsStr != "" {
+		cells = strings.Split(*cellsStr, ",")
+	}
+	return wr.RefreshTabletsByShard(ctx, si, nil /* tabletTypes */, cells)
+}
+
 func commandRunHealthCheck(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag.FlagSet, args []string) error {
 	if err := subFlags.Parse(args); err != nil {
 		return err

go/vt/wrangler/keyspace.go

@@ -178,7 +178,7 @@ func (wr *Wrangler) MigrateServedTypes(ctx context.Context, keyspace, shard stri
 		wr.Logger().Infof("WaitForDrain: Sleeping finished. Shutting down queryservice on old tablets now.")
 
 		for _, si := range refreshShards {
-			rec.RecordError(wr.RefreshTabletsByShard(ctx, si, servedType, cells))
+			rec.RecordError(wr.RefreshTabletsByShard(ctx, si, []topodatapb.TabletType{servedType}, cells))
 		}
 		return rec.Error()
 	}
@@ -383,7 +383,7 @@ func (wr *Wrangler) migrateServedTypesLocked(ctx context.Context, keyspace strin
 	if needToRefreshSourceTablets {
 		event.DispatchUpdate(ev, "refreshing source shard tablets so they restart their query service")
 		for _, si := range sourceShards {
-			wr.RefreshTabletsByShard(ctx, si, servedType, cells)
+			wr.RefreshTabletsByShard(ctx, si, []topodatapb.TabletType{servedType}, cells)
 		}
 	}
 
@@ -433,7 +433,7 @@ func (wr *Wrangler) migrateServedTypesLocked(ctx context.Context, keyspace strin
 	if needToRefreshDestinationTablets {
 		event.DispatchUpdate(ev, "refreshing destination shard tablets so they restart their query service")
 		for _, si := range destinationShards {
-			wr.RefreshTabletsByShard(ctx, si, servedType, cells)
+			wr.RefreshTabletsByShard(ctx, si, []topodatapb.TabletType{servedType}, cells)
 		}
 	}
 
@@ -690,7 +690,7 @@ func (wr *Wrangler) replicaMigrateServedFrom(ctx context.Context, ki *topo.Keysp
 	// Now refresh the source servers so they reload their
 	// blacklisted table list
 	event.DispatchUpdate(ev, "refreshing sources tablets state so they update their blacklisted tables")
-	if err := wr.RefreshTabletsByShard(ctx, sourceShard, servedType, cells); err != nil {
+	if err := wr.RefreshTabletsByShard(ctx, sourceShard, []topodatapb.TabletType{servedType}, cells); err != nil {
 		return err
 	}
 
@@ -801,7 +801,7 @@ func (wr *Wrangler) SetKeyspaceServedFrom(ctx context.Context, keyspace string,
 // RefreshTabletsByShard calls RefreshState on all the tables of a
 // given type in a shard. It would work for the master, but the
 // discovery wouldn't be very efficient.
-func (wr *Wrangler) RefreshTabletsByShard(ctx context.Context, si *topo.ShardInfo, tabletType topodatapb.TabletType, cells []string) error {
+func (wr *Wrangler) RefreshTabletsByShard(ctx context.Context, si *topo.ShardInfo, tabletTypes []topodatapb.TabletType, cells []string) error {
 	wr.Logger().Infof("RefreshTabletsByShard called on shard %v/%v", si.Keyspace(), si.ShardName())
 	tabletMap, err := wr.ts.GetTabletMapForShardByCell(ctx, si.Keyspace(), si.ShardName(), cells)
 	switch err {
@@ -816,7 +816,7 @@ func (wr *Wrangler) RefreshTabletsByShard(ctx context.Context, si *topo.ShardInf
 	// ignore errors in this phase
 	wg := sync.WaitGroup{}
 	for _, ti := range tabletMap {
-		if ti.Type != tabletType {
+		if tabletTypes != nil && !topoproto.IsTypeInList(ti.Type, tabletTypes) {
 			continue
 		}
 

test/tabletmanager.py

@@ -118,9 +118,12 @@ def _test_sanity(self):
         len(query_result['fields']), 2,
         'expected 2 fields in vt_select_test: %s' % str(query_result))
 
-    # check Ping / RefreshState
+    # check Ping / RefreshState / RefreshStateByShard
     utils.run_vtctl(['Ping', tablet_62344.tablet_alias])
     utils.run_vtctl(['RefreshState', tablet_62344.tablet_alias])
+    utils.run_vtctl(['RefreshStateByShard', 'test_keyspace/0'])
+    utils.run_vtctl(['RefreshStateByShard', '--cells=test_nj',
+                     'test_keyspace/0'])
 
     # Quickly check basic actions.
     utils.run_vtctl(['SetReadOnly', tablet_62344.tablet_alias])