Merge #148451

148451: kv,server: extract `Cluster` service and register with DRPC server  r=cthumuluru-crdb a=shubhamdhama

`` kv,server: extract `Cluster` service and register with DRPC server ``

This is a follow-up to #145195, where we extracted `KVBatch` from the
`Internal` service. Here, we do the same with `Cluster`. The TL;DR
is that smaller services are easier to maintain and can be more smoothly
migrated to DRPC.

We also enable this service on the DRPC server. This is controlled by
`rpc.experimental_drpc.enabled` (off by default).

This change is part of a series and is similar to: #146926

Note: This only registers the service; the client is not updated to use the
DRPC client, so this service will not have any functional effect.

`` server: replace `RPCClusterClient` for cluster join operation ``

This change replaces the direct client creation through `NewInternalClient`
with the gRPC adapter. See #147950 for more context. Also, since we have a
more targeted service for `Join` RPC, we should use its adapter interface.

Release note: none
Epic: CRDB-48923


Co-authored-by: Shubham Dhama <shubham.dhama@cockroachlabs.com>

Author: craig[bot]

pkg/kv/kvpb/api.proto

@@ -3727,6 +3727,14 @@ service TenantSpanConfig {
   rpc SpanConfigConformance(SpanConfigConformanceRequest) returns (SpanConfigConformanceResponse) {}
 }
 
+// Cluster service offers RPCs for inter-node communication essential for
+// cluster bootstrapping and operation.
+service Cluster {
+  // Join a bootstrapped cluster. If the target node is itself not part of a
+  // bootstrapped cluster, an appropriate error is returned.
+  rpc Join (JoinNodeRequest) returns (JoinNodeResponse) {}
+}
+
 // Batch and RangeFeed service implemented by nodes for KV API requests.
 service Internal {
   rpc Batch (BatchRequest) returns (BatchResponse) {}

pkg/server/init.go

@@ -466,7 +466,10 @@ func (s *initServer) attemptJoinTo(
 		BinaryVersion: &latestVersion,
 	}
 
-	initClient := kvpb.NewInternalClient(conn)
+	var initClient kvpb.RPCClusterClient
+	if !rpcbase.TODODRPC {
+		initClient = kvpb.NewGRPCInternalClientAdapter(conn)
+	}
 	resp, err := initClient.Join(ctx, req)
 	if err != nil {
 		status, ok := grpcstatus.FromError(errors.UnwrapAll(err))

pkg/server/server.go

@@ -995,6 +995,9 @@ func NewServer(cfg Config, stopper *stop.Stopper) (serverctl.ServerStartupInterf
 	if err := kvpb.DRPCRegisterTenantSpanConfig(drpcServer, node); err != nil {
 		return nil, err
 	}
+	if err := kvpb.DRPCRegisterCluster(drpcServer, node); err != nil {
+		return nil, err
+	}
 	kvserver.RegisterPerReplicaServer(grpcServer.Server, node.perReplicaServer)
 	if err := kvserver.DRPCRegisterPerReplica(drpcServer, node.perReplicaServer); err != nil {
 		return nil, err