cli,kv,server: consolidate AdminClient RPC client creation

This commit consolidates `AdminClient` RPC client creation logic in cli,
kv, and server packages. It is a continuation of the work done in #147606.

Epic: CRDB-48923
Fixes: #148158
Release note: none

Author: cthumuluru-crdb

pkg/cli/debug_recover_loss_of_quorum.go

@@ -308,11 +308,12 @@ func runDebugDeadReplicaCollect(cmd *cobra.Command, args []string) error {
 	var stats loqrecovery.CollectionStats
 
 	if len(debugRecoverCollectInfoOpts.Stores.Specs) == 0 {
-		c, finish, err := getAdminClient(ctx, serverCfg)
+		c, finish, err := dialAdminClient(ctx, serverCfg)
 		if err != nil {
 			return errors.Wrapf(err, "failed to get admin connection to cluster")
 		}
 		defer finish()
+
 		replicaInfo, stats, err = loqrecovery.CollectRemoteReplicaInfo(ctx, c,
 			debugRecoverCollectInfoOpts.maxConcurrency, stderr /* logOutput */)
 		if err != nil {
@@ -426,11 +427,12 @@ func runDebugPlanReplicaRemoval(cmd *cobra.Command, args []string) error {
 	if len(args) == 0 {
 		// If no replica info is provided, try to connect to a cluster default or
 		// explicitly provided to retrieve replica info.
-		c, finish, err := getAdminClient(ctx, serverCfg)
+		c, finish, err := dialAdminClient(ctx, serverCfg)
 		if err != nil {
 			return errors.Wrapf(err, "failed to get admin connection to cluster")
 		}
 		defer finish()
+
 		replicas, stats, err = loqrecovery.CollectRemoteReplicaInfo(ctx, c,
 			debugRecoverPlanOpts.maxConcurrency, stderr /* logOutput */)
 		if err != nil {
@@ -675,7 +677,7 @@ func stageRecoveryOntoCluster(
 	ignoreInternalVersion bool,
 	maxConcurrency int,
 ) error {
-	c, finish, err := getAdminClient(ctx, serverCfg)
+	c, finish, err := dialAdminClient(ctx, serverCfg)
 	if err != nil {
 		return errors.Wrapf(err, "failed to get admin connection to cluster")
 	}
@@ -686,6 +688,7 @@ func stageRecoveryOntoCluster(
 		nodeID roachpb.NodeID
 		planID string
 	}
+
 	vr, err := c.RecoveryVerify(ctx, &serverpb.RecoveryVerifyRequest{
 		MaxConcurrency: int32(maxConcurrency),
 	})
@@ -948,11 +951,12 @@ func runDebugVerify(cmd *cobra.Command, args []string) error {
 		_, _ = fmt.Printf("Checking application of recovery plan %s\n", updatePlan.PlanID)
 	}
 
-	c, finish, err := getAdminClient(ctx, serverCfg)
+	c, finish, err := dialAdminClient(ctx, serverCfg)
 	if err != nil {
 		return errors.Wrapf(err, "failed to get admin connection to cluster")
 	}
 	defer finish()
+
 	req := serverpb.RecoveryVerifyRequest{
 		DecommissionedNodeIDs: updatePlan.DecommissionedNodeIDs,
 		MaxReportedRanges:     20,

pkg/cli/debug_reset_quorum.go

@@ -45,15 +45,15 @@ func runDebugResetQuorum(cmd *cobra.Command, args []string) error {
 	}
 
 	// Set up GRPC Connection for running ResetQuorum.
-	cc, finish, err := getClientGRPCConn(ctx, serverCfg)
+	conn, finish, err := newClientConn(ctx, serverCfg)
 	if err != nil {
 		log.Errorf(ctx, "connection to server failed: %v", err)
 		return err
 	}
 	defer finish()
 
 	// Call ResetQuorum to reset quorum for given range on target node.
-	_, err = kvpb.NewInternalClient(cc).ResetQuorum(ctx, &kvpb.ResetQuorumRequest{
+	_, err = conn.NewInternalClient().ResetQuorum(ctx, &kvpb.ResetQuorumRequest{
 		RangeID: int32(rangeID),
 	})
 	if err != nil {

pkg/cli/debug_send_kv_batch.go

@@ -166,12 +166,11 @@ func runSendKVBatch(cmd *cobra.Command, args []string) error {
 	// Send BatchRequest.
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	conn, finish, err := getClientGRPCConn(ctx, serverCfg)
+	admin, finish, err := dialAdminClient(ctx, serverCfg)
 	if err != nil {
 		return errors.Wrap(err, "failed to connect to the node")
 	}
 	defer finish()
-	admin := serverpb.NewAdminClient(conn)
 
 	br, rec, err := sendKVBatchRequestWithTracingOption(ctx, enableTracing, admin, &ba)
 	if err != nil {

pkg/cli/init.go

@@ -89,7 +89,7 @@ func dialAndCheckHealth(ctx context.Context) error {
 		// (Attempt to) establish the gRPC connection. If that fails,
 		// it may be that the server hasn't started to listen yet, in
 		// which case we'll retry.
-		conn, finish, err := getClientGRPCConn(ctx, serverCfg)
+		conn, finish, err := newClientConn(ctx, serverCfg)
 		if err != nil {
 			return err
 		}
@@ -98,7 +98,7 @@ func dialAndCheckHealth(ctx context.Context) error {
 		// Access the /health endpoint. Until/unless this succeeds, the
 		// node is not yet fully initialized and ready to accept
 		// Bootstrap requests.
-		ac := serverpb.NewAdminClient(conn)
+		ac := conn.NewAdminClient()
 		_, err = ac.Health(ctx, &serverpb.HealthRequest{})
 		return err
 	}

pkg/cli/rpc_clients.go

@@ -73,20 +73,18 @@ func makeRPCClientConfig(cfg server.Config) rpc.ClientConnConfig {
 func newClientConn(ctx context.Context, cfg server.Config) (rpcConn, func(), error) {
 	ccfg := makeRPCClientConfig(cfg)
 	cc, finish, err := rpc.NewClientConn(ctx, ccfg)
-	return &grpcConn{conn: cc}, finish, errors.Wrap(err, "failed to connect to the node")
-}
-
-func getClientGRPCConn(ctx context.Context, cfg server.Config) (*grpc.ClientConn, func(), error) {
-	ccfg := makeRPCClientConfig(cfg)
-	return rpc.NewClientConn(ctx, ccfg)
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "failed to connect to the node")
+	}
+	return &grpcConn{conn: cc}, finish, nil
 }
 
-// getAdminClient returns an AdminClient and a closure that must be invoked
-// to free associated resources.
-func getAdminClient(ctx context.Context, cfg server.Config) (serverpb.AdminClient, func(), error) {
-	conn, finish, err := getClientGRPCConn(ctx, cfg)
+// dialAdminClient dials a client connection and returns an AdminClient and a
+// closure that must be invoked to free associated resources.
+func dialAdminClient(ctx context.Context, cfg server.Config) (serverpb.AdminClient, func(), error) {
+	cc, finish, err := newClientConn(ctx, cfg)
 	if err != nil {
-		return nil, nil, errors.Wrap(err, "failed to connect to the node")
+		return nil, nil, err
 	}
-	return serverpb.NewAdminClient(conn), finish, nil
+	return cc.NewAdminClient(), finish, nil
 }

pkg/kv/kvclient/kvtenant/BUILD.bazel

@@ -26,6 +26,7 @@ go_library(
         "//pkg/roachpb",
         "//pkg/rpc",
         "//pkg/rpc/nodedialer",
+        "//pkg/rpc/rpcbase",
         "//pkg/server/serverpb",
         "//pkg/server/settingswatcher",
         "//pkg/settings",

pkg/kv/kvclient/kvtenant/connector.go

@@ -28,6 +28,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/rpc"
 	"github.com/cockroachdb/cockroach/pkg/rpc/nodedialer"
+	"github.com/cockroachdb/cockroach/pkg/rpc/rpcbase"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
 	"github.com/cockroachdb/cockroach/pkg/server/settingswatcher"
 	"github.com/cockroachdb/cockroach/pkg/settings"
@@ -975,17 +976,19 @@ func (c *connector) dialAddrs(ctx context.Context) (*client, error) {
 		// Try each address on each retry iteration (in random order).
 		for _, i := range rand.Perm(len(c.addrs)) {
 			addr := c.addrs[i]
-			conn, err := c.dialAddr(ctx, addr)
-			if err != nil {
-				log.Warningf(ctx, "error dialing tenant KV address %s: %v", addr, err)
-				continue
+			if !rpcbase.TODODRPC {
+				conn, err := c.dialAddr(ctx, addr)
+				if err != nil {
+					log.Warningf(ctx, "error dialing tenant KV address %s: %v", addr, err)
+					continue
+				}
+				return &client{
+					InternalClient:   kvpb.NewInternalClient(conn),
+					StatusClient:     serverpb.NewStatusClient(conn),
+					AdminClient:      serverpb.NewAdminClient(conn),
+					TimeSeriesClient: tspb.NewTimeSeriesClient(conn),
+				}, nil
 			}
-			return &client{
-				InternalClient:   kvpb.NewInternalClient(conn),
-				StatusClient:     serverpb.NewStatusClient(conn),
-				AdminClient:      serverpb.NewAdminClient(conn),
-				TimeSeriesClient: tspb.NewTimeSeriesClient(conn),
-			}, nil
 		}
 	}
 	return nil, errors.Wrap(ctx.Err(), "dial addrs")

pkg/kv/kvserver/loqrecovery/BUILD.bazel

@@ -53,7 +53,6 @@ go_library(
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_pebble//vfs",
         "@com_github_cockroachdb_redact//:redact",
-        "@org_golang_google_grpc//:grpc",
         "@org_golang_x_sync//errgroup",
     ],
 )

pkg/kv/kvserver/loqrecovery/server.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
 	"sync/atomic"
 	"time"
 
@@ -35,7 +36,6 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
 	"golang.org/x/sync/errgroup"
-	"google.golang.org/grpc"
 )
 
 const rangeMetadataScanChunkSize = 100
@@ -720,6 +720,22 @@ func makeVisitAvailableNodesInParallel(
 			return err
 		}
 
+		// Initialize a nodedialer to establish connections with nodes. We'll
+		// create a custom resolver that uses the already available node list,
+		// which is more efficient than fetching node information again. Node
+		// dialer allows us to reuse utility methods to create RPC connections.
+		resolver := func(nodeID roachpb.NodeID) (net.Addr, roachpb.Locality, error) {
+			for _, node := range nodes {
+				if node.NodeID == nodeID {
+					addr := node.AddressForLocality(loc)
+					return addr, node.Locality, nil
+				}
+			}
+			// This should not happen since the visitor visits the exact same nodes.
+			return nil, roachpb.Locality{}, errors.Newf("node n%d not found in gossip", nodeID)
+		}
+		nd := nodedialer.New(rpcCtx, resolver)
+
 		var g errgroup.Group
 		if maxConcurrency == 0 {
 			// "A value of 0 disables concurrency."
@@ -729,7 +745,7 @@ func makeVisitAvailableNodesInParallel(
 		for _, node := range nodes {
 			node := node // copy for closure
 			g.Go(func() error {
-				return visitNodeWithRetry(ctx, loc, rpcCtx, retryOpts, visitor, node)
+				return visitNodeWithRetry(ctx, nd, retryOpts, visitor, node)
 			})
 		}
 		return g.Wait()
@@ -738,21 +754,19 @@ func makeVisitAvailableNodesInParallel(
 
 func visitNodeWithRetry(
 	ctx context.Context,
-	loc roachpb.Locality,
-	rpcCtx *rpc.Context,
+	nd rpcbase.NodeDialerNoBreaker,
 	retryOpts retry.Options,
 	visitor visitNodeAdminFn,
 	node roachpb.NodeDescriptor,
 ) error {
 	var err error
+	var ac serverpb.AdminClient
 	for r := retry.StartWithCtx(ctx, retryOpts); r.Next(); {
 		log.Infof(ctx, "visiting node n%d, attempt %d", node.NodeID, r.CurrentAttempt())
-		addr := node.AddressForLocality(loc)
-		var conn *grpc.ClientConn
 		// Note that we use ConnectNoBreaker here to avoid any race with probe
 		// running on current node and target node restarting. Errors from circuit
 		// breaker probes could confuse us and present node as unavailable.
-		conn, _, err = rpcCtx.GRPCDialNode(addr.String(), node.NodeID, node.Locality, rpcbase.DefaultClass).ConnectNoBreaker(ctx)
+		ac, err = serverpb.DialAdminClientNoBreaker(nd, ctx, node.NodeID)
 		// Nodes would contain dead nodes that we don't need to visit. We can skip
 		// them and let caller handle incomplete info.
 		if err != nil {
@@ -765,8 +779,7 @@ func visitNodeWithRetry(
 			// live.
 			continue
 		}
-		client := serverpb.NewAdminClient(conn)
-		err = visitor(node.NodeID, client)
+		err = visitor(node.NodeID, ac)
 		if err == nil {
 			return nil
 		}

pkg/server/admin.go

@@ -3723,11 +3723,7 @@ func (s *adminServer) queryTableID(
 func (s *adminServer) dialNode(
 	ctx context.Context, nodeID roachpb.NodeID,
 ) (serverpb.AdminClient, error) {
-	conn, err := s.serverIterator.dialNode(ctx, serverID(nodeID))
-	if err != nil {
-		return nil, err
-	}
-	return serverpb.NewAdminClient(conn), nil
+	return serverpb.DialAdminClient(s.nd, ctx, nodeID)
 }
 
 func (s *adminServer) ListTracingSnapshots(