Merge #156854 #158011

156854: api: allow admin user to view databases metadata r=jasonlmfong a=jasonlmfong

the admin user currently has no view into all the databases due to its non-existence in the role_members table, this change adds a third condition to the where clause to allow the view

Epic: None
Release: None

-----

api: refactor query logic for get databases metadata

pull the left join condition on role_members out into the where clause

Epic: None
Release: None

158011: *: clarify log/state engine use r=arulajmani a=pav-kv

This PR is one in the series of PRs clarifying the use of `storage.Engine`: whether it is the raft/log or state machine engine.

Here we solidify that:

- The sideloaded entry files are stored in the state engine `FS`. These are to be ingested into the state engine when the corresponding entry is applied.
- Store liveness uses `LogEngine` because it needs a timely sync. More generally (in upcoming PRs), all `Store`-local keys are in the log engine.
- Node liveness compactions in `startPeriodicLivenessCompaction` correctly use the `StateEngine` (works in the MVCC span). Just a drive-by style fixup.

Epic: CRDB-55220

Co-authored-by: Jason Fong <[email protected]>
Co-authored-by: Pavel Kalinnikov <[email protected]>

Author: 3 people

pkg/kv/kvserver/logstore/sideload_disk.go

@@ -17,7 +17,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
-	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/fs"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/errors"
@@ -36,7 +35,7 @@ type DiskSideloadStorage struct {
 	st      *cluster.Settings
 	limiter *rate.Limiter
 	dir     string
-	eng     storage.Engine
+	fs      *fs.Env
 }
 
 func sideloadedPath(baseDir string, rangeID roachpb.RangeID) string {
@@ -57,15 +56,11 @@ func sideloadedPath(baseDir string, rangeID roachpb.RangeID) string {
 // NewDiskSideloadStorage creates a SideloadStorage for a given replica, stored
 // in the specified engine.
 func NewDiskSideloadStorage(
-	st *cluster.Settings,
-	rangeID roachpb.RangeID,
-	baseDir string,
-	limiter *rate.Limiter,
-	eng storage.Engine,
+	st *cluster.Settings, rangeID roachpb.RangeID, baseDir string, limiter *rate.Limiter, fs *fs.Env,
 ) *DiskSideloadStorage {
 	return &DiskSideloadStorage{
 		dir:     sideloadedPath(baseDir, rangeID),
-		eng:     eng,
+		fs:      fs,
 		st:      st,
 		limiter: limiter,
 	}
@@ -86,14 +81,14 @@ func (ss *DiskSideloadStorage) Put(
 	for {
 		// Use 0644 since that's what RocksDB uses:
 		// https://github.com/facebook/rocksdb/blob/56656e12d67d8a63f1e4c4214da9feeec2bd442b/env/env_posix.cc#L171
-		if err := kvserverbase.WriteFileSyncing(ctx, filename, contents, ss.eng.Env(), 0644, ss.st, ss.limiter, fs.PebbleIngestionWriteCategory); err == nil {
+		if err := kvserverbase.WriteFileSyncing(ctx, filename, contents, ss.fs, 0644, ss.st, ss.limiter, fs.PebbleIngestionWriteCategory); err == nil {
 			return nil
 		} else if !oserror.IsNotExist(err) {
 			return err
 		}
 		// Ensure that ss.dir exists. The filename() is placed directly in ss.dir,
 		// so the next loop iteration should succeed.
-		if err := mkdirAllAndSyncParents(ss.eng.Env(), ss.dir, os.ModePerm); err != nil {
+		if err := mkdirAllAndSyncParents(ss.fs, ss.dir, os.ModePerm); err != nil {
 			return err
 		}
 		continue
@@ -102,7 +97,7 @@ func (ss *DiskSideloadStorage) Put(
 
 // Sync implements SideloadStorage.
 func (ss *DiskSideloadStorage) Sync() error {
-	dir, err := ss.eng.Env().OpenDir(ss.dir)
+	dir, err := ss.fs.OpenDir(ss.dir)
 	// The directory can be missing because we did not Put() any entry to it yet,
 	// or it has been removed by TruncateTo() or Clear().
 	//
@@ -127,7 +122,7 @@ func (ss *DiskSideloadStorage) Get(
 	ctx context.Context, index kvpb.RaftIndex, term kvpb.RaftTerm,
 ) ([]byte, error) {
 	filename := ss.filename(ctx, index, term)
-	b, err := fs.ReadFile(ss.eng.Env(), filename)
+	b, err := fs.ReadFile(ss.fs, filename)
 	if oserror.IsNotExist(err) {
 		return nil, errSideloadedFileNotFound
 	}
@@ -155,7 +150,7 @@ func (ss *DiskSideloadStorage) Purge(
 }
 
 func (ss *DiskSideloadStorage) fileSize(filename string) (int64, error) {
-	info, err := ss.eng.Env().Stat(filename)
+	info, err := ss.fs.Stat(filename)
 	if err != nil {
 		if oserror.IsNotExist(err) {
 			return 0, errSideloadedFileNotFound
@@ -170,7 +165,7 @@ func (ss *DiskSideloadStorage) purgeFile(ctx context.Context, filename string) (
 	if err != nil {
 		return 0, err
 	}
-	if err := ss.eng.Env().Remove(filename); err != nil {
+	if err := ss.fs.Remove(filename); err != nil {
 		if oserror.IsNotExist(err) {
 			return 0, errSideloadedFileNotFound
 		}
@@ -181,7 +176,7 @@ func (ss *DiskSideloadStorage) purgeFile(ctx context.Context, filename string) (
 
 // Clear implements SideloadStorage.
 func (ss *DiskSideloadStorage) Clear(_ context.Context) error {
-	return ss.eng.Env().RemoveAll(ss.dir)
+	return ss.fs.RemoveAll(ss.dir)
 }
 
 // TruncateTo implements SideloadStorage.
@@ -207,7 +202,7 @@ func (ss *DiskSideloadStorage) TruncateTo(ctx context.Context, lastIndex kvpb.Ra
 	if deletedAll {
 		// The directory may not exist, or it may exist and have been empty.
 		// Not worth trying to figure out which one, just try to delete.
-		err := ss.eng.Env().Remove(ss.dir)
+		err := ss.fs.Remove(ss.dir)
 		if err != nil && !oserror.IsNotExist(err) {
 			// TODO(pavelkalinnikov): this is possible because deletedAll can be left
 			// true despite existence of files with index < from which are skipped.
@@ -246,7 +241,7 @@ func (ss *DiskSideloadStorage) forEach(
 	ctx context.Context, visit func(index kvpb.RaftIndex, filename string) (bool, error),
 ) error {
 	// TODO(pavelkalinnikov): consider making the List method iterative.
-	matches, err := ss.eng.Env().List(ss.dir)
+	matches, err := ss.fs.List(ss.dir)
 	if oserror.IsNotExist(err) {
 		return nil // nothing to do
 	} else if err != nil {

pkg/kv/kvserver/logstore/sideload_test.go

@@ -86,7 +86,7 @@ func newTestingSideloadStorage(eng storage.Engine) *DiskSideloadStorage {
 	return NewDiskSideloadStorage(
 		cluster.MakeTestingClusterSettings(), 1,
 		filepath.Join(eng.GetAuxiliaryDir(), "fake", "testing", "dir"),
-		rate.NewLimiter(rate.Inf, math.MaxInt64), eng)
+		rate.NewLimiter(rate.Inf, math.MaxInt64), eng.Env())
 }
 
 // TODO(pavelkalinnikov): give these tests a good refactor.
@@ -96,7 +96,7 @@ func testSideloadingSideloadedStorage(t *testing.T, eng storage.Engine) {
 
 	assertExists := func(exists bool) {
 		t.Helper()
-		_, err := ss.eng.Env().Stat(ss.dir)
+		_, err := ss.fs.Stat(ss.dir)
 		if !exists {
 			require.True(t, oserror.IsNotExist(err), err)
 		} else {
@@ -537,7 +537,7 @@ func TestRaftSSTableSideloadingSideload(t *testing.T) {
 			if test.size != stats.SideloadedBytes {
 				t.Fatalf("expected %d sideloadedSize, but found %d", test.size, stats.SideloadedBytes)
 			}
-			actKeys, err := sideloaded.eng.Env().List(sideloaded.Dir())
+			actKeys, err := sideloaded.fs.List(sideloaded.Dir())
 			if oserror.IsNotExist(err) {
 				t.Log("swallowing IsNotExist")
 				err = nil

pkg/kv/kvserver/replica_init.go

@@ -218,7 +218,7 @@ func newUninitializedReplicaWithoutRaftGroup(store *Store, id roachpb.FullReplic
 		// can be ingested to the state machine locally, when being applied.
 		store.StateEngine().GetAuxiliaryDir(),
 		store.limiters.BulkIOWriteRate,
-		store.StateEngine(),
+		store.StateEngine().Env(),
 	)
 	r.logStorage = &replicaLogStorage{
 		ctx:                r.raftCtx,

pkg/kv/kvserver/replica_raftlog_test.go

@@ -102,7 +102,7 @@ func newReplicaLogStorageTest(t *testing.T) *replicaLogStorageTest {
 	st := cluster.MakeTestingClusterSettings()
 	eng := storage.NewDefaultInMemForTesting()
 	sideloaded := logstore.NewDiskSideloadStorage(st, rangeID,
-		eng.GetAuxiliaryDir(), nil /* limiter: unused */, eng)
+		eng.GetAuxiliaryDir(), nil /* limiter: unused */, eng.Env())
 
 	rt.ls = &replicaLogStorage{
 		ctx:     context.Background(),

pkg/kv/kvserver/store.go

@@ -2253,7 +2253,7 @@ func (s *Store) Start(ctx context.Context, stopper *stop.Stopper) error {
 
 	// Create the Store Liveness SupportManager.
 	sm := storeliveness.NewSupportManager(
-		slpb.StoreIdent{NodeID: s.nodeDesc.NodeID, StoreID: s.StoreID()}, s.StateEngine(),
+		slpb.StoreIdent{NodeID: s.nodeDesc.NodeID, StoreID: s.StoreID()}, s.LogEngine(),
 		s.cfg.StoreLiveness.Options, s.cfg.Settings, s.stopper, s.cfg.Clock,
 		s.cfg.StoreLiveness.HeartbeatTicker, s.cfg.StoreLiveness.Transport,
 		s.cfg.StoreLiveness.SupportManagerKnobs(),

pkg/server/api_v2_databases_metadata.go

@@ -465,15 +465,23 @@ func getTableMetadataBaseQuery(userName string) *safesql.Query {
 		FROM system.table_metadata tbm,
 		     (SELECT "sql.stats.automatic_collection.enabled" as auto_stats_enabled 
 		  		FROM [SHOW CLUSTER SETTING sql.stats.automatic_collection.enabled]) csc
-		LEFT JOIN system.role_members rm ON rm.role = 'admin' AND member = $
-		WHERE (rm.role = 'admin' OR tbm.db_name IN (
+		WHERE (
+			$ = 'admin'
+			OR EXISTS (
+				SELECT 1
+				FROM system.role_members rm
+				WHERE rm.member = $
+					AND rm.role = 'admin'
+			)
+			OR tbm.db_name IN (
 	  			SELECT cdp.database_name
 	  			FROM "".crdb_internal.cluster_database_privileges cdp
 	  			WHERE (grantee = $ OR grantee = 'public')
 	  			AND privilege_type = 'CONNECT'
-	  		))
+	  		)
+		)
 		AND tbm.table_type = 'TABLE'
-		`, userName, userName)
+		`, userName, userName, userName)
 
 	return query
 }
@@ -859,22 +867,30 @@ func getDatabaseMetadataBaseQuery(userName string) *safesql.Query {
 		COALESCE(s.store_ids, ARRAY[]) as store_ids,
 		count(*) OVER() as total_row_count
 		FROM system.namespace n
-		LEFT JOIN  system.table_metadata tbm ON n.id = tbm.db_id
-		LEFT JOIN system.role_members rm ON rm.role = 'admin' AND member = $
+		LEFT JOIN system.table_metadata tbm ON n.id = tbm.db_id
 		LEFT JOIN (
 			SELECT db_id, array_agg(DISTINCT unnested_ids) as store_ids
 			FROM system.table_metadata, unnest(store_ids) as unnested_ids
 			GROUP BY db_id
 		) s ON s.db_id = tbm.db_id
-		WHERE (rm.role = 'admin' OR n.name IN (
-	  			SELECT cdp.database_name
-	  			FROM "".crdb_internal.cluster_database_privileges cdp
-	  			WHERE (grantee = $ OR grantee = 'public')
-	  			AND privilege_type = 'CONNECT'
-		))
+		WHERE (
+			$ = 'admin'
+			OR EXISTS (
+				SELECT 1
+				FROM system.role_members rm
+				WHERE rm.member = $
+					AND rm.role = 'admin'
+			)
+			OR n.name IN (
+				SELECT cdp.database_name
+				FROM "".crdb_internal.cluster_database_privileges AS cdp
+				WHERE (cdp.grantee = $ OR cdp.grantee = 'public')
+					AND cdp.privilege_type = 'CONNECT'
+			)
+		)
 		AND n."parentID" = 0
 		AND n."parentSchemaID" = 0
-`, userName, userName)
+`, userName, userName, userName)
 
 	return query
 }

pkg/server/api_v2_databases_metadata_test.go

@@ -388,6 +388,23 @@ func TestGetTableMetadataWithDetails(t *testing.T) {
 			t, userClient, ts.AdminURL().WithPath(uri).String(), http.MethodGet)
 		require.NotEmpty(t, resp.Metadata)
 		require.Contains(t, resp.CreateStatement, table.tableName)
+
+		// Test with dedicated admin user (user named 'admin').
+		adminUsername := username.AdminRoleName()
+		adminClient, _, err := ts.GetAuthenticatedHTTPClientAndCookie(adminUsername, false, 1)
+		require.NoError(t, err)
+
+		// Admin user should be able to access the table even without explicit CONNECT grants.
+		resp = makeApiRequest[tableMetadataWithDetailsResponse](
+			t, adminClient, ts.AdminURL().WithPath(uri).String(), http.MethodGet)
+		require.NotEmpty(t, resp.Metadata)
+		require.Contains(t, resp.CreateStatement, table.tableName)
+
+		// Admin user should still have access even after revoking public access was already done above.
+		resp = makeApiRequest[tableMetadataWithDetailsResponse](
+			t, adminClient, ts.AdminURL().WithPath(uri).String(), http.MethodGet)
+		require.NotEmpty(t, resp.Metadata)
+		require.Contains(t, resp.CreateStatement, table.tableName)
 	})
 
 	t.Run("non GET method 405 error", func(t *testing.T) {
@@ -508,14 +525,14 @@ func TestGetDbMetadata(t *testing.T) {
 
 		// All databases grant CONNECT to public by default, so the user should see all databases.
 		// There should be 4: defaultdb, postgres, new_test_db_1, and new_test_db_2.
-		// The system db should not be included, since it doe snot have CONNECT granted to public.
+		// The system db should not be included, since it does not have CONNECT granted to public.
 		uri := "/api/v2/database_metadata/?sortBy=name"
 		mdResp := makeApiRequest[PaginatedResponse[[]dbMetadata]](t, userClient, ts.AdminURL().WithPath(uri).String(), http.MethodGet)
 		verifyDatabases([]string{"defaultdb", "new_test_db_1", "new_test_db_2", "postgres"}, mdResp.Results)
 
 		// Revoke connect access for public from db1.
 		conn.Exec(t, fmt.Sprintf("REVOKE CONNECT ON DATABASE %s FROM %s", db1Name, "public"))
-		// Asser that user no longer sees db1.
+		// Assert that user no longer sees db1.
 		mdResp = makeApiRequest[PaginatedResponse[[]dbMetadata]](t, userClient, ts.AdminURL().WithPath(uri).String(), http.MethodGet)
 		verifyDatabases([]string{"defaultdb", "new_test_db_2", "postgres"}, mdResp.Results)
 
@@ -535,6 +552,23 @@ func TestGetDbMetadata(t *testing.T) {
 		conn.Exec(t, fmt.Sprintf("GRANT admin TO %s", sessionUsername.Normalized()))
 		mdResp = makeApiRequest[PaginatedResponse[[]dbMetadata]](t, userClient, ts.AdminURL().WithPath(uri).String(), http.MethodGet)
 		verifyDatabases([]string{"defaultdb", "new_test_db_1", "new_test_db_2", "postgres", "system"}, mdResp.Results)
+
+		// Test with dedicated admin user (user named 'admin').
+		adminUsername := username.AdminRoleName()
+		adminClient, _, err := ts.GetAuthenticatedHTTPClientAndCookie(adminUsername, false, 1)
+		require.NoError(t, err)
+
+		// The admin user should see all databases even without explicit CONNECT grants.
+		// There should be 5: defaultdb, postgres, new_test_db_1, and new_test_db_2, system.
+		mdResp = makeApiRequest[PaginatedResponse[[]dbMetadata]](t, adminClient, ts.AdminURL().WithPath(uri).String(), http.MethodGet)
+		verifyDatabases([]string{"defaultdb", "new_test_db_1", "new_test_db_2", "postgres", "system"}, mdResp.Results)
+
+		// Revoke CONNECT on public from both test databases to ensure the admin user
+		// can still see them.
+		conn.Exec(t, fmt.Sprintf("REVOKE CONNECT ON DATABASE %s FROM %s", db1Name, "public"))
+		conn.Exec(t, fmt.Sprintf("REVOKE CONNECT ON DATABASE %s FROM %s", db2Name, "public"))
+		mdResp = makeApiRequest[PaginatedResponse[[]dbMetadata]](t, adminClient, ts.AdminURL().WithPath(uri).String(), http.MethodGet)
+		verifyDatabases([]string{"defaultdb", "new_test_db_1", "new_test_db_2", "postgres", "system"}, mdResp.Results)
 	})
 
 	t.Run("pagination", func(t *testing.T) {
@@ -700,6 +734,26 @@ func TestGetDbMetadataWithDetails(t *testing.T) {
 		// Assert that user can see system db.
 		resp = makeApiRequest[dbMetadataWithDetailsResponse](t, userClient, ts.AdminURL().WithPath(systemUri).String(), http.MethodGet)
 		require.Equal(t, int64(1), resp.Metadata.DbId)
+
+		// Test with dedicated admin user (user named 'admin').
+		adminUsername := username.AdminRoleName()
+		adminClient, _, err := ts.GetAuthenticatedHTTPClientAndCookie(adminUsername, false, 1)
+		require.NoError(t, err)
+
+		// Admin user should be able to access the database even without explicit CONNECT grants.
+		resp = makeApiRequest[dbMetadataWithDetailsResponse](
+			t, adminClient, ts.AdminURL().WithPath(uri).String(), http.MethodGet)
+		require.Equal(t, int64(db1Id), resp.Metadata.DbId)
+
+		// Admin user should also see the system database.
+		resp = makeApiRequest[dbMetadataWithDetailsResponse](
+			t, adminClient, ts.AdminURL().WithPath(systemUri).String(), http.MethodGet)
+		require.Equal(t, int64(1), resp.Metadata.DbId)
+
+		// Admin user should still have access even after public access was already revoked above.
+		resp = makeApiRequest[dbMetadataWithDetailsResponse](
+			t, adminClient, ts.AdminURL().WithPath(uri).String(), http.MethodGet)
+		require.Equal(t, int64(db1Id), resp.Metadata.DbId)
 	})
 
 	t.Run("non GET method 405 error", func(t *testing.T) {

pkg/server/node.go

@@ -1158,29 +1158,23 @@ func (n *Node) startPeriodicLivenessCompaction(
 				_ = n.stores.VisitStores(func(store *kvserver.Store) error {
 					store.VisitReplicas(func(repl *kvserver.Replica) bool {
 						span := repl.Desc().KeySpan().AsRawSpanWithNoLocals()
-						if keys.NodeLivenessSpan.Overlaps(span) {
-
-							// The CompactRange() method expects the start and end keys to be
-							// encoded.
-							startEngineKey :=
-								storage.EngineKey{
-									Key: span.Key,
-								}.Encode()
-
-							endEngineKey :=
-								storage.EngineKey{
-									Key: span.EndKey,
-								}.Encode()
-
-							timeBeforeCompaction := timeutil.Now()
-							if err := store.StateEngine().CompactRange(
-								context.Background(), startEngineKey, endEngineKey); err != nil {
-								log.Dev.Errorf(ctx, "failed compacting liveness replica: %+v with error: %s", repl, err)
-							}
-
-							log.Dev.Infof(ctx, "finished compacting liveness replica: %+v and it took: %+v",
-								repl, timeutil.Since(timeBeforeCompaction))
+						if !keys.NodeLivenessSpan.Overlaps(span) {
+							return true
 						}
+
+						// CompactRange() expects the start and end keys to be encoded.
+						startEngineKey := storage.EngineKey{Key: span.Key}.Encode()
+						endEngineKey := storage.EngineKey{Key: span.EndKey}.Encode()
+
+						timeBeforeCompaction := timeutil.Now()
+						if err := store.StateEngine().CompactRange(
+							context.Background(), startEngineKey, endEngineKey,
+						); err != nil {
+							log.Dev.Errorf(ctx, "failed compacting liveness replica: %+v with error: %s", repl, err)
+						}
+
+						log.Dev.Infof(ctx, "finished compacting liveness replica: %+v and it took: %+v",
+							repl, timeutil.Since(timeBeforeCompaction))
 						return true
 					})
 					return nil