Merge pull request #154260 from cockroachdb/blathers/backport-release-25.4-154012

release-25.4: server,drpc: disable DRPC for REST-RPC bridge

Author: shubhamdhama

pkg/acceptance/localcluster/cluster.go

@@ -478,7 +478,7 @@ func (n *Node) StatusClient(ctx context.Context) serverpb.RPCStatusClient {
 		return existingClient
 	}
 
-	if !rpcbase.DRPCEnabled(ctx, n.rpcCtx.Settings) {
+	if !rpcbase.TODODRPC || !rpcbase.DRPCEnabled(ctx, n.rpcCtx.Settings) {
 		conn, err := n.rpcCtx.GRPCUnvalidatedDial(n.RPCAddr(), roachpb.Locality{}).Connect(ctx)
 		if err != nil {
 			log.Dev.Fatalf(context.Background(), "failed to initialize status client: %s", err)

pkg/rpc/rpcbase/nodedialer.go

@@ -27,6 +27,10 @@ var ExperimentalDRPCEnabled = settings.RegisterBoolSetting(
 	"if true, use drpc to execute Batch RPCs (instead of gRPC)",
 	envExperimentalDRPCEnabled)
 
+// TODODRPC is a marker to identify sites that needs to be updated to support
+// DRPC.
+const TODODRPC = false
+
 // NodeDialer interface defines methods for dialing peer nodes using their
 // node IDs.
 type NodeDialer interface {

pkg/server/application_api/config_test.go

@@ -66,7 +66,9 @@ func TestAdminAPISettings(t *testing.T) {
 		"view_activity_user":          {userName: "view_activity_user", redactableSettings: true, grantRole: "SYSTEM VIEWACTIVITY", consoleOnly: true},
 		"view_activity_redacted_user": {userName: "view_activity_redacted_user", redactableSettings: true, grantRole: "SYSTEM VIEWACTIVITYREDACTED", consoleOnly: true},
 	}
-	ts := serverutils.StartServerOnly(t, base.TestServerArgs{})
+	ts := serverutils.StartServerOnly(t, base.TestServerArgs{
+		DefaultDRPCOption: base.TestDRPCDisabled,
+	})
 	defer ts.Stopper().Stop(ctx)
 	forSystemTenant := ts.ApplicationLayer().Codec().ForSystemTenant()
 	conn := sqlutils.MakeSQLRunner(ts.ApplicationLayer().SQLConn(t))

pkg/server/application_api/sql_stats_test.go

@@ -1643,7 +1643,9 @@ func TestDrainSqlStats_partialOutage(t *testing.T) {
 func TestDrainSqlStatsPermissionDenied(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
-	ts := serverutils.StartServerOnly(t, base.TestServerArgs{})
+	ts := serverutils.StartServerOnly(t, base.TestServerArgs{
+		DefaultDRPCOption: base.TestDRPCDisabled,
+	})
 	ctx := context.Background()
 	nonRootUser := apiconstants.TestingUserNameNoAdmin()
 	sqlutils.MakeSQLRunner(ts.SQLConn(t)).Exec(t, fmt.Sprintf("CREATE USER IF NOT EXISTS %s", nonRootUser))

pkg/server/server.go

@@ -2130,7 +2130,7 @@ func (s *topLevelServer) PreStart(ctx context.Context) error {
 		}
 	}
 	var apiInternalServer http.Handler
-	if rpcbase.DRPCEnabled(ctx, s.cfg.Settings) {
+	if rpcbase.TODODRPC && rpcbase.DRPCEnabled(ctx, s.cfg.Settings) {
 		// Pass our own node ID to connect to local RPC servers
 		apiInternalServer, err = apiinternal.NewAPIInternalServer(
 			ctx, s.kvNodeDialer, s.rpcContext.NodeID.Get(), s.cfg.Settings)

pkg/server/serverpb/rpc_clients.go

@@ -44,8 +44,15 @@ func DialStatusClientNoBreaker(
 func DialStatusClient(
 	nd rpcbase.NodeDialer, ctx context.Context, nodeID roachpb.NodeID, cs *cluster.Settings,
 ) (RPCStatusClient, error) {
-	return rpcbase.DialRPCClient(nd, ctx, nodeID, rpcbase.DefaultClass,
-		NewGRPCStatusClientAdapter, NewDRPCStatusClientAdapter, cs)
+	if rpcbase.TODODRPC {
+		return rpcbase.DialRPCClient(nd, ctx, nodeID, rpcbase.DefaultClass,
+			NewGRPCStatusClientAdapter, NewDRPCStatusClientAdapter, cs)
+	}
+	conn, err := nd.Dial(ctx, nodeID, rpcbase.DefaultClass)
+	if err != nil {
+		return nil, err
+	}
+	return NewGRPCStatusClientAdapter(conn), nil
 }
 
 // DialAdminClient establishes a DRPC connection if enabled; otherwise, it
@@ -54,8 +61,15 @@ func DialStatusClient(
 func DialAdminClient(
 	nd rpcbase.NodeDialer, ctx context.Context, nodeID roachpb.NodeID, cs *cluster.Settings,
 ) (RPCAdminClient, error) {
-	return rpcbase.DialRPCClient(nd, ctx, nodeID, rpcbase.DefaultClass,
-		NewGRPCAdminClientAdapter, NewDRPCAdminClientAdapter, cs)
+	if rpcbase.TODODRPC {
+		return rpcbase.DialRPCClient(nd, ctx, nodeID, rpcbase.DefaultClass,
+			NewGRPCAdminClientAdapter, NewDRPCAdminClientAdapter, cs)
+	}
+	conn, err := nd.Dial(ctx, nodeID, rpcbase.DefaultClass)
+	if err != nil {
+		return nil, err
+	}
+	return NewGRPCAdminClientAdapter(conn), nil
 }
 
 // DialAdminClientNoBreaker establishes a DRPC connection if enabled;

pkg/server/status_test.go

@@ -581,6 +581,7 @@ func TestStatusUpdateTableMetadataCache(t *testing.T) {
 	ctx := context.Background()
 	tc := serverutils.StartCluster(t, 3, base.TestClusterArgs{
 		ServerArgs: base.TestServerArgs{
+			DefaultDRPCOption: base.TestDRPCDisabled,
 			Knobs: base.TestingKnobs{
 				JobsTestingKnobs: &jobs.TestingKnobs{
 					IntervalOverrides: jobs.TestingIntervalOverrides{

pkg/server/tenant.go

@@ -826,7 +826,7 @@ func (s *SQLServerWrapper) PreStart(ctx context.Context) error {
 	}
 
 	var apiInternalServer http.Handler
-	if rpcbase.DRPCEnabled(ctx, s.cfg.Settings) {
+	if rpcbase.TODODRPC && rpcbase.DRPCEnabled(ctx, s.cfg.Settings) {
 		// Pass our own instance ID to connect to local RPC servers
 		apiInternalServer, err = apiinternal.NewAPIInternalServer(ctx,
 			s.sqlServer.sqlInstanceDialer,