Merge #155783

155783: sql: several preliminary changes to making tests work with secondary tenants r=yuzefovich a=yuzefovich

This PR contains several commits which audit tests in `pkg/sql` directory to ensure the right JobRegistry, codec, LeaseManager, and PGUrl are used. Almost all were pretty mechanical changes. Note that no new tests are enabled with secondary tenants yet - this will be done separately.

Epic: CRDB-48945


Co-authored-by: Yahor Yuzefovich <[email protected]>

Author: craig[bot]

pkg/ccl/multiregionccl/cold_start_latency_test.go

@@ -323,7 +323,7 @@ COMMIT;`}
 		defer tenant.AppStopper().Stop(ctx)
 		pgURL, cleanup, err := pgurlutils.PGUrlWithOptionalClientCertsE(
 			tenant.AdvSQLAddr(), "tenantdata", url.UserPassword("foo", password),
-			false, "", // withClientCerts
+			false /* withClientCerts */, "", /* certName */
 		)
 		if !assert.NoError(t, err) {
 			return

pkg/server/server_systemlog_gc.go

@@ -8,7 +8,7 @@ package server
 import (
 	"context"
 	"fmt"
-	math_rand "math/rand"
+	"math/rand"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
@@ -288,5 +288,5 @@ func startSystemLogsGC(ctx context.Context, sqlServer *SQLServer) error {
 // jitteredInterval returns a randomly jittered (+/-25%) duration
 // from the interval.
 func jitteredInterval(interval time.Duration) time.Duration {
-	return time.Duration(float64(interval) * (0.75 + 0.5*math_rand.Float64()))
+	return time.Duration(float64(interval) * (0.75 + 0.5*rand.Float64()))
 }

pkg/server/testserver.go

@@ -516,6 +516,7 @@ func (ts *testServer) SQLConnE(opts ...serverutils.SQLConnOption) (*gosql.DB, er
 		ts.cfg.Insecure,
 		options.ClientCerts,
 		options.CertsDirPrefix,
+		options.CertName,
 	)
 }
 
@@ -544,6 +545,7 @@ func (ts *testServer) PGUrlE(opts ...serverutils.SQLConnOption) (url.URL, func()
 		ts.cfg.Insecure,
 		options.ClientCerts,
 		options.CertsDirPrefix,
+		options.CertName,
 	)
 }
 
@@ -1039,6 +1041,7 @@ func (t *testTenant) SQLConnE(opts ...serverutils.SQLConnOption) (*gosql.DB, err
 		t.Cfg.Insecure,
 		options.ClientCerts,
 		options.CertsDirPrefix,
+		options.CertName,
 	)
 }
 
@@ -1073,6 +1076,7 @@ func (t *testTenant) PGUrlE(opts ...serverutils.SQLConnOption) (url.URL, func(),
 		t.Cfg.Insecure,
 		options.ClientCerts,
 		options.CertsDirPrefix,
+		options.CertName,
 	)
 }
 

pkg/server/testserver_sqlconn.go

@@ -45,6 +45,7 @@ func pgURL(
 	insecure bool,
 	clientCerts bool,
 	prefix string,
+	certName string,
 ) (pgURL url.URL, cleanupFn func(), err error) {
 	opts := url.Values{}
 	if tenantName != "" && !strings.HasPrefix(dbName, "cluster:") {
@@ -65,7 +66,7 @@ func pgURL(
 	}
 
 	// No LoopbackListener
-	pgURL, cleanupFn, err = pgurlutils.PGUrlWithOptionalClientCertsE(sqlAddr, prefix, user, clientCerts, "")
+	pgURL, cleanupFn, err = pgurlutils.PGUrlWithOptionalClientCertsE(sqlAddr, prefix, user, clientCerts, certName)
 	if err != nil {
 		return pgURL, cleanupFn, err
 	}
@@ -95,9 +96,10 @@ func openTestSQLConn(
 	insecure bool,
 	clientCerts bool,
 	prefix string,
+	certName string,
 ) (*gosql.DB, error) {
 	var goDB *gosql.DB
-	u, cleanupFn, err := pgURL(dbName, user, tenantName, sqlAddr, insecure, clientCerts, prefix)
+	u, cleanupFn, err := pgURL(dbName, user, tenantName, sqlAddr, insecure, clientCerts, prefix, certName)
 	if err != nil {
 		return nil, err
 	}

pkg/sql/BUILD.bazel

@@ -942,7 +942,6 @@ go_test(
         "//pkg/testutils/jobutils",
         "//pkg/testutils/kvclientutils",
         "//pkg/testutils/pgtest",
-        "//pkg/testutils/pgurlutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",

pkg/sql/catalog/bootstrap/splits_test.go

@@ -51,6 +51,9 @@ func TestInitialSplitPoints(t *testing.T) {
 
 		ctx := context.Background()
 		tc := testcluster.StartTestCluster(t, 1, base.TestClusterArgs{
+			ServerArgs: base.TestServerArgs{
+				DefaultTestTenant: base.TestControlsTenantsExplicitly,
+			},
 			ReplicationMode: base.ReplicationManual,
 		})
 		defer tc.Stopper().Stop(ctx)

pkg/sql/catalog/externalcatalog/external_catalog_test.go

@@ -117,7 +117,7 @@ func TestExtractIngestExternalCatalog(t *testing.T) {
 		sqlDB.CheckQueryResults(t, "SELECT schema_name,table_name FROM [SHOW TABLES]", [][]string{{"public", "tab1"}, {"public", "tab2_rename"}})
 
 		require.NoError(t, sqltestutils.TestingDescsTxn(ctx, srv, func(ctx context.Context, txn isql.Txn, col *descs.Collection) error {
-			return DropIngestedExternalCatalog(ctx, &execCfg, sqlUser, written, txn, srv.JobRegistry().(*jobs.Registry), col, "test gc")
+			return DropIngestedExternalCatalog(ctx, &execCfg, sqlUser, written, txn, s.JobRegistry().(*jobs.Registry), col, "test gc")
 		}))
 		var res int
 		sqlDB.QueryRow(t, "SELECT count(*) FROM [SHOW TABLES]").Scan(&res)

pkg/sql/catalog/lease/lease_internal_test.go

@@ -1786,6 +1786,7 @@ func TestLeaseManagerLockedTimestampBasic(t *testing.T) {
 			Settings: st,
 		})
 	defer srv.Stopper().Stop(context.Background())
+	s := srv.ApplicationLayer()
 
 	r := sqlutils.MakeSQLRunner(db)
 
@@ -1802,7 +1803,7 @@ func TestLeaseManagerLockedTimestampBasic(t *testing.T) {
 	}()
 	var id int
 	r.QueryRow(t, "SELECT 't1'::REGCLASS::OID;").Scan(&id)
-	lm := srv.LeaseManager().(*Manager)
+	lm := s.LeaseManager().(*Manager)
 
 	// Note: We only need to hold old leases because the logic for doing this
 	// automatically is not merged yet. Once it is merged, the release of them
@@ -1817,7 +1818,7 @@ func TestLeaseManagerLockedTimestampBasic(t *testing.T) {
 	defer releaseHeldDescriptors()
 
 	getDescriptorVersion := func() descpb.DescriptorVersion {
-		ts := lm.GetReadTimestamp(srv.Clock().Now())
+		ts := lm.GetReadTimestamp(s.Clock().Now())
 		state := lm.findDescriptorState(descpb.ID(id), false)
 		require.NotNilf(t, state, "the descriptor was not leased yet")
 		ld, _, err := state.findForTimestamp(ctx, ts)
@@ -1869,12 +1870,12 @@ func TestLeaseManagerLockedTimestampCluster(t *testing.T) {
 	tc := serverutils.StartCluster(
 		t, 3, base.TestClusterArgs{
 			ServerArgs: base.TestServerArgs{
-				Settings:          st,
+				Settings: st,
+				// Avoid using tenants since async tenant migration steps can acquire
+				// leases on our user tables.
 				DefaultTestTenant: base.TestNeedsTightIntegrationBetweenAPIsAndTestingKnobs,
 			},
 			ServerArgsPerNode: map[int]base.TestServerArgs{2: {
-				// Avoid using tenants since async tenant migration steps can acquire
-				// leases on our user tables.
 				Knobs: base.TestingKnobs{
 					SQLLeaseManager: &ManagerTestingKnobs{
 						TestingDescriptorRefreshedEvent: func(descriptor *descpb.Descriptor) {
@@ -1907,8 +1908,6 @@ func TestLeaseManagerLockedTimestampCluster(t *testing.T) {
 		_, err := node1Conn.Exec("ALTER TABLE d1.public.t1 ADD COLUMN n2 int DEFAULT 364")
 		return err
 	})
-	go func() {
-	}()
 	lm := tc.Server(2).LeaseManager().(*Manager)
 	assertDescriptorsCount := func(expectedCount int) {
 		state := lm.findDescriptorState(descpb.ID(id), false)

pkg/sql/catalog/lease/lease_test.go

@@ -1828,12 +1828,13 @@ func TestDeleteOrphanedLeasesBySession(t *testing.T) {
 	defer log.Scope(t).Close(t)
 
 	ctx := context.Background()
-	s, conn, _ := serverutils.StartServer(t, base.TestServerArgs{
+	srv, conn, _ := serverutils.StartServer(t, base.TestServerArgs{
 		Locality: roachpb.Locality{
 			Tiers: []roachpb.Tier{{Key: "region", Value: "us-east1"}},
 		},
 	})
-	defer s.Stop(ctx)
+	defer srv.Stop(ctx)
+	s := srv.ApplicationLayer()
 	idb := s.InternalDB().(*sql.InternalDB)
 	ie := idb.Executor()
 	// Validate only one session exists.
@@ -2841,11 +2842,12 @@ func TestLeaseTxnDeadlineExtensionWithSession(t *testing.T) {
 		},
 	}
 
-	s, sqlDB, _ := serverutils.StartServer(t, params)
+	srv, sqlDB, _ := serverutils.StartServer(t, params)
+	defer srv.Stopper().Stop(ctx)
+	s := srv.ApplicationLayer()
 	lm := s.LeaseManager().(*lease.Manager)
 	leasesWaitingToExpireGauge := lm.TestingSessionBasedLeasesWaitingToExpireGauge()
 	leasesExpiredGauge := lm.TestingSessionBasedLeasesExpiredGauge()
-	defer s.Stopper().Stop(ctx)
 	// Setup tables for the test.
 	_, err := sqlDB.Exec(`
 CREATE TABLE t1(val int);
