diff --git a/pkg/cluster/database.go b/pkg/cluster/database.go index d46dc6067..e95bb372a 100644 --- a/pkg/cluster/database.go +++ b/pkg/cluster/database.go @@ -34,6 +34,8 @@ const ( AND right(r.rolname, 6) ~ '^[0-9\.]+$' ORDER BY 1;` + isInRecoverySQL = `SELECT pg_catalog.pg_is_in_recovery();` + getDatabasesSQL = `SELECT datname, pg_get_userbyid(datdba) AS owner FROM pg_database;` getSchemasSQL = `SELECT n.nspname AS dbschema FROM pg_catalog.pg_namespace n WHERE n.nspname !~ '^pg_' AND n.nspname <> 'information_schema' ORDER BY 1` @@ -199,6 +201,14 @@ func (c *Cluster) closeDbConn() (err error) { return nil } +func (c *Cluster) isInRecovery() (bool, error) { + var inRecovery bool + if err := c.pgDb.QueryRow(isInRecoverySQL).Scan(&inRecovery); err != nil { + return false, fmt.Errorf("could not check if database is in recovery: %v", err) + } + return inRecovery, nil +} + func (c *Cluster) readPgUsersFromDatabase(userNames []string) (users spec.PgUserMap, err error) { c.setProcessName("reading users from the database") var rows *sql.Rows diff --git a/pkg/cluster/sync.go b/pkg/cluster/sync.go index 664c67b7e..8bb21b639 100644 --- a/pkg/cluster/sync.go +++ b/pkg/cluster/sync.go @@ -17,6 +17,7 @@ import ( "github.com/zalando/postgres-operator/v2/pkg/util" "github.com/zalando/postgres-operator/v2/pkg/util/constants" "github.com/zalando/postgres-operator/v2/pkg/util/k8sutil" + "github.com/zalando/postgres-operator/v2/pkg/util/retryutil" batchv1 "k8s.io/api/batch/v1" v1 "k8s.io/api/core/v1" policyv1 "k8s.io/api/policy/v1" @@ -1418,11 +1419,7 @@ func (c *Cluster) updatePgUser(secretUsername string, degraded bool) { func (c *Cluster) syncRoles() (err error) { c.setProcessName("syncing roles") - var ( - dbUsers spec.PgUserMap - newUsers spec.PgUserMap - userNames []string - ) + var userNames []string err = c.initDbConn() if err != nil { @@ -1441,7 +1438,6 @@ func (c *Cluster) syncRoles() (err error) { // mapping between original role name and with deletion suffix deletedUsers := map[string]string{} - newUsers = make(map[string]spec.PgUser) // create list of database roles to query for _, u := range c.pgUsers { @@ -1471,50 +1467,93 @@ func (c *Cluster) syncRoles() (err error) { // search also for system users for _, systemUser := range c.systemUsers { userNames = append(userNames, systemUser.Name) - newUsers[systemUser.Name] = systemUser } - dbUsers, err = c.readPgUsersFromDatabase(userNames) + // read the current state of roles and compute the sync requests on every + // attempt, since a previous attempt might have partially applied them + var lastErr error + err = retryutil.Retry( + constants.PostgresConnectTimeout, + constants.PostgresConnectRetryTimeout, + func() (bool, error) { + inRecovery, err := c.isInRecovery() + if err != nil { + lastErr = err // connection problem / startup: transient + } else if inRecovery { + lastErr = fmt.Errorf("database is in recovery") + } else if lastErr = c.executeRolesSync(userNames, deletedUsers); lastErr != nil { + // the failure is only transient if the database got demoted meanwhile + if inRecovery, rerr := c.isInRecovery(); rerr == nil && !inRecovery { + return false, lastErr // permanent error (e.g. rename conflict): fail fast + } + } + if lastErr != nil { + c.logger.Warningf("could not sync roles, retrying: %v", lastErr) + return false, nil + } + return true, nil + }) + if err != nil { + return fmt.Errorf("error executing sync statements: %v", err) + } + + return nil +} + +// executeRolesSync reads the given roles from the database, compares them with +// the desired roles and executes the resulting sync requests +func (c *Cluster) executeRolesSync(userNames []string, deletedUsers map[string]string) error { + dbUsers, err := c.readPgUsersFromDatabase(userNames) if err != nil { return fmt.Errorf("error getting users from the database: %v", err) } + newUsers := c.buildNewUsers(dbUsers, deletedUsers) + pgSyncRequests := c.userSyncStrategy.ProduceSyncRequests(dbUsers, newUsers) + return c.userSyncStrategy.ExecuteSyncRequests(pgSyncRequests, c.pgDb) +} + +// buildNewUsers returns the desired roles to send to ProduceSyncRequests, adjusted +// to the current state of dbUsers. It works on copies of the cluster's users, so +// that repeated attempts always start from the same desired state. Group roles of +// rotation users are removed from dbUsers. +func (c *Cluster) buildNewUsers(dbUsers spec.PgUserMap, deletedUsers map[string]string) spec.PgUserMap { + newUsers := make(spec.PgUserMap) + for _, systemUser := range c.systemUsers { + newUsers[systemUser.Name] = systemUser + } + for _, pgUser := range c.pgUsers { + newUsers[pgUser.Name] = pgUser + } + DBUSERS: for _, dbUser := range dbUsers { // copy rolconfig to rotation users - for pgUserName, pgUser := range c.pgUsers { + for _, pgUser := range c.pgUsers { if pgUser.Rotated && pgUser.MemberOf[0] == dbUser.Name { - pgUser.Parameters = dbUser.Parameters - c.pgUsers[pgUserName] = pgUser + rotationUser := newUsers[pgUser.Name] + rotationUser.Parameters = dbUser.Parameters + newUsers[pgUser.Name] = rotationUser // remove group role from dbUsers to not count as deleted role delete(dbUsers, dbUser.Name) continue DBUSERS } } - // update pgUsers where a deleted role was found + // update newUsers where a deleted role was found // so that they are skipped in ProduceSyncRequests originalUsername, foundDeletedUser := deletedUsers[dbUser.Name] // check if original user does not exist in dbUsers _, originalUserAlreadyExists := dbUsers[originalUsername] if foundDeletedUser && !originalUserAlreadyExists { - recreatedUser := c.pgUsers[originalUsername] - recreatedUser.Deleted = true - c.pgUsers[originalUsername] = recreatedUser + if recreatedUser, exists := newUsers[originalUsername]; exists { + recreatedUser.Deleted = true + newUsers[originalUsername] = recreatedUser + } } } - // last but not least copy pgUsers to newUsers to send to ProduceSyncRequests - for _, pgUser := range c.pgUsers { - newUsers[pgUser.Name] = pgUser - } - - pgSyncRequests := c.userSyncStrategy.ProduceSyncRequests(dbUsers, newUsers) - if err = c.userSyncStrategy.ExecuteSyncRequests(pgSyncRequests, c.pgDb); err != nil { - return fmt.Errorf("error executing sync statements: %v", err) - } - - return nil + return newUsers } func (c *Cluster) syncDatabases() error { diff --git a/pkg/cluster/sync_test.go b/pkg/cluster/sync_test.go index 1c705ea44..58f308c8c 100644 --- a/pkg/cluster/sync_test.go +++ b/pkg/cluster/sync_test.go @@ -1053,3 +1053,40 @@ func TestUpdateSecretNameConflict(t *testing.T) { expectedError := fmt.Sprintf("syncing secret %s failed: error while checking for password rotation: could not update secret because of user name mismatch", "default/prepared-owner-user.acid-test-cluster.credentials") assert.Contains(t, err.Error(), expectedError) } + +func TestBuildNewUsers(t *testing.T) { + cluster := &Cluster{ + systemUsers: map[string]spec.PgUser{ + constants.SuperuserKeyName: {Name: "postgres"}, + }, + pgUsers: map[string]spec.PgUser{ + "alice": {Name: "alice", Origin: spec.RoleOriginTeamsAPI}, + "foo": {Name: "foo260927", Rotated: true, MemberOf: []string{"foo"}}, + }, + } + deletedUsers := map[string]string{"alice_deleted": "alice"} + + // first attempt: only the deprecated team member role exists in the database + dbUsers := spec.PgUserMap{ + "alice_deleted": {Name: "alice_deleted", Deleted: true}, + "foo": {Name: "foo", Parameters: map[string]string{"search_path": "data"}}, + } + newUsers := cluster.buildNewUsers(dbUsers, deletedUsers) + + assert.Contains(t, newUsers, "postgres") + assert.True(t, newUsers["alice"].Deleted, "alice should be skipped while alice_deleted is renamed back") + assert.Equal(t, map[string]string{"search_path": "data"}, newUsers["foo260927"].Parameters) + assert.NotContains(t, dbUsers, "foo", "group role of rotation user should be removed from dbUsers") + + // cluster users must stay untouched so that retries start from the same state + assert.False(t, cluster.pgUsers["alice"].Deleted) + assert.Nil(t, cluster.pgUsers["foo"].Parameters) + + // retry after the rename was applied: alice must be synced again + dbUsers = spec.PgUserMap{ + "alice": {Name: "alice"}, + "foo": {Name: "foo"}, + } + newUsers = cluster.buildNewUsers(dbUsers, deletedUsers) + assert.False(t, newUsers["alice"].Deleted, "alice should not be skipped once the role exists again") +}