Merge #144918 #148537 #148604 #148609

144918: sql, opt: add UpdateSwap and DeleteSwap to exec factory  r=DrewKimball a=michae2

This set of commits adds UpdateSwap and DeleteSwap constructors to the exec factory. The next PR will actually call these constructor functions during execbuild.

See individual commits for details.

Informs: #144503

Release note: None

148537: jobs: remove deprecated custom per-job auth r=asg0451 a=dt

This was deprecated a couple releases ago in favor of using more standard approaches of either a) telling users to grant or revoke membership in a role that has ownership of a job to manage access to that job or b) features built on top of jobs can have their own control statements that perform their own auth checks before modifying or creating jobs if they so choose.

With this having been deprecated for a couple of major releases now with the public docs suggesting using role membership instead, it can now be deleted to simplify the auth checks the jobs system needs to perform, paving the way for replacing the complex logic in the vtable for SHOW JOBS with a simple view instead.

Release note (ops change): Non-admin users no longer have access to changefeed jobs they do not own and which are not owned by a role of which they are a member, regardless of whether they have the changefeed privilege on the table or tables those jobs may be watching.
Epic: CRDB-48791.

148604: roachtest: update wal-failover/among-stores/with-progress to use process monitor r=xinhaoz a=xinhaoz

Epic: none

148609: rpc: remove DRPC dependency from `Peer` and `Connection` generics r=cthumuluru-crdb a=cthumuluru-crdb

Although `Peer` and `Connection` are generic and intended to support both gRPC and DRPC connections, the current implementation has a hardcoded dependency on DRPC. As a result, a DRPC connection is always dialed, regardless of the intended type. This PR removes the direct dependency on DRPC, allowing the appropriate connection type (gRPC or DRPC) to be used based on the generic parameter.

Epic: CRDB-48923
Fixes: none
Release note: none

Co-authored-by: Michael Erickson <[email protected]>
Co-authored-by: David Taylor <[email protected]>
Co-authored-by: Xin Hao Zhang <[email protected]>
Co-authored-by: Chandra Thumuluru <[email protected]>

