sql: add sql.log.user_audit.reduced_config.enabled to support single

role checks for role-based audit logs

This change introduces the `sql.log.user_audit.reduced_config.enabled`
cluster setting. When enabled, this cluster setting computes a "reduced"
audit configuration based on the user's current role memberships and the
current value for the `sql.log.user_audit` cluster setting. The
"reduced" audit configuration is computed at the _first SQL event emit
by the user after the setting is enabled_. Enabling the cluster setting
allows us to compute the audit configuration once at session start,
instead of computing at each SQL event (providing ~5% increase in
throughput). The tradeoff is that changes to the audit configuration
(user role memberships or cluster setting configuration) are not
reflected within session. Users will need to start a new session to see
these changes in their auditing behaviour.

Release note (sql change): Introduce new cluster setting
`sql.log.user_audit.reduced_config.enabled`. When enabled, this cluster
setting computes a "reduced" audit configuration based on the user's
current role memberships and the current value for the
`sql.log.user_audit` cluster setting. The "reduced" audit configuration
is computed at the _first SQL event emit by the user after the setting
is enabled_. Enabling the cluster setting allows us to compute the audit
configuration once at session start, instead of computing at each SQL
event (providing ~5% increase in throughput). The tradeoff is that
changes to the audit configuration (user role memberships or cluster
setting configuration) are not reflected within session. Users will need
to start a new session to see these changes in their auditing behaviour.

https://cockroachlabs.atlassian.net/browse/DOC-1355 #Informs:
cockroachdb#33316 #Epic: CRDB-8035

pkg/sql/audit_logging.go

@@ -45,6 +45,12 @@ func (p *planner) maybeAuditRoleBasedAuditEvent(ctx context.Context) {
 		return
 	}
 
+	// Use reduced audit config is enabled.
+	if auditlogging.UserAuditEnableReducedConfig.Get(&p.execCfg.Settings.SV) == true {
+		p.logReducedAuditConfig(ctx)
+		return
+	}
+
 	user := p.User()
 	userRoles, err := p.MemberOfWithAdminOption(ctx, user)
 	if err != nil {
@@ -68,6 +74,41 @@ func (p *planner) maybeAuditRoleBasedAuditEvent(ctx context.Context) {
 	}
 }
 
+func (p *planner) logReducedAuditConfig(ctx context.Context) {
+	if !p.reducedAuditConfig.Initialized {
+		p.initializeReducedAuditConfig(ctx)
+	}
+
+	// Return early if no matching audit setting was found.
+	if p.reducedAuditConfig.AuditSetting == nil {
+		return
+	}
+
+	if p.reducedAuditConfig.AuditSetting.IncludeStatements {
+		p.curPlan.auditEventBuilders = append(p.curPlan.auditEventBuilders,
+			&auditevents.RoleBasedAuditEvent{
+				Role: p.reducedAuditConfig.AuditSetting.Role.Normalized(),
+			},
+		)
+	}
+}
+
+func (p *planner) initializeReducedAuditConfig(ctx context.Context) {
+	// Set the initialized flag to true, even for an attempt that errors.
+	// We do this to avoid the potential overhead of continuously retrying
+	// to fetch user role memberships.
+	p.reducedAuditConfig.Initialized = true
+
+	user := p.User()
+	userRoles, err := p.MemberOfWithAdminOption(ctx, user)
+	if err != nil {
+		log.Errorf(ctx, "initialize reduced audit config: error getting user role memberships: %v", err)
+		return
+	}
+	// Get matching audit setting.
+	p.reducedAuditConfig.AuditSetting = p.AuditConfig().GetMatchingAuditSetting(userRoles, user)
+}
+
 // shouldNotRoleBasedAudit checks if we should do any auditing work for RoleBasedAuditEvents.
 func (p *planner) shouldNotRoleBasedAudit() bool {
 	// Do not do audit work if the cluster setting is empty.

pkg/sql/audit_logging_test.go

@@ -269,6 +269,126 @@ func TestMultiRoleAuditLogging(t *testing.T) {
 	}
 }
 
