Report and retry failing default privileges of prepared schemas

pkg/cluster/database.go

@@ -472,33 +472,46 @@ func (c *Cluster) execCreateDatabaseSchema(databaseName, schemaName, dbOwner, sc
 		return fmt.Errorf("could not execute %s: %v", operation, err)
 	}
 
-	// set default privileges for schema
+	// set default privileges for schema, collect errors so that they get reported and retried
+	var errs []string
 	// the schemaOwner defines them for global database roles
-	c.execAlterSchemaDefaultPrivileges(schemaName, schemaOwner, databaseName)
+	if err := c.execAlterSchemaDefaultPrivileges(schemaName, schemaOwner, databaseName); err != nil {
+		errs = append(errs, err.Error())
+	}
 
 	// if schemaOwner and dbOwner differ we know that <databaseName>_<schemaName> default roles were created
 	if schemaOwner != dbOwner {
 		defaultUsers := c.Spec.PreparedDatabases[databaseName].PreparedSchemas[schemaName].DefaultUsers
 
 		// define schema privileges of <databaseName>_<schemaName>_owner_user for global roles, too
 		if defaultUsers {
-			c.execAlterSchemaDefaultPrivileges(schemaName, schemaOwner+constants.UserRoleNameSuffix, databaseName)
+			if err := c.execAlterSchemaDefaultPrivileges(schemaName, schemaOwner+constants.UserRoleNameSuffix, databaseName); err != nil {
+				errs = append(errs, err.Error())
+			}
 		}
 
 		// collect all possible owner roles and define default schema privileges
 		// for <databaseName>_<schemaName>_reader/writer roles
 		owners := c.getOwnerRoles(databaseName, c.Spec.PreparedDatabases[databaseName].DefaultUsers)
 		owners = append(owners, c.getOwnerRoles(databaseName+"_"+schemaName, defaultUsers)...)
 		for _, owner := range owners {
-			c.execAlterSchemaDefaultPrivileges(schemaName, owner, databaseName+"_"+schemaName)
+			if err := c.execAlterSchemaDefaultPrivileges(schemaName, owner, databaseName+"_"+schemaName); err != nil {
+				errs = append(errs, err.Error())
+			}
 		}
 	} else {
 		// define schema privileges of <databaseName>_owner_user for global roles, too
 		if c.Spec.PreparedDatabases[databaseName].DefaultUsers {
-			c.execAlterSchemaDefaultPrivileges(schemaName, schemaOwner+constants.UserRoleNameSuffix, databaseName)
+			if err := c.execAlterSchemaDefaultPrivileges(schemaName, schemaOwner+constants.UserRoleNameSuffix, databaseName); err != nil {
+				errs = append(errs, err.Error())
+			}
 		}
 	}
 
+	if len(errs) > 0 {
+		return fmt.Errorf("could not set default privileges for database schema %s: %s", schemaName, strings.Join(errs, `', '`))
+	}
+
 	return nil
 }
 

pkg/cluster/prepared_schemas_test.go