@@ -3501,6 +3503,7 @@ func TestLongLeaseWaitMetrics(t *testing.T) {
 		},
 	})
 	defer srv.Stopper().Stop(ctx)
+	s := srv.ApplicationLayer()
 	runner := sqlutils.MakeSQLRunner(sqlDB)
 	runner.Exec(t, "CREATE TABLE t1(n int)")
 	descIDRow := runner.QueryRow(t, "SELECT 't1'::REGCLASS::INT")
@@ -3509,7 +3512,7 @@ func TestLongLeaseWaitMetrics(t *testing.T) {
 	grp := ctxgroup.WithContext(ctx)
 
 	startWaiters := make(chan struct{})
-	cachedDatabaseRegions, err := regions.NewCachedDatabaseRegions(ctx, srv.DB(), srv.LeaseManager().(*lease.Manager))
+	cachedDatabaseRegions, err := regions.NewCachedDatabaseRegions(ctx, s.DB(), s.LeaseManager().(*lease.Manager))
 	require.NoError(t, err)
 
 	grp.GoCtx(func(ctx context.Context) error {
@@ -3529,13 +3532,13 @@ func TestLongLeaseWaitMetrics(t *testing.T) {
 		r := retry.StartWithCtx(ctx, retry.Options{})
 		// Wait until long waits are detected in our metrics.
 		for r.Next() {
-			if srv.MustGetSQLCounter("sql.leases.long_wait_for_no_version") == 0 {
+			if s.MustGetSQLCounter("sql.leases.long_wait_for_no_version") == 0 {
 				continue
 			}
-			if srv.MustGetSQLCounter("sql.leases.long_wait_for_two_version_invariant") == 0 {
+			if s.MustGetSQLCounter("sql.leases.long_wait_for_two_version_invariant") == 0 {
 				continue
 			}
-			if srv.MustGetSQLCounter("sql.leases.long_wait_for_one_version") == 0 {
+			if s.MustGetSQLCounter("sql.leases.long_wait_for_one_version") == 0 {
 				continue
 			}
 			break
@@ -3563,11 +3566,11 @@ func TestLongLeaseWaitMetrics(t *testing.T) {
 		r := retry.StartWithCtx(ctx, retry.Options{})
 		for r.Next() {
 			// Wait for the two versions to exist.
-			if srv.MustGetSQLCounter("sql.leases.long_wait_for_two_version_invariant") == 0 {
+			if s.MustGetSQLCounter("sql.leases.long_wait_for_two_version_invariant") == 0 {
 				continue
 			}
 			// Wait for there to be a single version.
-			lm := srv.ApplicationLayer().LeaseManager().(*lease.Manager)
+			lm := s.LeaseManager().(*lease.Manager)
 			_, err := lm.WaitForOneVersion(ctx, descpb.ID(descID), cachedDatabaseRegions, retry.Options{})
 			return err
 		}
@@ -3577,16 +3580,16 @@ func TestLongLeaseWaitMetrics(t *testing.T) {
 	// Waits for no version of the descriptor to exist.
 	grp.GoCtx(func(ctx context.Context) error {
 		<-startWaiters
-		lm := srv.ApplicationLayer().LeaseManager().(*lease.Manager)
+		lm := s.LeaseManager().(*lease.Manager)
 		return lm.WaitForNoVersion(ctx, descpb.ID(descID), cachedDatabaseRegions, retry.Options{})
 	})
 
 	require.NoError(t, grp.Wait())
 
 	// Validate the metrics are 0 again.
-	require.Equal(t, int64(0), srv.MustGetSQLCounter("sql.leases.long_wait_for_no_version"))
-	require.Equal(t, int64(0), srv.MustGetSQLCounter("sql.leases.long_wait_for_two_version_invariant"))
-	require.Equal(t, int64(0), srv.MustGetSQLCounter("sql.leases.long_wait_for_one_version"))
+	require.Equal(t, int64(0), s.MustGetSQLCounter("sql.leases.long_wait_for_no_version"))
+	require.Equal(t, int64(0), s.MustGetSQLCounter("sql.leases.long_wait_for_two_version_invariant"))
+	require.Equal(t, int64(0), s.MustGetSQLCounter("sql.leases.long_wait_for_one_version"))
 }
 
 // TestWaitForInitialVersionConcurrent this test is a basic sanity test that
@@ -3672,8 +3675,9 @@ func BenchmarkAcquireLeaseConcurrent(b *testing.B) {
 	defer log.Scope(b).Close(b)
 
 	ctx := context.Background()
-	s, sqlDB, _ := serverutils.StartServer(b, base.TestServerArgs{})
-	defer s.Stopper().Stop(ctx)
+	srv, sqlDB, _ := serverutils.StartServer(b, base.TestServerArgs{})
+	defer srv.Stopper().Stop(ctx)
+	s := srv.ApplicationLayer()
 
 	runner := sqlutils.MakeSQLRunner(sqlDB)
 
@@ -3710,8 +3714,9 @@ func TestLeaseManagerIsMemoryMonitored(t *testing.T) {
 	skip.UnderDuress(t)
 
 	ctx := context.Background()
-	s, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{})
-	defer s.Stopper().Stop(ctx)
+	srv, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer srv.Stopper().Stop(ctx)
+	s := srv.ApplicationLayer()
 	lm := s.LeaseManager().(*lease.Manager)
 	startBytes := lm.TestingGetBoundAccount().Used()
 	lastBytes := startBytes

pkg/sql/catalog/schematelemetry/schema_telemetry_test.go

@@ -97,8 +97,9 @@ func TestSchemaTelemetryJob(t *testing.T) {
 	defer log.Scope(t).Close(t)
 
 	ctx := context.Background()
-	s, db, _ := serverutils.StartServer(t, makeTestServerArgs())
-	defer s.Stopper().Stop(ctx)
+	srv, db, _ := serverutils.StartServer(t, makeTestServerArgs())
+	defer srv.Stopper().Stop(ctx)
+	s := srv.ApplicationLayer()
 	tdb := sqlutils.MakeSQLRunner(db)
 	tdb.Exec(t, `SET CLUSTER SETTING server.eventlog.enabled = true`)