kv: extract TenantService 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 `TenantService`. 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.

Epic: CRDB-48925
Release note: None

Author: shubhamdhama

pkg/kv/kvpb/api.proto

@@ -3650,19 +3650,40 @@ message JoinNodeResponse {
   roachpb.Version active_version = 4;
 }
 
+// TODO(server): Remove the methods of these serivces from the 'Internal'
+// service once the migration to DRPC complete, making these service the sole
+// definition. Also, update the comments here.
+
+// TODO(server,kv): Improve the code comment documentation of these methods.
+// Many of these methods are simply undocumented.
+
 // KVBatch defines Batch RPCs of Internal service (see below) for DRPC use only.
 //
 // During the gRPC to DRPC migration:
 // - gRPC clients will continue to use the full 'Internal' service client.
 // - DRPC clients targeting Batch* RPCs will use clients generated from this.
-//
-// TODO(server): Remove these methods from the 'Internal' service once the
-// migration to DRPC complete, making this service the sole definition.
 service KVBatch {
   rpc Batch(BatchRequest) returns (BatchResponse) {}
   rpc BatchStream(stream BatchRequest) returns (stream BatchResponse) {}
 }
 
+// TenantService provides RPCs for secondary tenants to access data that is
+// exclusively available in the system tenant. This service is used by the
+// tenant connector.
+service TenantService {
+  // TenantSettings is used by tenants to obtain and stay up to date with tenant
+  // setting overrides.
+  rpc TenantSettings(TenantSettingsRequest) returns (stream TenantSettingsEvent) {}
+
+  rpc GossipSubscription(GossipSubscriptionRequest) returns (stream GossipSubscriptionEvent) {}
+
+  rpc RangeLookup(RangeLookupRequest) returns (RangeLookupResponse) {}
+
+  // GetRangeDescriptors is used by tenants to get range descriptors for their
+  // own ranges.
+  rpc GetRangeDescriptors(GetRangeDescriptorsRequest) returns (stream GetRangeDescriptorsResponse) {}
+}
+
 // Batch and RangeFeed service implemented by nodes for KV API requests.
 service Internal {
   rpc Batch (BatchRequest) returns (BatchResponse) {}

pkg/server/node.go

@@ -1899,19 +1899,19 @@ func (n *Node) batchStreamImpl(stream kvpb.RPCKVBatch_BatchStreamStream) error {
 	}
 }
 
+type kvBatchServer Node
+
 func (n *Node) AsDRPCKVBatchServer() kvpb.DRPCKVBatchServer {
-	return (*drpcNode)(n)
+	return (*kvBatchServer)(n)
 }
 
-type drpcNode Node
-
-func (n *drpcNode) Batch(
+func (n *kvBatchServer) Batch(
 	ctx context.Context, request *kvpb.BatchRequest,
 ) (*kvpb.BatchResponse, error) {
 	return (*Node)(n).Batch(ctx, request)
 }
 
-func (n *drpcNode) BatchStream(stream kvpb.DRPCKVBatch_BatchStreamStream) error {
+func (n *kvBatchServer) BatchStream(stream kvpb.DRPCKVBatch_BatchStreamStream) error {
 	return (*Node)(n).batchStreamImpl(stream)
 }
 
@@ -2055,7 +2055,6 @@ func filterRangeLookupResponseForTenant(
 	return truncated
 }
 
-// RangeLookup implements the kvpb.InternalServer interface.
 func (n *Node) RangeLookup(
 	ctx context.Context, req *kvpb.RangeLookupRequest,
 ) (*kvpb.RangeLookupResponse, error) {
@@ -2392,6 +2391,12 @@ func (n *Node) ResetQuorum(
 // GossipSubscription implements the kvpb.InternalServer interface.
 func (n *Node) GossipSubscription(
 	args *kvpb.GossipSubscriptionRequest, stream kvpb.Internal_GossipSubscriptionServer,
+) error {
+	return n.gossipSubscription(args, stream)
+}
+
+func (n *Node) gossipSubscription(
+	args *kvpb.GossipSubscriptionRequest, stream kvpb.RPCInternal_GossipSubscriptionStream,
 ) error {
 	ctx := n.storeCfg.AmbientCtx.AnnotateCtx(stream.Context())
 	ctxDone := ctx.Done()
@@ -2481,6 +2486,12 @@ func (n *Node) waitForTenantWatcherReadiness(
 // TenantSettings implements the kvpb.InternalServer interface.
 func (n *Node) TenantSettings(
 	args *kvpb.TenantSettingsRequest, stream kvpb.Internal_TenantSettingsServer,
+) error {
+	return n.tenantSettings(args, stream)
+}
+
+func (n *Node) tenantSettings(
+	args *kvpb.TenantSettingsRequest, stream kvpb.RPCTenantService_TenantSettingsStream,
 ) error {
 	ctx := n.storeCfg.AmbientCtx.AnnotateCtx(stream.Context())
 	ctxDone := ctx.Done()
@@ -2888,6 +2899,12 @@ func (n *Node) SpanConfigConformance(
 func (n *Node) GetRangeDescriptors(
 	args *kvpb.GetRangeDescriptorsRequest, stream kvpb.Internal_GetRangeDescriptorsServer,
 ) error {
+	return n.getRangeDescriptors(args, stream)
+}
+
+func (n *Node) getRangeDescriptors(
+	args *kvpb.GetRangeDescriptorsRequest, stream kvpb.RPCTenantService_GetRangeDescriptorsStream,
+) error {
 
 	iter, err := n.execCfg.RangeDescIteratorFactory.NewLazyIterator(stream.Context(), args.Span, int(args.BatchSize))
 	if err != nil {
@@ -2914,3 +2931,37 @@ func (n *Node) GetRangeDescriptors(
 		RangeDescriptors: rangeDescriptors,
 	})
 }
+
+type tenantServiceServer Node
+
+func (n *Node) AsDRPCTenantServiceServer() kvpb.DRPCTenantServiceServer {
+	return (*tenantServiceServer)(n)
+}
+
+// GossipSubscription implements the kvpb.DRPCTenantServiceServer interface.
+func (n *tenantServiceServer) GossipSubscription(
+	args *kvpb.GossipSubscriptionRequest, stream kvpb.DRPCTenantService_GossipSubscriptionStream,
+) error {
+	return (*Node)(n).gossipSubscription(args, stream)
+}
+
+// TenantSettings implements the kvpb.DRPCTenantServiceServer interface.
+func (n *tenantServiceServer) TenantSettings(
+	args *kvpb.TenantSettingsRequest, stream kvpb.DRPCTenantService_TenantSettingsStream,
+) error {
+	return (*Node)(n).tenantSettings(args, stream)
+}
+
+// RangeLookup implements the kvpb.DRPCTenantServiceServer interface.
+func (n *tenantServiceServer) RangeLookup(
+	ctx context.Context, req *kvpb.RangeLookupRequest,
+) (*kvpb.RangeLookupResponse, error) {
+	return (*Node)(n).RangeLookup(ctx, req)
+}
+
+// GetRangeDescriptors implements the kvpb.DRPCTenantServiceServer interface.
+func (n *tenantServiceServer) GetRangeDescriptors(
+	args *kvpb.GetRangeDescriptorsRequest, stream kvpb.DRPCTenantService_GetRangeDescriptorsStream,
+) error {
+	return (*Node)(n).getRangeDescriptors(args, stream)
+}

pkg/server/server.go

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