kv/rangefeed: filter and test internal rangefeed registry

Filter and test internal rangefeed registry.

Epic: None
Release note: None

Author: asg0451

pkg/ccl/changefeedccl/BUILD.bazel

@@ -324,6 +324,7 @@ go_test(
         "//pkg/util/intsets",
         "//pkg/util/ioctx",
         "//pkg/util/json",
+        "//pkg/util/keysutil",
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "//pkg/util/log/eventpb",

pkg/ccl/changefeedccl/changefeed_test.go

@@ -91,6 +91,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/json"
+	"github.com/cockroachdb/cockroach/pkg/util/keysutil"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
@@ -9128,22 +9129,66 @@ func TestDistSenderRangeFeedPopulatesVirtualTable(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	s, cleanup := makeServer(t)
-	defer cleanup()
+	scanner := keysutil.MakePrettyScanner(nil, nil)
 
-	sqlDB := sqlutils.MakeSQLRunner(s.DB)
-	sqlDB.Exec(t, `CREATE TABLE tbl (a INT, b STRING);`)
-	sqlDB.Exec(t, `INSERT INTO tbl VALUES (1, 'one'), (2, 'two'), (3, 'three');`)
-	sqlDB.Exec(t, `CREATE CHANGEFEED FOR tbl INTO 'null://';`)
-
-	var tableID int
-	sqlDB.QueryRow(t, "SELECT table_id FROM crdb_internal.tables WHERE name='tbl'").Scan(&tableID)
-	tableKey := s.Codec.TablePrefix(uint32(tableID))
-
-	numRangesQuery := fmt.Sprintf(
-		"SELECT count(*) FROM crdb_internal.active_range_feeds WHERE range_start LIKE '%s/%%'",
-		tableKey)
-	sqlDB.CheckQueryResultsRetry(t, numRangesQuery, [][]string{{"1"}})
+	observeTables := func(sqlDB *sqlutils.SQLRunner, codec keys.SQLCodec) []int {
+		rows := sqlDB.Query(t, "SELECT range_start FROM crdb_internal.active_range_feeds")
+		defer rows.Close()
+		var tableIDs []int
+		for rows.Next() {
+			var prettyKey string
+			require.NoError(t, rows.Scan(&prettyKey))
+			key, err := scanner.Scan(prettyKey)
+			require.NoError(t, err)
+			_, tableID, err := codec.DecodeTablePrefix(key)
+			require.NoError(t, err)
+			tableIDs = append(tableIDs, int(tableID))
+		}
+		return tableIDs
+	}
+
+	cases := []struct {
+		user           string
+		shouldSeeTable bool
+	}{
+		{`feedCreator`, false},
+		{`regularUser`, false},
+		{`adminUser`, true},
+		{`viewClusterMetadataUser`, true},
+	}
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+
+		// Creates several different tables, users, and roles for us to use.
+		ChangefeedJobPermissionsTestSetup(t, s)
+
+		var tableID int
+		sqlDB.QueryRow(t, "SELECT table_id FROM crdb_internal.tables WHERE name = 'table_a'").Scan(&tableID)
+
+		var cf cdctest.TestFeed
+		asUser(t, f, `feedCreator`, func(userDB *sqlutils.SQLRunner) {
+			cf = feed(t, f, `CREATE CHANGEFEED FOR table_a;`)
+		})
+		defer closeFeed(t, cf)
+
+		for _, c := range cases {
+			testutils.SucceedsSoon(t, func() error {
+				asUser(t, f, c.user, func(userDB *sqlutils.SQLRunner) {
+					tableIDs := observeTables(userDB, s.Codec)
+					if c.shouldSeeTable {
+						require.Containsf(t, tableIDs, tableID, "user %s should see table %d", c.user, tableID)
+					} else {
+						require.Emptyf(t, tableIDs, "user %s should not see any tables", c.user)
+					}
+				})
+				return nil
+			})
+		}
+
+	}
+
+	cdcTest(t, testFn, feedTestEnterpriseSinks)
 }
 
 func TestChangefeedCaseInsensitiveOpts(t *testing.T) {

pkg/ccl/changefeedccl/helpers_test.go

@@ -1577,6 +1577,9 @@ func ChangefeedJobPermissionsTestSetup(t *testing.T, s TestServer) {
 		`GRANT CHANGEFEED ON table_a TO userWithSomeGrants`,
 
 		`CREATE USER regularUser`,
+
+		`CREATE USER viewClusterMetadataUser`,
+		`GRANT SYSTEM VIEWCLUSTERMETADATA TO viewClusterMetadataUser`,
 	)
 }
 

pkg/sql/crdb_internal.go

@@ -1780,12 +1780,12 @@ var crdbInternalSessionBasedLeases = virtualSchemaView{
 	schema: `
 CREATE VIEW crdb_internal.kv_session_based_leases (
   desc_id,
-  version, 
+  version,
   sql_instance_id,
-  session_id, 
+  session_id,
   crdb_region
 ) AS (
-	SELECT desc_id, version, sql_instance_id, session_id, crdb_region 
+	SELECT desc_id, version, sql_instance_id, session_id, crdb_region
 	FROM system.lease
 );
 `,
@@ -7592,6 +7592,13 @@ CREATE TABLE crdb_internal.active_range_feeds (
   last_err STRING
 );`,
 	populate: func(ctx context.Context, p *planner, _ catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) error {
+		privileged, err := p.HasPrivilege(ctx, syntheticprivilege.GlobalPrivilegeObject, privilege.VIEWCLUSTERMETADATA, p.User())
+		if err != nil {
+			return err
+		}
+		if !privileged {
+			return nil
+		}
 		return p.execCfg.DistSender.ForEachActiveRangeFeed(
 			func(rfCtx kvcoord.RangeFeedContext, rf kvcoord.PartialRangeFeed) error {
 				now := p.EvalContext().GetStmtTimestamp()