Merge #60707 #60953

60707: cli,storage/cloud: add CLI flag to disable outbound network storage access r=dt a=dt

Multi-tenant clusters will initially restrict outbound network access.
In addition to the firewall rules at deployment that will actually block the
network traffic itself, this CLI flag can be used to indeed prevent even trying
make such network requests and return a clearer error instead.

Release note: none.

Release justification: Low risk (opt-in flag), high reward (better errors
    for serverless users) changes to existing functionality.

60953: sql: use AOST -10s when building crdb_internal.table_row_statistics r=rytaft a=rytaft

This commit updates the query used to build the virtual table
`crdb_internal.table_row_statistics` so that it is always run at 
`AS OF SYSTEM TIME '-10s'`. This will reduce contention on the
`system.table_statistics` table and avoid delays when running `SHOW TABLES`,
which queries `crdb_internal.table_row_statistics`.

Informs #58189

Release note (performance improvement): Updated the query used to build
the virtual table crdb_internal.table_row_statistics so that it is always
run at AS OF SYSTEM TIME '-10s'. This should reduce contention on
the table and improve performance for transactions that rely on
crdb_internal.table_row_statistics, such as SHOW TABLES.

Co-authored-by: David Taylor <tinystatemachine@gmail.com>
Co-authored-by: Rebecca Taft <becca@cockroachlabs.com>

pkg/base/config.go

@@ -560,6 +560,10 @@ type ExternalIODirConfig struct {
 	// This turns off implicit credentials, and requires the user to provide
 	// necessary access keys.
 	DisableImplicitCredentials bool
+
+	// DisableOutbound disables the use of any external-io that dials out such as
+	// to s3, gcs, or even `nodelocal` as it may need to dial another node.
+	DisableOutbound bool
 }
 
 // TempStorageConfigFromEnv creates a TempStorageConfig.

pkg/ccl/storageccl/export.go

@@ -101,13 +101,10 @@ func evalExport(
 	}
 
 	if makeExternalStorage {
-		// TODO(dt): this blanket ban means we must do all uploads from the caller
-		// which is nice and simple but imposes extra copies/overhead/cost. We might
-		// want to instead allow *some* forms of external storage for *some* tenants
-		// e.g. allow some tenants to dial out to s3 directly -- if we do though we
-		// would need to continue to restrict unsafe ones like userfile here.
 		if _, ok := roachpb.TenantFromContext(ctx); ok {
-			return result.Result{}, errors.Errorf("requests on behalf of tenants are not allowed to contact external storage")
+			if args.Storage.Provider == roachpb.ExternalStorageProvider_FileTable {
+				return result.Result{}, errors.Errorf("requests to userfile on behalf of tenants must be made by the tenant's SQL process")
+			}
 		}
 	}
 

pkg/cli/cliflags/flags.go