Author: 5 people

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -113,11 +113,8 @@ func alterChangefeedPlanHook(
 		if err != nil {
 			return err
 		}
-		getLegacyPayload := func(ctx context.Context) (*jobspb.Payload, error) {
-			return &jobPayload, nil
-		}
-		err = jobsauth.AuthorizeAllowLegacyAuth(
-			ctx, p, jobID, getLegacyPayload, jobPayload.UsernameProto.Decode(), jobPayload.Type(), jobsauth.ControlAccess, globalPrivileges,
+		err = jobsauth.Authorize(
+			ctx, p, jobID, jobPayload.UsernameProto.Decode(), jobsauth.ControlAccess, globalPrivileges,
 		)
 		if err != nil {
 			return err

pkg/ccl/changefeedccl/alter_changefeed_test.go

@@ -82,6 +82,8 @@ func TestAlterChangefeedAddTargetPrivileges(t *testing.T) {
 		`CREATE TABLE table_b (id int, type type_a)`,
 		`CREATE TABLE table_c (id int, type type_a)`,
 		`CREATE USER feedCreator`,
+		`CREATE ROLE feedowner`,
+		`GRANT feedowner TO feedCreator`,
 		`GRANT SELECT ON table_a TO feedCreator`,
 		`GRANT CHANGEFEED ON table_a TO feedCreator`,
 		`CREATE EXTERNAL CONNECTION "first" AS 'kafka://nope'`,
@@ -123,22 +125,24 @@ func TestAlterChangefeedAddTargetPrivileges(t *testing.T) {
 			row.Scan(&jobID)
 			userDB.Exec(t, `PAUSE JOB $1`, jobID)
 			waitForJobState(userDB, t, catpb.JobID(jobID), `paused`)
+			userDB.Exec(t, `ALTER JOB $1 OWNER TO feedowner`, jobID)
 		})
 
 		// user1 is missing the CHANGEFEED privilege on table_b and table_c.
 		withUser(t, "user1", func(userDB *sqlutils.SQLRunner) {
 			userDB.ExpectErr(t,
-				"user user1 requires the CHANGEFEED privilege on all target tables to be able to run an enterprise changefeed",
+				"user user1 does not have privileges for job",
 				fmt.Sprintf("ALTER CHANGEFEED %d ADD table_b, table_c set sink='external://second'", jobID),
 			)
 		})
 		rootDB.Exec(t, `GRANT CHANGEFEED ON table_b TO user1`)
 		withUser(t, "user1", func(userDB *sqlutils.SQLRunner) {
 			userDB.ExpectErr(t,
-				"user user1 requires the CHANGEFEED privilege on all target tables to be able to run an enterprise changefeed",
+				"user user1 does not have privileges for job",
 				fmt.Sprintf("ALTER CHANGEFEED %d ADD table_b, table_c set sink='external://second'", jobID),
 			)
 		})
+		rootDB.Exec(t, `GRANT feedowner TO user1`)
 		rootDB.Exec(t, `GRANT CHANGEFEED ON table_c TO user1`)
 		withUser(t, "user1", func(userDB *sqlutils.SQLRunner) {
 			userDB.Exec(t,
@@ -175,19 +179,27 @@ func TestAlterChangefeedAddTargetPrivileges(t *testing.T) {
 			row.Scan(&jobID)
 			userDB.Exec(t, `PAUSE JOB $1`, jobID)
 			waitForJobState(userDB, t, catpb.JobID(jobID), `paused`)
+			userDB.Exec(t, `ALTER JOB $1 OWNER TO feedowner`, jobID)
 		})
 
 		// user2 is missing the SELECT privilege on table_b and table_c.
 		withUser(t, "user2", func(userDB *sqlutils.SQLRunner) {
 			userDB.ExpectErr(t,
-				"pq: user user2 with CONTROLCHANGEFEED role option requires the SELECT privilege on all target tables to be able to run an enterprise changefeed",
+				"pq: user user2 does not have privileges for job",
 				fmt.Sprintf("ALTER CHANGEFEED %d ADD table_b, table_c set sink='kafka://bar'", jobID),
 			)
 		})
 		rootDB.Exec(t, `GRANT SELECT ON table_b TO user2`)
 		withUser(t, "user2", func(userDB *sqlutils.SQLRunner) {
 			userDB.ExpectErr(t,
-				"pq: user user2 with CONTROLCHANGEFEED role option requires the SELECT privilege on all target tables to be able to run an enterprise changefeed",
+				"pq: user user2 does not have privileges for job",
+				fmt.Sprintf("ALTER CHANGEFEED %d ADD table_b, table_c set sink='kafka://bar'", jobID),
+			)
+		})
+		rootDB.Exec(t, `GRANT feedowner TO user2`)
+		withUser(t, "user2", func(userDB *sqlutils.SQLRunner) {
+			userDB.ExpectErr(t,
+				"requires the SELECT privilege on all target tables",
 				fmt.Sprintf("ALTER CHANGEFEED %d ADD table_b, table_c set sink='kafka://bar'", jobID),
 			)
 		})
@@ -1859,6 +1871,7 @@ func TestAlterChangefeedAccessControl(t *testing.T) {
 		})
 		rootDB.Exec(t, "PAUSE job $1", currentFeed.JobID())
 		waitForJobState(rootDB, t, currentFeed.JobID(), `paused`)
+		rootDB.Exec(t, "ALTER JOB $1 OWNER TO feedowner", currentFeed.JobID())
 
 		// Verify who can modify the existing changefeed.
 		asUser(t, f, `userWithAllGrants`, func(userDB *sqlutils.SQLRunner) {
@@ -1872,10 +1885,10 @@ func TestAlterChangefeedAccessControl(t *testing.T) {
 			userDB.ExpectErr(t, "pq: user jobcontroller requires the CHANGEFEED privilege on all target tables to be able to run an enterprise changefeed", fmt.Sprintf(`ALTER CHANGEFEED %d DROP table_b`, currentFeed.JobID()))
 		})
 		asUser(t, f, `userWithSomeGrants`, func(userDB *sqlutils.SQLRunner) {
-			userDB.ExpectErr(t, "pq: user userwithsomegrants does not have CHANGEFEED privilege on relation table_b", fmt.Sprintf(`ALTER CHANGEFEED %d ADD table_b`, currentFeed.JobID()))
+			userDB.ExpectErr(t, "does not have privileges for job", fmt.Sprintf(`ALTER CHANGEFEED %d ADD table_b`, currentFeed.JobID()))
 		})
 		asUser(t, f, `regularUser`, func(userDB *sqlutils.SQLRunner) {
-			userDB.ExpectErr(t, "pq: user regularuser does not have CHANGEFEED privilege on relation (table_a|table_b)", fmt.Sprintf(`ALTER CHANGEFEED %d ADD table_b`, currentFeed.JobID()))
+			userDB.ExpectErr(t, "does not have privileges for job", fmt.Sprintf(`ALTER CHANGEFEED %d ADD table_b`, currentFeed.JobID()))
 		})
 		closeCf()
 

pkg/ccl/changefeedccl/authorization.go

@@ -11,8 +11,6 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/cloud/externalconn"
-	"github.com/cockroachdb/cockroach/pkg/jobs/jobsauth"
-	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
@@ -135,45 +133,3 @@ func authorizeUserToCreateChangefeed(
 
 	return nil
 }
-
-// AuthorizeChangefeedJobAccess determines if a user has access to the changefeed job denoted
-// by the supplied jobID and payload.
-func AuthorizeChangefeedJobAccess(
-	ctx context.Context,
-	a jobsauth.AuthorizationAccessor,
-	jobID jobspb.JobID,
-	getLegacyPayload func(ctx context.Context) (*jobspb.Payload, error),
-) error {
-	payload, err := getLegacyPayload(ctx)
-	if err != nil {
-		return err
-	}
-	specs, ok := payload.UnwrapDetails().(jobspb.ChangefeedDetails)
-	if !ok {
-		return errors.Newf("could not unwrap details from the payload of job %d", jobID)
-	}
-
-	if len(specs.TargetSpecifications) == 0 {
-		return pgerror.Newf(pgcode.InsufficientPrivilege, "job contains no tables on which the user has %s privilege", privilege.CHANGEFEED)
-	}
-
-	for _, spec := range specs.TargetSpecifications {
-		err := a.CheckPrivilegeForTableID(ctx, spec.TableID, privilege.CHANGEFEED)
-		if err != nil {
-			// When performing SHOW JOBS or SHOW CHANGEFEED JOBS, there may be old changefeed
-			// records that reference tables which have been dropped or are being
-			// dropped. In this case, we would prefer to skip the permissions check on
-			// the dropped descriptor.
-			if pgerror.GetPGCode(err) == pgcode.UndefinedTable || errors.Is(err, catalog.ErrDescriptorDropped) {
-				continue
-			}
-
-			return err
-		}
-	}
-	return nil
-}
-
-func init() {
-	jobsauth.RegisterAuthorizer(jobspb.TypeChangefeed, AuthorizeChangefeedJobAccess)
-}

pkg/ccl/changefeedccl/changefeed_test.go

@@ -3749,6 +3749,7 @@ func TestChangefeedJobControl(t *testing.T) {
 		asUser(t, f, `adminUser`, func(userDB *sqlutils.SQLRunner) {
 			userDB.Exec(t, "PAUSE job $1", currentFeed.JobID())
 			waitForJobState(userDB, t, currentFeed.JobID(), "paused")
+			userDB.Exec(t, "ALTER JOB $1 OWNER TO feedowner", currentFeed.JobID())
 		})
 		asUser(t, f, `userWithAllGrants`, func(userDB *sqlutils.SQLRunner) {
 			userDB.Exec(t, "RESUME job $1", currentFeed.JobID())
@@ -3759,10 +3760,10 @@ func TestChangefeedJobControl(t *testing.T) {
 			waitForJobState(userDB, t, currentFeed.JobID(), "running")
 		})
 		asUser(t, f, `userWithSomeGrants`, func(userDB *sqlutils.SQLRunner) {
-			userDB.ExpectErr(t, "user userwithsomegrants does not have CHANGEFEED privilege on relation table_b", "PAUSE job $1", currentFeed.JobID())
+			userDB.ExpectErr(t, "does not have privileges for job", "PAUSE job $1", currentFeed.JobID())
 		})
 		asUser(t, f, `regularUser`, func(userDB *sqlutils.SQLRunner) {
-			userDB.ExpectErr(t, "user regularuser does not have CHANGEFEED privilege on relation (table_a|table_b)", "PAUSE job $1", currentFeed.JobID())
+			userDB.ExpectErr(t, "does not have privileges for job", "PAUSE job $1", currentFeed.JobID())
 		})
 		closeCf()
 

pkg/ccl/changefeedccl/helpers_test.go

@@ -1736,6 +1736,7 @@ func ChangefeedJobPermissionsTestSetup(t *testing.T, s TestServer) {
 
 		`CREATE USER adminUser`,
 		`GRANT ADMIN TO adminUser`,
+		`CREATE ROLE feedowner`,
 
 		`CREATE USER otherAdminUser`,
 		`GRANT ADMIN TO otherAdminUser`,
@@ -1747,6 +1748,7 @@ func ChangefeedJobPermissionsTestSetup(t *testing.T, s TestServer) {
 		`CREATE USER jobController with CONTROLJOB`,
 
 		`CREATE USER userWithAllGrants`,
+		`GRANT feedowner TO userWithAllGrants`,
 		`GRANT CHANGEFEED ON table_a TO userWithAllGrants`,
 		`GRANT CHANGEFEED ON table_b TO userWithAllGrants`,
 

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go

@@ -568,7 +568,6 @@ func TestShowChangefeedJobsAuthorization(t *testing.T) {
 			require.NoError(t, err)
 			jobID = successfulFeed.(cdctest.EnterpriseTestFeed).JobID()
 		}
-		rootDB := sqlutils.MakeSQLRunner(s.DB)
 
 		// Create a changefeed and assert who can see it.
 		asUser(t, f, `feedCreator`, func(userDB *sqlutils.SQLRunner) {
@@ -579,7 +578,7 @@ func TestShowChangefeedJobsAuthorization(t *testing.T) {
 			userDB.CheckQueryResults(t, `SELECT job_id FROM [SHOW CHANGEFEED JOBS]`, [][]string{{expectedJobIDStr}})
 		})
 		asUser(t, f, `userWithAllGrants`, func(userDB *sqlutils.SQLRunner) {
-			userDB.CheckQueryResults(t, `SELECT job_id FROM [SHOW CHANGEFEED JOBS]`, [][]string{{expectedJobIDStr}})
+			userDB.CheckQueryResults(t, `SELECT job_id FROM [SHOW CHANGEFEED JOBS]`, [][]string{})
 		})
 		asUser(t, f, `userWithSomeGrants`, func(userDB *sqlutils.SQLRunner) {
 			userDB.CheckQueryResults(t, `SELECT job_id FROM [SHOW CHANGEFEED JOBS]`, [][]string{})
@@ -590,16 +589,6 @@ func TestShowChangefeedJobsAuthorization(t *testing.T) {
 		asUser(t, f, `regularUser`, func(userDB *sqlutils.SQLRunner) {
 			userDB.CheckQueryResults(t, `SELECT job_id FROM [SHOW CHANGEFEED JOBS]`, [][]string{})
 		})
-
-		// Assert behavior when one of the tables is dropped.
-		rootDB.Exec(t, "DROP TABLE table_b")
-		// Having CHANGEFEED on only table_a is now sufficient.
-		asUser(t, f, `userWithSomeGrants`, func(userDB *sqlutils.SQLRunner) {
-			userDB.CheckQueryResults(t, `SELECT job_id FROM [SHOW CHANGEFEED JOBS]`, [][]string{{expectedJobIDStr}})
-		})
-		asUser(t, f, `regularUser`, func(userDB *sqlutils.SQLRunner) {
-			userDB.CheckQueryResults(t, `SELECT job_id FROM [SHOW CHANGEFEED JOBS]`, [][]string{})
-		})
 	}
 
 	// Only enterprise sinks create jobs.

pkg/cmd/roachtest/tests/disk_stall.go

@@ -11,17 +11,18 @@ import (
 	"math/rand"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/cluster"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/clusterstats"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/option"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/registry"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil"
+	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil/task"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/spec"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/test"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/install"
+	"github.com/cockroachdb/cockroach/pkg/roachprod/logger"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/prometheus"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/stretchr/testify/require"
@@ -435,6 +436,7 @@ func registerDiskStalledWALFailoverWithProgress(r registry.Registry) {
 		SkipPostValidations: registry.PostValidationNoDeadNodes,
 		EncryptionSupport:   registry.EncryptionMetamorphic,
 		Leases:              registry.MetamorphicLeases,
+		Monitor:             true,
 		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
 			runDiskStalledWALFailoverWithProgress(ctx, t, c)
 		},
@@ -484,7 +486,7 @@ func runDiskStalledWALFailoverWithProgress(ctx context.Context, t test.Test, c c
 
 	t.Status("starting oscillating workload and disk stall pattern")
 	testStartedAt := timeutil.Now()
-	m := c.NewMonitor(ctx, c.CRDBNodes())
+	g := t.NewGroup(task.WithContext(ctx))
 
 	// Setup stats collector.
 	promCfg := &prometheus.Config{}
@@ -511,7 +513,6 @@ func runDiskStalledWALFailoverWithProgress(ctx context.Context, t test.Test, c c
 	for timeutil.Since(testStartedAt) < testDuration {
 		if t.Failed() {
 			t.Fatalf("test failed, stopping further iterations")
-			return
 		}
 
 		workloadWaitDur := operationWaitBase + time.Duration(rand.Int63n(int64(waitJitterMax)))
@@ -521,11 +522,7 @@ func runDiskStalledWALFailoverWithProgress(ctx context.Context, t test.Test, c c
 		workloadStarted := make(chan struct{})
 		workloadFinished := make(chan struct{})
 
-		var wg sync.WaitGroup
-		wg.Add(1)
-		m.Go(func(ctx context.Context) error {
-			defer wg.Done()
-
+		g.Go(func(ctx context.Context, _ *logger.Logger) error {
 			select {
 			case <-ctx.Done():
 				t.Fatalf("context done before workload started: %s", ctx.Err())
@@ -540,14 +537,12 @@ func runDiskStalledWALFailoverWithProgress(ctx context.Context, t test.Test, c c
 				return nil
 			}
 			return nil
-		})
+		}, task.Name("workload-run"))
 
 		// Collecting QPS samples while the workload is running and verify
 		// that the throughput is within errorTolerance of the mean.
 		var samples []float64
-		wg.Add(1)
-		m.Go(func(ctx context.Context) error {
-			defer wg.Done()
+		g.Go(func(ctx context.Context, _ *logger.Logger) error {
 
 			// Wait for workload to start.
 			select {
@@ -602,17 +597,15 @@ func runDiskStalledWALFailoverWithProgress(ctx context.Context, t test.Test, c c
 
 			t.Status(fmt.Sprintf("workload finished, %d samples collected", len(samples)))
 			return nil
-		})
+		}, task.Name("qps-sampling"))
 
 		// Every 4th iteration, we'll skip the disk stall phase.
 		if iteration%4 != 0 {
 			// Calculate next stall phase with jitter.
 			diskStallWaitDur := operationWaitBase + time.Duration(rand.Int63n(int64(waitJitterMax)))
 			t.Status("next stall phase in ", diskStallWaitDur)
 
-			wg.Add(1)
-			m.Go(func(ctx context.Context) error {
-				defer wg.Done()
+			g.Go(func(ctx context.Context, _ *logger.Logger) error {
 				select {
 				case <-ctx.Done():
 					t.Fatalf("context done before stall started: %s", ctx.Err())
@@ -650,13 +643,13 @@ func runDiskStalledWALFailoverWithProgress(ctx context.Context, t test.Test, c c
 				}
 
 				return nil
-			})
+			}, task.Name("disk-stall-phase"))
 		} else {
 			t.Status("skipping disk stall phase for this iteration")
 		}
 
 		// Wait for all goroutines to complete.
-		wg.Wait()
+		g.Wait()
 
 		// Validate throughput samples are within tolerance.
 		meanThroughput := roachtestutil.GetMeanOverLastN(len(samples), samples)
@@ -698,12 +691,6 @@ func runDiskStalledWALFailoverWithProgress(ctx context.Context, t test.Test, c c
 	if durInFailover < 10*time.Minute {
 		t.Errorf("expected s1 to spend at least 10m writing to secondary, but spent %s", durInFailover)
 	}
-
-	// Wait for the workload to finish (if it hasn't already).
-	m.Wait()
-
-	// Shut down the nodes, allowing any devices to be unmounted during cleanup.
-	c.Stop(ctx, t.L(), option.DefaultStopOpts(), c.CRDBNodes())
 }
 
 func getProcessStartMonotonic(

pkg/crosscluster/producer/replication_manager.go

@@ -442,7 +442,7 @@ func (r *replicationStreamManagerImpl) AuthorizeViaJob(
 	}
 
 	if err := jobsauth.Authorize(
-		ctx, planHook, jobspb.JobID(streamID), planHook.User(), jobspb.TypeReplicationStreamProducer, jobsauth.ControlAccess, globalPrivileges,
+		ctx, planHook, jobspb.JobID(streamID), planHook.User(), jobsauth.ControlAccess, globalPrivileges,
 	); err != nil {
 		return err
 	}

pkg/jobs/jobsauth/BUILD.bazel

@@ -13,7 +13,6 @@ go_library(
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/privilege",
         "//pkg/sql/roleoption",
-        "@com_github_cockroachdb_errors//:errors",
     ],
 )