[vtadmin-api] Reorganize tablet-related functions into vtadmin/cluster/cluster.go

go/vt/vtadmin/api.go

@@ -31,6 +31,7 @@ import (
 	"vitess.io/vitess/go/vt/concurrency"
 	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/vtadmin/cluster"
+	"vitess.io/vitess/go/vt/vtadmin/errors"
 	"vitess.io/vitess/go/vt/vtadmin/grpcserver"
 	vtadminhttp "vitess.io/vitess/go/vt/vtadmin/http"
 	vthandlers "vitess.io/vitess/go/vt/vtadmin/http/handlers"
@@ -302,7 +303,7 @@ func (api *API) GetSchemas(ctx context.Context, req *vtadminpb.GetSchemasRequest
 
 			// Since tablets are per-cluster, we can fetch them once
 			// and use them throughout the other waitgroups.
-			tablets, err := api.getTablets(ctx, c)
+			tablets, err := c.GetTablets(ctx)
 			if err != nil {
 				er.RecordError(err)
 				return
@@ -449,7 +450,7 @@ func (api *API) GetTablet(ctx context.Context, req *vtadminpb.GetTabletRequest)
 		go func(c *cluster.Cluster) {
 			defer wg.Done()
 
-			ts, err := api.getTablets(ctx, c)
+			ts, err := c.GetTablets(ctx)
 			if err != nil {
 				er.RecordError(err)
 				return
@@ -477,12 +478,12 @@ func (api *API) GetTablet(ctx context.Context, req *vtadminpb.GetTabletRequest)
 
 	switch len(tablets) {
 	case 0:
-		return nil, vterrors.Errorf(vtrpcpb.Code_NOT_FOUND, "%s: %s, searched clusters = %v", ErrNoTablet, req.Hostname, ids)
+		return nil, vterrors.Errorf(vtrpcpb.Code_NOT_FOUND, "%s: %s, searched clusters = %v", errors.ErrNoTablet, req.Hostname, ids)
 	case 1:
 		return tablets[0], nil
 	}
 
-	return nil, vterrors.Errorf(vtrpcpb.Code_NOT_FOUND, "%s: %s, searched clusters = %v", ErrAmbiguousTablet, req.Hostname, ids)
+	return nil, vterrors.Errorf(vtrpcpb.Code_NOT_FOUND, "%s: %s, searched clusters = %v", errors.ErrAmbiguousTablet, req.Hostname, ids)
 }
 
 // GetTablets is part of the vtadminpb.VTAdminServer interface.
@@ -505,7 +506,7 @@ func (api *API) GetTablets(ctx context.Context, req *vtadminpb.GetTabletsRequest
 		go func(c *cluster.Cluster) {
 			defer wg.Done()
 
-			ts, err := api.getTablets(ctx, c)
+			ts, err := c.GetTablets(ctx)
 			if err != nil {
 				er.RecordError(err)
 				return
@@ -528,60 +529,6 @@ func (api *API) GetTablets(ctx context.Context, req *vtadminpb.GetTabletsRequest
 	}, nil
 }
 
-func (api *API) getTablets(ctx context.Context, c *cluster.Cluster) ([]*vtadminpb.Tablet, error) {
-	if err := c.DB.Dial(ctx, ""); err != nil {
-		return nil, err
-	}
-
-	rows, err := c.DB.ShowTablets(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	return ParseTablets(rows, c)
-}
-
-// findTablet returns the first tablet in a given cluster that satisfies the filter function.
-func (api *API) findTablet(ctx context.Context, cluster *cluster.Cluster, filter func(*vtadminpb.Tablet) bool) (*vtadminpb.Tablet, error) {
-	tablets, err := api.findTablets(ctx, cluster, filter, 1)
-	if err != nil {
-		return nil, err
-	}
-
-	if len(tablets) != 1 {
-		return nil, ErrNoTablet
-	}
-
-	return tablets[0], nil
-}
-
-// findTablets returns the first N tablets in the given cluster that satisfy
-// the filter function. If N = -1, then all matching tablets are returned.
-// Ordering is not guaranteed, and callers should write their filter functions accordingly.
-func (api *API) findTablets(ctx context.Context, cluster *cluster.Cluster, filter func(*vtadminpb.Tablet) bool, n int) ([]*vtadminpb.Tablet, error) {
-	tablets, err := api.getTablets(ctx, cluster)
-	if err != nil {
-		return nil, err
-	}
-
-	if n == -1 {
-		n = len(tablets)
-	}
-
-	results := make([]*vtadminpb.Tablet, 0, n)
-	for _, t := range tablets {
-		if len(results) >= n {
-			break
-		}
-
-		if filter(t) {
-			results = append(results, t)
-		}
-	}
-
-	return results, nil
-}
-
 func (api *API) getClustersForRequest(ids []string) ([]*cluster.Cluster, []string) {
 	if len(ids) == 0 {
 		clusterIDs := make([]string, 0, len(api.clusters))
@@ -610,23 +557,23 @@ func (api *API) VTExplain(ctx context.Context, req *vtadminpb.VTExplainRequest)
 	defer span.Finish()
 
 	if req.Cluster == "" {
-		return nil, fmt.Errorf("%w: cluster ID is required", ErrInvalidRequest)
+		return nil, fmt.Errorf("%w: cluster ID is required", errors.ErrInvalidRequest)
 	}
 
 	if req.Keyspace == "" {
-		return nil, fmt.Errorf("%w: keyspace name is required", ErrInvalidRequest)
+		return nil, fmt.Errorf("%w: keyspace name is required", errors.ErrInvalidRequest)
 	}
 
 	if req.Sql == "" {
-		return nil, fmt.Errorf("%w: SQL query is required", ErrInvalidRequest)
+		return nil, fmt.Errorf("%w: SQL query is required", errors.ErrInvalidRequest)
 	}
 
 	c, ok := api.clusterMap[req.Cluster]
 	if !ok {
-		return nil, ErrUnsupportedCluster
+		return nil, errors.ErrUnsupportedCluster
 	}
 
-	tablet, err := api.findTablet(ctx, c, func(t *vtadminpb.Tablet) bool {
+	tablet, err := c.FindTablet(ctx, func(t *vtadminpb.Tablet) bool {
 		return t.Tablet.Keyspace == req.Keyspace && topo.IsInServingGraph(t.Tablet.Type) && t.Tablet.Type != topodatapb.TabletType_MASTER && t.State == vtadminpb.Tablet_SERVING
 	})
 
@@ -690,7 +637,7 @@ func (api *API) VTExplain(ctx context.Context, req *vtadminpb.VTExplainRequest)
 
 		ksvs, ok := res.SrvVSchema.Keyspaces[req.Keyspace]
 		if !ok {
-			er.RecordError(fmt.Errorf("%w: keyspace %s", ErrNoSrvVSchema, req.Keyspace))
+			er.RecordError(fmt.Errorf("%w: keyspace %s", errors.ErrNoSrvVSchema, req.Keyspace))
 			return
 		}