@@ -678,6 +678,11 @@ Also see: ` + build.MakeIssueURL(53404) + `
 Disable use of implicit credentials when accessing external data.
 Instead, require the user to always specify access keys.`,
 	}
+	ExternalIODisabled = FlagInfo{
+		Name: "external-io-disabled",
+		Description: `
+Disable use of "external" IO, such as to S3, GCS, or the file system (nodelocal), or anything other than userfile.`,
+	}
 
 	// KeySize, CertificateLifetime, AllowKeyReuse, and OverwriteFiles are used for
 	// certificate generation functions.

pkg/cli/flags.go

@@ -397,6 +397,7 @@ func init() {
 
 		// Enable/disable various external storage endpoints.
 		boolFlag(f, &serverCfg.ExternalIODirConfig.DisableHTTP, cliflags.ExternalIODisableHTTP)
+		boolFlag(f, &serverCfg.ExternalIODirConfig.DisableOutbound, cliflags.ExternalIODisabled)
 		boolFlag(f, &serverCfg.ExternalIODirConfig.DisableImplicitCredentials, cliflags.ExternalIODisableImplicitCredentials)
 
 		// Certificates directory. Use a server-specific flag and value to ignore environment
@@ -848,6 +849,11 @@ func init() {
 		stringSliceFlag(f, &serverCfg.SQLConfig.TenantKVAddrs, cliflags.KVAddrs)
 
 		durationFlag(f, &serverCfg.IdleExitAfter, cliflags.IdleExitAfter)
+
+		boolFlag(f, &serverCfg.ExternalIODirConfig.DisableHTTP, cliflags.ExternalIODisableHTTP)
+		boolFlag(f, &serverCfg.ExternalIODirConfig.DisableOutbound, cliflags.ExternalIODisabled)
+		boolFlag(f, &serverCfg.ExternalIODirConfig.DisableImplicitCredentials, cliflags.ExternalIODisableImplicitCredentials)
+
 	}
 }
 

pkg/sql/catalog/descs/collection.go

@@ -309,7 +309,8 @@ func (tc *Collection) getDescriptorFromStore(
 ) (found bool, desc catalog.Descriptor, err error) {
 	// Bypass the namespace lookup from the store for system tables.
 	descID := bootstrap.LookupSystemTableDescriptorID(ctx, tc.settings, tc.codec(), parentID, name)
-	if descID == descpb.InvalidID {
+	isSystemDescriptor := descID != descpb.InvalidID
+	if !isSystemDescriptor {
 		var found bool
 		var err error
 		found, descID, err = catalogkv.LookupObjectID(ctx, txn, codec, parentID, parentSchemaID, name)
@@ -321,6 +322,10 @@ func (tc *Collection) getDescriptorFromStore(
 	desc, err = catalogkv.GetAnyDescriptorByID(ctx, txn, codec, descID, catalogkv.Mutable)
 	if err != nil {
 		return false, nil, err
+	} else if desc == nil && isSystemDescriptor {
+		// This can happen during startup because we're not actually looking up the
+		// system descriptor IDs in KV.
+		return false, nil, errors.Wrapf(catalog.ErrDescriptorNotFound, "descriptor %d not found", descID)
 	} else if desc == nil {
 		// Having done the namespace lookup, the descriptor must exist.
 		return false, nil, errors.AssertionFailedf("descriptor %d not found", descID)

pkg/sql/crdb_internal.go

@@ -409,30 +409,49 @@ CREATE TABLE crdb_internal.tables (
 	},
 }
 
+// statsAsOfTimeClusterMode controls the cluster setting for the duration which
+// is used to define the AS OF time for querying the system.table_statistics
+// table when building crdb_internal.table_row_statistics.
+var statsAsOfTimeClusterMode = settings.RegisterDurationSetting(
+	"sql.crdb_internal.table_row_statistics.as_of_time",
+	"historical query time used to build the crdb_internal.table_row_statistics table",
+	-10*time.Second,
+)
+
 var crdbInternalTablesTableLastStats = virtualSchemaTable{
-	comment: "the latest stats for all tables accessible by current user in current database (KV scan)",
+	comment: "stats for all tables accessible by current user in current database as of 10s ago",
 	schema: `
 CREATE TABLE crdb_internal.table_row_statistics (
   table_id                   INT         NOT NULL,
   table_name                 STRING      NOT NULL,
   estimated_row_count        INT
 )`,
 	populate: func(ctx context.Context, p *planner, db *dbdesc.Immutable, addRow func(...tree.Datum) error) error {
-		// Collect the latests statistics for all tables.
-		query := `
+		// Collect the statistics for all tables AS OF 10 seconds ago to avoid
+		// contention on the stats table. We pass a nil transaction so that the AS
+		// OF clause can be independent of any outer query.
+		query := fmt.Sprintf(`
            SELECT s."tableID", max(s."rowCount")
              FROM system.table_statistics AS s
              JOIN (
                     SELECT "tableID", max("createdAt") AS last_dt
                       FROM system.table_statistics
                      GROUP BY "tableID"
                   ) AS l ON l."tableID" = s."tableID" AND l.last_dt = s."createdAt"
-            GROUP BY s."tableID"`
+            AS OF SYSTEM TIME '%s'
+            GROUP BY s."tableID"`, statsAsOfTimeClusterMode.String(&p.ExecCfg().Settings.SV))
 		statRows, err := p.ExtendedEvalContext().ExecCfg.InternalExecutor.QueryEx(
-			ctx, "crdb-internal-statistics-table", p.txn,
+			ctx, "crdb-internal-statistics-table", nil,
 			sessiondata.InternalExecutorOverride{User: security.RootUserName()},
 			query)
 		if err != nil {
+			// This query is likely to cause errors due to SHOW TABLES being run less
+			// than 10 seconds after cluster startup (10s is the default AS OF time
+			// for the query), causing the error "descriptor not found". We should
+			// tolerate this error and return nil.
+			if errors.Is(err, catalog.ErrDescriptorNotFound) {
+				return nil
+			}
 			return err
 		}
 

pkg/sql/logictest/logic.go

@@ -1577,6 +1577,14 @@ func (t *logicTest) newCluster(serverArgs TestServerArgs) {
 		if _, err := conn.Exec("SET CLUSTER SETTING sql.defaults.interleaved_tables.enabled = true"); err != nil {
 			t.Fatal(err)
 		}
+
+		// Update the default AS OF time for querying the system.table_statistics
+		// table to create the crdb_internal.table_row_statistics table.
+		if _, err := conn.Exec(
+			"SET CLUSTER SETTING sql.crdb_internal.table_row_statistics.as_of_time = '-1µs'",
+		); err != nil {
+			t.Fatal(err)
+		}
 	}
 
 	if cfg.overrideDistSQLMode != "" {

pkg/sql/logictest/testdata/logic_test/event_log

@@ -470,6 +470,7 @@ ORDER BY "timestamp", info
 0  1  {"EventType": "set_cluster_setting", "SettingName": "kv.range_merge.queue_enabled", "Statement": "SET CLUSTER SETTING \"kv.range_merge.queue_enabled\" = false", "User": "root", "Value": "false"}
 0  1  {"EventType": "set_cluster_setting", "PlaceholderValues": ["5"], "SettingName": "sql.stats.automatic_collection.min_stale_rows", "Statement": "SET CLUSTER SETTING \"sql.stats.automatic_collection.min_stale_rows\" = $1::INT8", "User": "root", "Value": "5"}
 0  1  {"EventType": "set_cluster_setting", "SettingName": "sql.defaults.interleaved_tables.enabled", "Statement": "SET CLUSTER SETTING \"sql.defaults.interleaved_tables.enabled\" = true", "User": "root", "Value": "true"}
+0  1  {"EventType": "set_cluster_setting", "SettingName": "sql.crdb_internal.table_row_statistics.as_of_time", "Statement": "SET CLUSTER SETTING \"sql.crdb_internal.table_row_statistics.as_of_time\" = e'-1\\u00B5s'", "User": "root", "Value": "-00:00:00.000001"}
 0  1  {"EventType": "set_cluster_setting", "SettingName": "kv.allocator.load_based_lease_rebalancing.enabled", "Statement": "SET CLUSTER SETTING \"kv.allocator.load_based_lease_rebalancing.enabled\" = false", "User": "root", "Value": "false"}
 0  1  {"EventType": "set_cluster_setting", "SettingName": "kv.allocator.load_based_lease_rebalancing.enabled", "Statement": "SET CLUSTER SETTING \"kv.allocator.load_based_lease_rebalancing.enabled\" = DEFAULT", "User": "root", "Value": "DEFAULT"}
 0  1  {"EventType": "set_cluster_setting", "PlaceholderValues": ["'some string'"], "SettingName": "cluster.organization", "Statement": "SET CLUSTER SETTING \"cluster.organization\" = $1", "User": "root", "Value": "'some string'"}

pkg/sql/logictest/testdata/logic_test/pg_catalog

@@ -2119,7 +2119,7 @@ objoid      classoid    objsubid  description
 4294967261  4294967194  0         session variables (RAM)
 4294967259  4294967194  0         details for all columns accessible by current user in current database (KV scan)
 4294967258  4294967194  0         indexes accessible by current user in current database (KV scan)
-4294967256  4294967194  0         the latest stats for all tables accessible by current user in current database (KV scan)
+4294967256  4294967194  0         stats for all tables accessible by current user in current database as of 10s ago
 4294967257  4294967194  0         table descriptors accessible by current user, including non-public and virtual (KV scan; expensive!)
 4294967253  4294967194  0         decoded zone configurations from system.zones (KV scan)
 4294967249  4294967194  0         roles for which the current user has admin option

pkg/sql/logictest/testdata/logic_test/system

@@ -589,6 +589,7 @@ ORDER BY name
 cluster.secret
 diagnostics.reporting.enabled
 kv.range_merge.queue_enabled
+sql.crdb_internal.table_row_statistics.as_of_time
 sql.defaults.interleaved_tables.enabled
 sql.stats.automatic_collection.min_stale_rows
 version
@@ -605,11 +606,12 @@ WHERE name NOT IN ('version', 'sql.defaults.distsql', 'cluster.secret',
   'sql.defaults.experimental_distsql_planning')
 ORDER BY name
 ----
-diagnostics.reporting.enabled                  true
-kv.range_merge.queue_enabled                   false
-somesetting                                    somevalue
-sql.defaults.interleaved_tables.enabled        true
-sql.stats.automatic_collection.min_stale_rows  5
+diagnostics.reporting.enabled                      true
+kv.range_merge.queue_enabled                       false
+somesetting                                        somevalue
+sql.crdb_internal.table_row_statistics.as_of_time  -1µs
+sql.defaults.interleaved_tables.enabled            true
+sql.stats.automatic_collection.min_stale_rows      5
 
 user testuser
 

pkg/storage/cloudimpl/aws_kms.go

@@ -69,6 +69,9 @@ func resolveKMSURIParams(kmsURI url.URL) kmsURIParams {
 // MakeAWSKMS is the factory method which returns a configured, ready-to-use
 // AWS KMS object.
 func MakeAWSKMS(uri string, env cloud.KMSEnv) (cloud.KMS, error) {
+	if env.KMSConfig().DisableOutbound {
+		return nil, errors.New("external IO must be enabled to use AWS KMS")
+	}
 	kmsURI, err := url.ParseRequestURI(uri)
 	if err != nil {
 		return nil, err

pkg/storage/cloudimpl/cloudimpltests/http_storage_test.go

@@ -308,6 +308,26 @@ func TestCanDisableHttp(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestCanDisableOutbound(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	conf := base.ExternalIODirConfig{
+		DisableOutbound: true,
+	}
+	for _, provider := range []roachpb.ExternalStorageProvider{
+		roachpb.ExternalStorageProvider_Http,
+		roachpb.ExternalStorageProvider_S3,
+		roachpb.ExternalStorageProvider_GoogleCloud,
+		roachpb.ExternalStorageProvider_LocalFile,
+	} {
+		s, err := cloudimpl.MakeExternalStorage(
+			context.Background(),
+			roachpb.ExternalStorage{Provider: provider},
+			conf, testSettings, blobs.TestEmptyBlobClientFactory, nil, nil)
+		require.Nil(t, s)
+		require.Error(t, err)
+	}
+}
+
 func TestExternalStorageCanUseHTTPProxy(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

pkg/storage/cloudimpl/external_storage.go

@@ -304,6 +304,9 @@ func MakeExternalStorage(
 	ie *sql.InternalExecutor,
 	kvDB *kv.DB,
 ) (cloud.ExternalStorage, error) {
+	if conf.DisableOutbound && dest.Provider != roachpb.ExternalStorageProvider_FileTable {
+		return nil, errors.New("external network access is disabled")
+	}
 	switch dest.Provider {
 	case roachpb.ExternalStorageProvider_LocalFile:
 		telemetry.Count("external-io.nodelocal")