server: plumb AC pacer support into UA servers

This change threads the AdmissionPacerFactory from the top-level server
down to shared process tenant servers, enabling elastic admission control
for tenant workloads. For separate process tenants, we continue to use a
noop coordinator until full SQL admission control is implemented.

Release note (sql change): clusters utilizing cluster virtualization, such
as those running PCR, now apply the same admission control pacing to various bulk
operations that is used in by clusters not using cluster virtualization.

Epic: none.

Author: dt

pkg/server/server_controller_new_server.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/storage/fs"
 	"github.com/cockroachdb/cockroach/pkg/util"
+	"github.com/cockroachdb/cockroach/pkg/util/admission"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/netutil/addr"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
@@ -78,7 +79,7 @@ func (s *topLevelServer) newTenantServer(
 	// Apply the TestTenantArgs, if any.
 	baseCfg.TestingKnobs = testArgs.Knobs
 
-	tenantServer, err := newTenantServerInternal(ctx, baseCfg, sqlCfg, tenantStopper, tenantNameContainer)
+	tenantServer, err := newTenantServerInternal(ctx, baseCfg, sqlCfg, tenantStopper, tenantNameContainer, s.db.AdmissionPacerFactory)
 	if err != nil {
 		return nil, err
 	}
@@ -130,6 +131,7 @@ func newTenantServerInternal(
 	sqlCfg SQLConfig,
 	stopper *stop.Stopper,
 	tenantNameContainer *roachpb.TenantNameContainer,
+	elastic admission.PacerFactory,
 ) (*SQLServerWrapper, error) {
 	ambientCtx := baseCfg.AmbientCtx
 	stopper.SetTracer(baseCfg.Tracer)
@@ -141,7 +143,7 @@ func newTenantServerInternal(
 	log.Infof(newCtx, "creating tenant server")
 
 	// Now instantiate the tenant server proper.
-	return newSharedProcessTenantServer(newCtx, stopper, baseCfg, sqlCfg, tenantNameContainer)
+	return newSharedProcessTenantServer(newCtx, stopper, baseCfg, sqlCfg, tenantNameContainer, elastic)
 }
 
 func (s *topLevelServer) makeSharedProcessTenantConfig(

pkg/server/tenant.go

@@ -216,8 +216,9 @@ func NewSeparateProcessTenantServer(
 			return spanconfiglimiter.New(ie, st, knobs)
 		},
 	}
-
-	return newTenantServer(ctx, stopper, baseCfg, sqlCfg, tenantNameContainer, deps, mtinfopb.ServiceModeExternal)
+	// TODO(irfansharif): hook up NewGrantCoordinatorSQL.
+	var noopElasticCPUGrantCoord *admission.ElasticCPUGrantCoordinator = nil
+	return newTenantServer(ctx, stopper, baseCfg, sqlCfg, tenantNameContainer, deps, mtinfopb.ServiceModeExternal, noopElasticCPUGrantCoord)
 }
 
 // newSharedProcessTenantServer creates a tenant-specific, SQL-only
@@ -232,6 +233,7 @@ func newSharedProcessTenantServer(
 	baseCfg BaseConfig,
 	sqlCfg SQLConfig,
 	tenantNameContainer *roachpb.TenantNameContainer,
+	elastic admission.PacerFactory,
 ) (*SQLServerWrapper, error) {
 	if baseCfg.IDContainer.Get() == 0 {
 		return nil, errors.AssertionFailedf("programming error: NewSharedProcessTenantServer called before NodeID was assigned.")
@@ -255,7 +257,7 @@ func newSharedProcessTenantServer(
 			return spanconfiglimiter.NoopLimiter{}
 		},
 	}
-	return newTenantServer(ctx, stopper, baseCfg, sqlCfg, tenantNameContainer, deps, mtinfopb.ServiceModeShared)
+	return newTenantServer(ctx, stopper, baseCfg, sqlCfg, tenantNameContainer, deps, mtinfopb.ServiceModeShared, elastic)
 }
 
 // newTenantServer constructs a SQLServerWrapper.
@@ -269,6 +271,7 @@ func newTenantServer(
 	tenantNameContainer *roachpb.TenantNameContainer,
 	deps tenantServerDeps,
 	serviceMode mtinfopb.TenantServiceMode,
+	elastic admission.PacerFactory,
 ) (*SQLServerWrapper, error) {
 	// TODO(knz): Make the license application a per-server thing
 	// instead of a global thing.
@@ -316,7 +319,7 @@ func newTenantServer(
 	// then rely on some mechanism to retrieve the ID from the name to
 	// initialize the rest of the server.
 	baseCfg.idProvider.SetTenantID(sqlCfg.TenantID)
-	args, err := makeTenantSQLServerArgs(ctx, stopper, baseCfg, sqlCfg, tenantNameContainer, deps, serviceMode)
+	args, err := makeTenantSQLServerArgs(ctx, stopper, baseCfg, sqlCfg, tenantNameContainer, deps, serviceMode, elastic)
 	if err != nil {
 		return nil, err
 	}
@@ -1093,6 +1096,7 @@ func makeTenantSQLServerArgs(
 	tenantNameContainer *roachpb.TenantNameContainer,
 	deps tenantServerDeps,
 	serviceMode mtinfopb.TenantServiceMode,
+	elastic admission.PacerFactory,
 ) (sqlServerArgs, error) {
 	st := baseCfg.Settings
 
@@ -1248,7 +1252,7 @@ func makeTenantSQLServerArgs(
 	dbCtx := kv.DefaultDBContext(st, stopper)
 	dbCtx.NodeID = deps.instanceIDContainer
 	db := kv.NewDBWithContext(baseCfg.AmbientCtx, tcsFactory, clock, dbCtx)
-
+	db.AdmissionPacerFactory = elastic
 	rangeFeedKnobs, _ := baseCfg.TestingKnobs.RangeFeed.(*rangefeed.TestingKnobs)
 	rangeFeedFactory, err := rangefeed.NewFactory(stopper, db, st, rangeFeedKnobs)
 	if err != nil {
@@ -1325,8 +1329,6 @@ func makeTenantSQLServerArgs(
 	remoteFlowRunnerAcc := monitorAndMetrics.rootSQLMemoryMonitor.MakeBoundAccount()
 	remoteFlowRunner := flowinfra.NewRemoteFlowRunner(baseCfg.AmbientCtx, stopper, &remoteFlowRunnerAcc)
 
-	// TODO(irfansharif): hook up NewGrantCoordinatorSQL.
-	var noopElasticCPUGrantCoord *admission.ElasticCPUGrantCoordinator = nil
 	return sqlServerArgs{
 		sqlServerOptionalKVArgs: sqlServerOptionalKVArgs{
 			nodesStatusServer: serverpb.MakeOptionalNodesStatusServer(nil),
@@ -1381,7 +1383,7 @@ func makeTenantSQLServerArgs(
 		grpc:                     grpcServer,
 		drpc:                     drpcServer,
 		externalStorageBuilder:   esb,
-		admissionPacerFactory:    noopElasticCPUGrantCoord,
+		admissionPacerFactory:    elastic,
 		rangeDescIteratorFactory: tenantConnect,
 		tenantTimeSeriesServer:   sTS,
 		tenantCapabilitiesReader: sql.EmptySystemTenantOnly[tenantcapabilities.Reader](),

pkg/sql/row/kv_batch_fetcher.go

@@ -61,7 +61,7 @@ var logAdmissionPacerErr = log.Every(100 * time.Millisecond)
 // each time we seek admission for response handling during internally submitted
 // low priority reads (like row-level TTL selects).
 var elasticCPUDurationPerLowPriReadResponse = settings.RegisterDurationSetting(
-	settings.SystemOnly,
+	settings.ApplicationLevel,
 	"sqladmission.elastic_cpu.duration_per_low_pri_read_response",
 	"controls how many CPU tokens are allotted for handling responses for internally submitted low priority reads",
 	// NB: Experimentally, during TTL reads, we observed cumulative on-CPU time
@@ -75,7 +75,7 @@ var elasticCPUDurationPerLowPriReadResponse = settings.RegisterDurationSetting(
 // internally submitted low-priority reads (like row-level TTL selects)
 // integrate with elastic CPU control.
 var internalLowPriReadElasticControlEnabled = settings.RegisterBoolSetting(
-	settings.SystemOnly,
+	settings.ApplicationLevel,
 	"sqladmission.low_pri_read_response_elastic_control.enabled",
 	"determines whether the sql portion of internally submitted reads integrate with elastic CPU controller",
 	true,