Merge #147950

147950: *: replace grpc clients with rpc adapters r=cthumuluru-crdb a=cthumuluru-crdb

gRPC and DRPC clients have different signatures. To unify them, we created a common interface for both types of clients, and replaced the gRPC and DRPC clients with adapters.

Epic: CRDB-48923
Informs: #148353
Release note: none

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

Author: craig[bot]

pkg/acceptance/localcluster/cluster.go

@@ -413,7 +413,7 @@ type Node struct {
 	cmd            *exec.Cmd
 	rpcPort, pgURL string // legacy: remove once 1.0.x is no longer tested
 	db             *gosql.DB
-	statusClient   serverpb.StatusClient
+	statusClient   serverpb.RPCStatusClient
 }
 
 // RPCPort returns the RPC + Postgres port.
@@ -469,7 +469,7 @@ func (n *Node) Alive() bool {
 }
 
 // StatusClient returns a StatusClient set up to talk to this node.
-func (n *Node) StatusClient(ctx context.Context) serverpb.StatusClient {
+func (n *Node) StatusClient(ctx context.Context) serverpb.RPCStatusClient {
 	n.Lock()
 	existingClient := n.statusClient
 	n.Unlock()
@@ -483,7 +483,7 @@ func (n *Node) StatusClient(ctx context.Context) serverpb.StatusClient {
 		if err != nil {
 			log.Fatalf(context.Background(), "failed to initialize status client: %s", err)
 		}
-		return serverpb.NewStatusClient(conn)
+		return serverpb.NewGRPCStatusClientAdapter(conn)
 	}
 	return nil // This should never happen
 }

pkg/blobs/blobspb/rpc_clients.go

@@ -14,16 +14,16 @@ import (
 
 // DialBlobClient establishes a DRPC connection if enabled; otherwise,
 // it falls back to gRPC. The established connection is used to create a
-// BlobClient.
+// RPCBlobClient.
 func DialBlobClient(
 	nd rpcbase.NodeDialer, ctx context.Context, nodeID roachpb.NodeID, class rpcbase.ConnectionClass,
-) (BlobClient, error) {
+) (RPCBlobClient, error) {
 	if !rpcbase.TODODRPC {
 		conn, err := nd.Dial(ctx, nodeID, class)
 		if err != nil {
 			return nil, err
 		}
-		return NewBlobClient(conn), nil
+		return NewGRPCBlobClientAdapter(conn), nil
 	}
 	return nil, nil
 }

pkg/blobs/client.go

@@ -47,11 +47,11 @@ var _ BlobClient = &remoteClient{}
 // remoteClient uses the node dialer and blob service clients
 // to Read or Write bulk files from/to other nodes.
 type remoteClient struct {
-	blobClient blobspb.BlobClient
+	blobClient blobspb.RPCBlobClient
 }
 
 // newRemoteClient instantiates a remote blob service client.
-func newRemoteClient(blobClient blobspb.BlobClient) BlobClient {
+func newRemoteClient(blobClient blobspb.RPCBlobClient) BlobClient {
 	return &remoteClient{blobClient: blobClient}
 }
 
@@ -71,7 +71,7 @@ func (c *remoteClient) ReadFile(
 }
 
 type streamWriter struct {
-	s   blobspb.Blob_PutStreamClient
+	s   blobspb.RPCBlob_PutStreamClient
 	buf blobspb.StreamChunk
 }
 

pkg/blobs/stream.go

@@ -43,7 +43,7 @@ type streamReceiver interface {
 // This is needed as Blob_GetStreamClient does not have a Close() function, whereas
 // the other sender, Blob_PutStreamServer, does.
 type nopSendAndClose struct {
-	blobspb.Blob_GetStreamClient
+	blobspb.RPCBlob_GetStreamClient
 }
 
 func (*nopSendAndClose) SendAndClose(*blobspb.StreamResponse) error {
@@ -52,7 +52,7 @@ func (*nopSendAndClose) SendAndClose(*blobspb.StreamResponse) error {
 
 // newGetStreamReader creates an io.ReadCloser that uses gRPC's streaming API
 // to read chunks of data.
-func newGetStreamReader(client blobspb.Blob_GetStreamClient) ioctx.ReadCloserCtx {
+func newGetStreamReader(client blobspb.RPCBlob_GetStreamClient) ioctx.ReadCloserCtx {
 	return &blobStreamReader{
 		stream: &nopSendAndClose{client},
 	}

pkg/ccl/serverccl/statusccl/tenant_grpc_test.go

@@ -135,7 +135,7 @@ func TestTenantGRPCServices(t *testing.T) {
 		conn, err := rpcCtx.GRPCDialNode(grpcAddr, nodeID, roachpb.Locality{}, rpcbase.DefaultClass).Connect(ctx)
 		require.NoError(t, err)
 
-		client := serverpb.NewStatusClient(conn)
+		client := serverpb.NewGRPCStatusClientAdapter(conn)
 
 		resp, err := client.Statements(ctx, &serverpb.StatementsRequest{NodeID: "local"})
 		require.NoError(t, err)
@@ -149,7 +149,7 @@ func TestTenantGRPCServices(t *testing.T) {
 		conn, err := rpcCtx.GRPCDialNode(grpcAddr, server.NodeID(), roachpb.Locality{}, rpcbase.DefaultClass).Connect(ctx)
 		require.NoError(t, err)
 
-		client := serverpb.NewStatusClient(conn)
+		client := serverpb.NewGRPCStatusClientAdapter(conn)
 
 		_, err = client.Statements(ctx, &serverpb.StatementsRequest{NodeID: "local"})
 		require.Errorf(t, err, "statements endpoint should not be accessed on KV node by tenant")

pkg/ccl/sqlproxyccl/proxy_handler.go

@@ -279,7 +279,7 @@ func newProxyHandler(
 		dirOpts = append(dirOpts, handler.testingKnobs.dirOpts...)
 	}
 
-	client := tenant.NewDirectoryClient(conn)
+	client := tenant.NewGRPCDirectoryClientAdapter(conn)
 	handler.directoryCache, err = tenant.NewDirectoryCache(ctx, stopper, client, dirOpts...)
 	if err != nil {
 		return nil, err

pkg/ccl/sqlproxyccl/tenant/directory_cache.go

@@ -114,7 +114,7 @@ func TenantWatcher(tenantWatcher chan *WatchTenantsResponse) func(opts *dirOptio
 type directoryCache struct {
 	// client is the directory client instance used to make directory server
 	// calls.
-	client DirectoryClient
+	client RPCDirectoryClient
 
 	// stopper is used for graceful shutdown of the pod watcher.
 	stopper *stop.Stopper
@@ -144,7 +144,7 @@ var _ DirectoryCache = &directoryCache{}
 // NOTE: stopper.Stop must be called on the directory when it is no longer
 // needed.
 func NewDirectoryCache(
-	ctx context.Context, stopper *stop.Stopper, client DirectoryClient, opts ...DirOption,
+	ctx context.Context, stopper *stop.Stopper, client RPCDirectoryClient, opts ...DirOption,
 ) (DirectoryCache, error) {
 	dir := &directoryCache{client: client, stopper: stopper}
 
@@ -380,7 +380,7 @@ func (d *directoryCache) watchPods(ctx context.Context, stopper *stop.Stopper) e
 	waitInit.Add(1)
 
 	err := stopper.RunAsyncTask(ctx, "watch-pods-client", func(ctx context.Context) {
-		var client Directory_WatchPodsClient
+		var client RPCDirectory_WatchPodsClient
 		var err error
 		firstRun := true
 		ctx, cancel := stopper.WithCancelOnQuiesce(ctx)
@@ -492,7 +492,7 @@ func (d *directoryCache) updateTenantPodEntry(ctx context.Context, pod *Pod) {
 // notification and update the directory to reflect that change.
 func (d *directoryCache) watchTenants(ctx context.Context, stopper *stop.Stopper) error {
 	return stopper.RunAsyncTask(ctx, "watch-tenants-client", func(ctx context.Context) {
-		var client Directory_WatchTenantsClient
+		var client RPCDirectory_WatchTenantsClient
 		var err error
 		ctx, cancel := stopper.WithCancelOnQuiesce(ctx)
 		defer cancel()

pkg/ccl/sqlproxyccl/tenant/directory_cache_test.go

@@ -697,7 +697,7 @@ func newTestDirectoryCache(
 	require.NoError(t, err)
 	// nolint:grpcconnclose
 	clusterStopper.AddCloser(stop.CloserFn(func() { require.NoError(t, conn.Close() /* nolint:grpcconnclose */) }))
-	client := tenant.NewDirectoryClient(conn)
+	client := tenant.NewGRPCDirectoryClientAdapter(conn)
 	directoryCache, err = tenant.NewDirectoryCache(context.Background(), clusterStopper, client, opts...)
 	require.NoError(t, err)
 	return

pkg/ccl/sqlproxyccl/tenant/entry.go

@@ -72,7 +72,7 @@ type tenantEntry struct {
 // Initialize fetches metadata about a tenant, such as its cluster name, and
 // stores that in the entry. If the tenant's metadata is stale, they will be
 // refreshed.
-func (e *tenantEntry) Initialize(ctx context.Context, client DirectoryClient) error {
+func (e *tenantEntry) Initialize(ctx context.Context, client RPCDirectoryClient) error {
 	e.calls.Lock()
 	defer e.calls.Unlock()
 
@@ -112,7 +112,7 @@ func (e *tenantEntry) Initialize(ctx context.Context, client DirectoryClient) er
 
 // RefreshPods makes a synchronous directory server call to fetch the latest
 // information about the tenant's available pods, such as their IP addresses.
-func (e *tenantEntry) RefreshPods(ctx context.Context, client DirectoryClient) error {
+func (e *tenantEntry) RefreshPods(ctx context.Context, client RPCDirectoryClient) error {
 	// Lock so that only one thread at a time will refresh, since there's no
 	// point in multiple threads doing it within a short span of time - it's
 	// likely nothing has changed.
@@ -181,7 +181,7 @@ func (e *tenantEntry) GetPods() []*Pod {
 // errorIfNoPods is true, then EnsureTenantPod returns an error if there are no
 // pods available rather than blocking.
 func (e *tenantEntry) EnsureTenantPod(
-	ctx context.Context, client DirectoryClient, errorIfNoPods bool,
+	ctx context.Context, client RPCDirectoryClient, errorIfNoPods bool,
 ) (pods []*Pod, err error) {
 	const retryDelay = 100 * time.Millisecond
 
@@ -282,7 +282,7 @@ func (e *tenantEntry) ToProto() *Tenant {
 //
 // NOTE: Caller must lock the "calls" mutex before calling fetchPodsLocked.
 func (e *tenantEntry) fetchPodsLocked(
-	ctx context.Context, client DirectoryClient,
+	ctx context.Context, client RPCDirectoryClient,
 ) (tenantPods []*Pod, err error) {
 	// List the pods for the given tenant.
 	//

pkg/ccl/sqlproxyccl/tenantdirsvr/testutils.go

@@ -44,7 +44,7 @@ func SetupTestDirectory(
 	stopper.AddCloser(stop.CloserFn(func() {
 		_ = conn.Close() // nolint:grpcconnclose
 	}))
-	client := tenant.NewDirectoryClient(conn)
+	client := tenant.NewGRPCDirectoryClientAdapter(conn)
 	directoryCache, err := tenant.NewDirectoryCache(ctx, stopper, client, opts...)
 	require.NoError(t, err)