server: register Timeseries service with DRPC server

Enable these service on the DRPC server in addition to gRPC. 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/server/server.go

@@ -1256,6 +1256,9 @@ func NewServer(cfg Config, stopper *stop.Stopper) (serverctl.ServerStartupInterf
 	if err := sAuth.RegisterDRPCService(drpcServer); err != nil {
 		return nil, err
 	}
+	if err := sTS.RegisterDRPCService(drpcServer); err != nil {
+		return nil, err
+	}
 
 	// Tell the node event logger (join, restart) how to populate SQL entries
 	// into system.eventlog.

pkg/server/tenant.go

@@ -483,6 +483,9 @@ func newTenantServer(
 	if err := sAuth.RegisterDRPCService(args.drpc); err != nil {
 		return nil, err
 	}
+	if err := args.tenantTimeSeriesServer.RegisterDRPCService(args.drpc); err != nil {
+		return nil, err
+	}
 
 	// Tell the status/admin servers how to access SQL structures.
 	sStatus.setStmtDiagnosticsRequester(sqlServer.execCfg.StmtDiagnosticsRecorder)

pkg/ts/BUILD.bazel

@@ -42,6 +42,7 @@ go_library(
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
         "@com_github_grpc_ecosystem_grpc_gateway//runtime:go_default_library",
+        "@io_storj_drpc//:drpc",
         "@org_golang_google_grpc//:grpc",
         "@org_golang_google_grpc//codes",
         "@org_golang_google_grpc//status",

pkg/ts/server.go

@@ -26,6 +26,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"storj.io/drpc"
 )
 
 const (
@@ -133,6 +134,15 @@ func (s *TenantServer) RegisterService(g *grpc.Server) {
 	tspb.RegisterTimeSeriesServer(g, s)
 }
 
+type drpcTenantServer struct {
+	*TenantServer
+}
+
+// RegisterService registers the DRPC service.
+func (s *TenantServer) RegisterDRPCService(d drpc.Mux) error {
+	return tspb.DRPCRegisterTimeSeries(d, &drpcTenantServer{TenantServer: s})
+}
+
 // RegisterGateway starts the gateway (i.e. reverse proxy) that proxies HTTP requests
 // to the appropriate gRPC endpoints.
 func (s *TenantServer) RegisterGateway(
@@ -210,6 +220,15 @@ func (s *Server) RegisterService(g *grpc.Server) {
 	tspb.RegisterTimeSeriesServer(g, s)
 }
 
+type drpcServer struct {
+	*Server
+}
+
+// RegisterService registers the DRPC service.
+func (s *Server) RegisterDRPCService(d drpc.Mux) error {
+	return tspb.DRPCRegisterTimeSeries(d, &drpcServer{Server: s})
+}
+
 // RegisterGateway starts the gateway (i.e. reverse proxy) that proxies HTTP requests
 // to the appropriate gRPC endpoints.
 func (s *Server) RegisterGateway(
@@ -377,25 +396,63 @@ func (s *Server) Query(
 // set up a KV store and write some keys into it (`MakeDataKey`) to do so without
 // setting up a `*Server`.
 func (s *Server) Dump(req *tspb.DumpRequest, stream tspb.TimeSeries_DumpServer) error {
+	return s.dump(req, stream)
+}
+
+// Dump returns a stream of raw timeseries data that has been stored on the
+// server.
+func (s *drpcServer) Dump(req *tspb.DumpRequest, stream tspb.DRPCTimeSeries_DumpStream) error {
+	return s.dump(req, stream)
+}
+
+func (s *Server) dump(req *tspb.DumpRequest, stream tspb.RPCTimeSeries_DumpStream) error {
 	d := DefaultDumper{stream.Send}.Dump
 	return dumpImpl(stream.Context(), s.db.db, req, d)
 
 }
 
 // DumpRaw is like Dump, but it returns a stream of raw KV pairs.
 func (s *Server) DumpRaw(req *tspb.DumpRequest, stream tspb.TimeSeries_DumpRawServer) error {
+	return s.dumpRaw(req, stream)
+}
+
+// DumpRaw is like Dump, but it returns a stream of raw KV pairs.
+func (s *drpcServer) DumpRaw(
+	req *tspb.DumpRequest, stream tspb.DRPCTimeSeries_DumpRawStream,
+) error {
+	return s.dumpRaw(req, stream)
+}
+
+func (s *Server) dumpRaw(req *tspb.DumpRequest, stream tspb.RPCTimeSeries_DumpRawStream) error {
 	d := rawDumper{stream}.Dump
 	return dumpImpl(stream.Context(), s.db.db, req, d)
 }
 
-func (s *TenantServer) DumpRaw(_ *tspb.DumpRequest, _ tspb.TimeSeries_DumpRawServer) error {
+// DumpRaw is like Dump, but it returns a stream of raw KV pairs.
+func (s *drpcTenantServer) DumpRaw(_ *tspb.DumpRequest, _ tspb.DRPCTimeSeries_DumpRawStream) error {
+	return s.dumpRaw()
+}
+
+func (t *TenantServer) DumpRaw(_ *tspb.DumpRequest, _ tspb.TimeSeries_DumpRawServer) error {
+	return t.dumpRaw()
+}
+
+func (t *TenantServer) dumpRaw() error {
 	return status.Errorf(codes.Unimplemented, "DumpRaw is not implemented for virtual clusters. "+
 		"If you are attempting to take a tsdump, please connect to the system virtual cluster, "+
 		"not an application virtual cluster. System virtual clusters will dump all persisted "+
 		"metrics from all virtual clusters.")
 }
 
-func (s *TenantServer) Dump(_ *tspb.DumpRequest, _ tspb.TimeSeries_DumpServer) error {
+func (s *drpcTenantServer) Dump(_ *tspb.DumpRequest, _ tspb.DRPCTimeSeries_DumpStream) error {
+	return s.dump()
+}
+
+func (t *TenantServer) Dump(_ *tspb.DumpRequest, _ tspb.TimeSeries_DumpServer) error {
+	return t.dump()
+}
+
+func (t *TenantServer) dump() error {
 	return status.Errorf(codes.Unimplemented, "Dump is not implemented for virtual clusters. "+
 		"If you are attempting to take a tsdump, please connect to the system virtual cluster, "+
 		"not an application virtual cluster. System virtual clusters will dump all persisted "+
@@ -466,7 +523,7 @@ func (dd DefaultDumper) Dump(kv *roachpb.KeyValue) error {
 }
 
 type rawDumper struct {
-	stream tspb.TimeSeries_DumpRawServer
+	stream tspb.RPCTimeSeries_DumpRawStream
 }
 
 func (rd rawDumper) Dump(kv *roachpb.KeyValue) error {