+func TestReducedAuditConfig(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	sc := log.ScopeWithoutShowLogs(t)
+	defer sc.Close(t)
+
+	cleanup := logtestutils.InstallLogFileSink(sc, t, logpb.Channel_SENSITIVE_ACCESS)
+	defer cleanup()
+
+	s, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	rootRunner := sqlutils.MakeSQLRunner(sqlDB)
+	defer s.Stopper().Stop(context.Background())
+
+	// Dummy table/user used by tests.
+	setupQueries(t, rootRunner)
+
+	// Enable reduced config.
+	rootRunner.Exec(t, `SET CLUSTER SETTING sql.log.user_audit.reduced_config = true`)
+	testutils.SucceedsSoon(t, func() error {
+		var currentVal string
+		rootRunner.QueryRow(t,
+			"SHOW CLUSTER SETTING sql.log.user_audit.reduced_config",
+		).Scan(&currentVal)
+
+		if currentVal == "false" {
+			return errors.Newf("waiting for reduced config cluster setting to be true")
+		}
+		return nil
+	})
+
+	testUserURL, cleanupFn := sqlutils.PGUrl(t,
+		s.ServingSQLAddr(), t.Name(), url.User(username.TestUser))
+	defer cleanupFn()
+
+	testUserDb, err := gosql.Open("postgres", testUserURL.String())
+	require.NoError(t, err)
+	defer testUserDb.Close()
+	testRunner := sqlutils.MakeSQLRunner(testUserDb)
+
+	// Set a cluster configuration.
+	roleA := "roleA"
+	rootRunner.Exec(t, `SET CLUSTER SETTING sql.log.user_audit = '
+		roleA ALL
+	'`)
+
+	testutils.SucceedsSoon(t, func() error {
+		var currentVal string
+		rootRunner.QueryRow(t,
+			"SHOW CLUSTER SETTING sql.log.user_audit",
+		).Scan(&currentVal)
+
+		if currentVal == "" {
+			return errors.Newf("waiting for cluster setting to be set")
+		}
+		return nil
+	})
+
+	// Run a query. This initializes the reduced audit configuration for the user.
+	// Currently, there are no corresponding roles for the user in the audit configuration.
+	// Consequently, the user's reduced audit config will be nil.
+	testQuery := `SELECT * FROM u`
+	testRunner.Exec(t, testQuery)
+
+	// Grant a role the user that corresponds to an audit setting.
+	rootRunner.Exec(t, fmt.Sprintf("CREATE ROLE IF NOT EXISTS %s", roleA))
+	rootRunner.Exec(t, fmt.Sprintf("GRANT %s to testuser", roleA))
+
+	// Run the query again. We expect no log from this query even though the user now has a corresponding role
+	// as the reduced audit configuration has already been computed, and there were no corresponding audit settings
+	// for the user at that time.
+	testRunner.Exec(t, testQuery)
+
+	log.Flush()
+
+	entries, err := log.FetchEntriesFromFiles(
+		0,
+		math.MaxInt64,
+		10000,
+		regexp.MustCompile(`"EventType":"role_based_audit_event"`),
+		log.WithMarkedSensitiveData,
+	)
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Expect the number of entries to be 0.
+	if len(entries) != 0 {
+		t.Fatal(errors.Newf("unexpected entries found"))
+	}
+
+	// Open 2nd connection for the test user.
+	testUserDb2, err := gosql.Open("postgres", testUserURL.String())
+	require.NoError(t, err)
+	defer testUserDb2.Close()
+	testRunner2 := sqlutils.MakeSQLRunner(testUserDb2)
+
+	// Run a query on the new connection. The new connection will cause the reduced audit config to be re-computed.
+	// The user now has a corresponding audit setting. We use a new query here to differentiate.
+	testRunner2.Exec(t, `GRANT SELECT ON TABLE u TO root`)
+
+	log.Flush()
+
+	entries, err = log.FetchEntriesFromFiles(
+		0,
+		math.MaxInt64,
+		10000,
+		regexp.MustCompile(`GRANT SELECT ON TABLE ‹u› TO root`),
+		log.WithMarkedSensitiveData,
+	)
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Expect the number of entries to be 1.
+	if len(entries) != 1 {
+		t.Fatal(errors.Newf("unexpected number of entries found (not 1)"))
+	}
+}
+
 func setupQueries(t *testing.T, rootRunner *sqlutils.SQLRunner) {
 	// Dummy table/user used by tests.
 	rootRunner.Exec(t, `CREATE TABLE u(x int)`)

pkg/sql/auditlogging/audit_log.go

@@ -52,6 +52,12 @@ type AuditConfigLock struct {
 	Config *AuditConfig
 }
 
+// ReducedAuditConfig is a computed audit configuration initialized at the first SQL event emit by the user.
+type ReducedAuditConfig struct {
+	Initialized  bool
+	AuditSetting *AuditSetting
+}
+
 // GetMatchingAuditSetting returns the first audit setting in the cluster setting
 // configuration that matches the user's username/roles. If no such audit setting
 // exists, returns nil.

pkg/sql/auditlogging/audit_log_config.go

@@ -28,6 +28,23 @@ var UserAuditLogConfig = settings.RegisterValidatedStringSetting(
 	validateAuditLogConfig,
 )
 
+// UserAuditEnableReducedConfig is a cluster setting that enables/disables a computed
+// reduced configuration. This allows us to compute the audit configuration once at
+// session start, instead of computing at each SQL event. The tradeoff is that changes to
+// the audit configuration (user role memberships or cluster setting configuration) are not
+// reflected within session. Users will need to start a new session to see these changes in their
+// auditing behaviour.
+var UserAuditEnableReducedConfig = settings.RegisterBoolSetting(
+	settings.TenantWritable,
+	"sql.log.user_audit.reduced_config",
+	"enables logic to compute a reduced audit configuration, computing the audit "+
+		"configuration only once at session start instead of at each SQL event. The tradeoff "+
+		"with the increase in performance (~5%), is that changes to the audit configuration "+
+		"(user role memberships/cluster setting) are not reflected within session. "+
+		"Users will need to start a new session to see these changes in their auditing behaviour.",
+	false,
+)
+
 func validateAuditLogConfig(_ *settings.Values, input string) error {
 	if input == "" {
 		// Empty config

pkg/sql/planner.go

@@ -271,6 +271,8 @@ type planner struct {
 
 	// trackDependency is used to track circular dependencies when dropping views.
 	trackDependency map[catid.DescID]bool
+
+	reducedAuditConfig auditlogging.ReducedAuditConfig
 }
 
 // hasFlowForPausablePortal returns true if the planner is for re-executing a