@@ -0,0 +1,119 @@
+package cluster
+
+import (
+	"database/sql"
+	"database/sql/driver"
+	"fmt"
+	"io"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	acidv1 "github.com/zalando/postgres-operator/v2/pkg/apis/acid.zalan.do/v1"
+)
+
+// fakeSchemaDB is a minimal database/sql driver. Statements that mention
+// failOn are rejected, getSchemasSQL returns the schemas created so far.
+type fakeSchemaDB struct {
+	mu      sync.Mutex
+	failOn  string
+	schemas []string
+	execs   []string
+}
+
+type fakeConn struct{ db *fakeSchemaDB }
+type fakeRows struct {
+	schemas []string
+	pos     int
+}
+
+func (f *fakeSchemaDB) Open(string) (driver.Conn, error) { return &fakeConn{f}, nil }
+func (c *fakeConn) Prepare(q string) (driver.Stmt, error) {
+	return nil, fmt.Errorf("prepare not supported")
+}
+func (c *fakeConn) Close() error              { return nil }
+func (c *fakeConn) Begin() (driver.Tx, error) { return nil, fmt.Errorf("tx not supported") }
+
+func (c *fakeConn) Exec(q string, _ []driver.Value) (driver.Result, error) {
+	c.db.mu.Lock()
+	defer c.db.mu.Unlock()
+	c.db.execs = append(c.db.execs, q)
+	if c.db.failOn != "" && strings.Contains(q, c.db.failOn) {
+		return nil, fmt.Errorf("role does not exist")
+	}
+	if strings.Contains(q, "CREATE SCHEMA") {
+		c.db.schemas = append(c.db.schemas, "data")
+	}
+	return driver.RowsAffected(0), nil
+}
+
+func (c *fakeConn) Query(q string, _ []driver.Value) (driver.Rows, error) {
+	c.db.mu.Lock()
+	defer c.db.mu.Unlock()
+	return &fakeRows{schemas: append([]string{}, c.db.schemas...)}, nil
+}
+
+func (r *fakeRows) Columns() []string { return []string{"dbschema"} }
+func (r *fakeRows) Close() error      { return nil }
+func (r *fakeRows) Next(dest []driver.Value) error {
+	if r.pos >= len(r.schemas) {
+		return io.EOF
+	}
+	dest[0] = r.schemas[r.pos]
+	r.pos++
+	return nil
+}
+
+func (f *fakeSchemaDB) privilegeExecs() (n int) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for _, q := range f.execs {
+		if strings.Contains(q, "ALTER DEFAULT PRIVILEGES") {
+			n++
+		}
+	}
+	return n
+}
+
+func TestPreparedSchemaPrivilegeErrors(t *testing.T) {
+	driverName := "fake-schema-db-" + strings.ReplaceAll(t.Name(), "/", "-")
+	fake := &fakeSchemaDB{failOn: "schema_db_writer"}
+	sql.Register(driverName, fake)
+	db, err := sql.Open(driverName, "")
+	assert.NoError(t, err)
+	defer db.Close()
+
+	noDefaults := false
+	var cluster Cluster
+	cluster.logger = logrus.New().WithField("test", "prepared-schemas")
+	cluster.pgDb = db
+	cluster.Spec.PreparedDatabases = map[string]acidv1.PreparedDatabase{
+		"schema_db": {
+			DefaultUsers: false,
+			PreparedSchemas: map[string]acidv1.PreparedSchema{
+				"data": {DefaultRoles: &noDefaults},
+			},
+		},
+	}
+	schemas := cluster.Spec.PreparedDatabases["schema_db"].PreparedSchemas
+
+	// schema is created, but granting privileges fails: error must be returned
+	err = cluster.syncPreparedSchemas("schema_db", schemas)
+	assert.Error(t, err, "failing default privileges must be reported")
+	assert.Contains(t, fmt.Sprint(err), "could not alter default privileges")
+
+	// the schema exists now; a later sync must still report the problem ...
+	err = cluster.syncPreparedSchemas("schema_db", schemas)
+	assert.Error(t, err, "missing privileges must be retried while the schema exists")
+
+	// ... and succeed once the missing role was created
+	fake.mu.Lock()
+	fake.failOn = ""
+	fake.mu.Unlock()
+	before := fake.privilegeExecs()
+	err = cluster.syncPreparedSchemas("schema_db", schemas)
+	assert.NoError(t, err)
+	assert.Greater(t, fake.privilegeExecs(), before, "privileges must be applied on retry")
+}

pkg/cluster/sync.go

@@ -1659,29 +1659,18 @@ func (c *Cluster) syncPreparedSchemas(databaseName string, preparedSchemas map[s
 	c.setProcessName("syncing prepared schemas")
 	errors := make([]string, 0)
 
-	currentSchemas, err := c.getSchemas()
-	if err != nil {
-		return fmt.Errorf("could not get current schemas: %v", err)
-	}
-
-	var schemas []string
-
-	for schema := range preparedSchemas {
-		schemas = append(schemas, schema)
-	}
-
-	if createPreparedSchemas, equal := util.SubstractStringSlices(schemas, currentSchemas); !equal {
-		for _, schemaName := range createPreparedSchemas {
-			owner := constants.OwnerRoleNameSuffix
-			dbOwner := fmt.Sprintf("%s%s", databaseName, owner)
-			if preparedSchemas[schemaName].DefaultRoles == nil || *preparedSchemas[schemaName].DefaultRoles {
-				owner = fmt.Sprintf("%s_%s%s", databaseName, schemaName, owner)
-			} else {
-				owner = dbOwner
-			}
-			if err = c.executeCreateDatabaseSchema(databaseName, schemaName, dbOwner, owner); err != nil {
-				errors = append(errors, err.Error())
-			}
+	// CREATE SCHEMA IF NOT EXISTS and ALTER DEFAULT PRIVILEGES are idempotent, so handle
+	// existing schemas, too. This retries privileges that could not be set earlier.
+	for schemaName := range preparedSchemas {
+		owner := constants.OwnerRoleNameSuffix
+		dbOwner := fmt.Sprintf("%s%s", databaseName, owner)
+		if preparedSchemas[schemaName].DefaultRoles == nil || *preparedSchemas[schemaName].DefaultRoles {
+			owner = fmt.Sprintf("%s_%s%s", databaseName, schemaName, owner)
+		} else {
+			owner = dbOwner
+		}
+		if err := c.executeCreateDatabaseSchema(databaseName, schemaName, dbOwner, owner); err != nil {
+			errors = append(errors, err.Error())
 		}
 	}