Merge #153032

153032: kv: use generic dialer to create RPC client r=cthumuluru-crdb a=cthumuluru-crdb

Changes in this PR are a followup to #152948.

Epic: CRDB-48935
Informs: None
Release note: None

Co-authored-by: Chandra Thumuluru <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/closedts/ctpb/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
         "//pkg/kv/kvpb",
         "//pkg/roachpb",
         "//pkg/rpc/rpcbase",
+        "//pkg/settings/cluster",
         "//pkg/util/timeutil",
     ],
 )

pkg/kv/kvserver/closedts/ctpb/rpc_clients.go

@@ -10,24 +10,18 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/rpc/rpcbase"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 )
 
 // DialSideTransportClient establishes a DRPC connection if enabled; otherwise,
 // it falls back to gRPC. The established connection is used to create a
 // RPCSideTransportClient.
 func DialSideTransportClient(
-	nd rpcbase.NodeDialer, ctx context.Context, nodeID roachpb.NodeID, class rpcbase.ConnectionClass,
+	nd rpcbase.NodeDialer,
+	ctx context.Context,
+	nodeID roachpb.NodeID,
+	class rpcbase.ConnectionClass,
+	cs *cluster.Settings,
 ) (RPCSideTransportClient, error) {
-	if !rpcbase.TODODRPC {
-		conn, err := nd.Dial(ctx, nodeID, class)
-		if err != nil {
-			return nil, err
-		}
-		return NewGRPCSideTransportClientAdapter(conn), nil
-	}
-	conn, err := nd.DRPCDial(ctx, nodeID, class)
-	if err != nil {
-		return nil, err
-	}
-	return NewDRPCSideTransportClientAdapter(conn), nil
+	return rpcbase.DialRPCClient(nd, ctx, nodeID, class, NewGRPCSideTransportClientAdapter, NewDRPCSideTransportClientAdapter, cs)
 }

pkg/kv/kvserver/closedts/sidetransport/BUILD.bazel

@@ -30,8 +30,6 @@ go_library(
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_errors//:errors",
-        "@io_storj_drpc//:drpc",
-        "@org_golang_google_grpc//:grpc",
     ],
 )
 

pkg/kv/kvserver/closedts/sidetransport/sender.go

@@ -35,8 +35,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
-	"google.golang.org/grpc"
-	"storj.io/drpc"
 )
 
 // Sender represents the sending-side of the closed timestamps "side-transport".
@@ -687,11 +685,11 @@ type conn interface {
 // rpcConnFactory is an implementation of connFactory that establishes
 // connections to other nodes using gRPC.
 type rpcConnFactory struct {
-	dialer       nodeDialer
+	dialer       rpcbase.NodeDialer
 	testingKnobs connTestingKnobs
 }
 
-func newRPCConnFactory(dialer nodeDialer, testingKnobs connTestingKnobs) connFactory {
+func newRPCConnFactory(dialer rpcbase.NodeDialer, testingKnobs connTestingKnobs) connFactory {
 	return &rpcConnFactory{
 		dialer:       dialer,
 		testingKnobs: testingKnobs,
@@ -703,12 +701,6 @@ func (f *rpcConnFactory) new(s *Sender, nodeID roachpb.NodeID) conn {
 	return newRPCConn(f.dialer, s, nodeID, f.testingKnobs)
 }
 
-// nodeDialer abstracts *nodedialer.Dialer.
-type nodeDialer interface {
-	Dial(ctx context.Context, nodeID roachpb.NodeID, class rpcbase.ConnectionClass) (_ *grpc.ClientConn, err error)
-	DRPCDial(ctx context.Context, nodeID roachpb.NodeID, class rpcbase.ConnectionClass) (_ drpc.Conn, err error)
-}
-
 // On sending errors, we sleep a bit as to not spin on a tripped
 // circuit-breaker in the Dialer.
 const sleepOnErr = time.Second
@@ -720,7 +712,7 @@ const sleepOnErr = time.Second
 // snapshot before we can resume sending regular messages.
 type rpcConn struct {
 	log.AmbientContext
-	dialer       nodeDialer
+	dialer       rpcbase.NodeDialer
 	producer     *Sender
 	nodeID       roachpb.NodeID
 	testingKnobs connTestingKnobs
@@ -739,7 +731,7 @@ type rpcConn struct {
 }
 
 func newRPCConn(
-	dialer nodeDialer, producer *Sender, nodeID roachpb.NodeID, testingKnobs connTestingKnobs,
+	dialer rpcbase.NodeDialer, producer *Sender, nodeID roachpb.NodeID, testingKnobs connTestingKnobs,
 ) conn {
 	r := &rpcConn{
 		dialer:       dialer,
@@ -791,7 +783,7 @@ func (r *rpcConn) maybeConnect(ctx context.Context, _ *stop.Stopper) error {
 		return nil
 	}
 
-	client, err := ctpb.DialSideTransportClient(r.dialer, ctx, r.nodeID, rpcbase.SystemClass)
+	client, err := ctpb.DialSideTransportClient(r.dialer, ctx, r.nodeID, rpcbase.SystemClass, r.producer.st)
 	if err != nil {
 		return err
 	}

pkg/kv/kvserver/closedts/sidetransport/sender_test.go

@@ -622,7 +622,7 @@ type mockDialer struct {
 	}
 }
 
-var _ nodeDialer = &mockDialer{}
+var _ rpcbase.NodeDialer = &mockDialer{}
 
 type nodeAddr struct {
 	nid  roachpb.NodeID
@@ -825,7 +825,7 @@ type failingDialer struct {
 	dialCount int32
 }
 
-var _ nodeDialer = &failingDialer{}
+var _ rpcbase.NodeDialer = &failingDialer{}
 
 func (f *failingDialer) Dial(
 	ctx context.Context, nodeID roachpb.NodeID, class rpcbase.ConnectionClass,
@@ -858,8 +858,14 @@ func TestRPCConnStopOnClose(t *testing.T) {
 
 	dialer := &failingDialer{}
 	factory := newRPCConnFactory(dialer, connTestingKnobs{sleepOnErrOverride: sleepTime})
-	connection := factory.new(nil, /* sender is not needed as dialer always fails Dial attempts */
-		roachpb.NodeID(1))
+
+	s, stopper := newMockSender(factory)
+	defer stopper.Stop(ctx)
+
+	// While sender is strictly not needed to dial a connection as dialer
+	// always fails dial attempts, it is needed to check if DRPC is enabled
+	// or disabled.
+	connection := factory.new(s, roachpb.NodeID(1))
 	connection.run(ctx, stopper)
 
 	// Wait for first dial attempt for sanity reasons.