Rename roles that are removed from PostgresTeam CRD (#1457)

* rename db roles that are removed from manifests

* extend PostgresTeam e2e test

* make suffix configurable and add deprecated field to pgUser struct

* deny LOGIN from deprecated roles

* update feature documentation

pkg/cluster/cluster.go

@@ -74,6 +74,7 @@ type Cluster struct {
 	eventRecorder    record.EventRecorder
 	patroni          patroni.Interface
 	pgUsers          map[string]spec.PgUser
+	pgUsersCache     map[string]spec.PgUser
 	systemUsers      map[string]spec.PgUser
 	podSubscribers   map[spec.NamespacedName]chan PodEvent
 	podSubscribersMu sync.RWMutex
@@ -129,7 +130,9 @@ func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec acidv1.Postgres
 			Secrets:   make(map[types.UID]*v1.Secret),
 			Services:  make(map[PostgresRole]*v1.Service),
 			Endpoints: make(map[PostgresRole]*v1.Endpoints)},
-		userSyncStrategy:    users.DefaultUserSyncStrategy{PasswordEncryption: passwordEncryption},
+		userSyncStrategy: users.DefaultUserSyncStrategy{
+			PasswordEncryption: passwordEncryption,
+			RoleDeletionSuffix: cfg.OpConfig.RoleDeletionSuffix},
 		deleteOptions:       metav1.DeleteOptions{PropagationPolicy: &deletePropagationPolicy},
 		podEventsQueue:      podEventsQueue,
 		KubeClient:          kubeClient,
@@ -190,6 +193,17 @@ func (c *Cluster) isNewCluster() bool {
 func (c *Cluster) initUsers() error {
 	c.setProcessName("initializing users")
 
+	// if team member deprecation is enabled save current state of pgUsers
+	// to check for deleted roles
+	c.pgUsersCache = map[string]spec.PgUser{}
+	if c.OpConfig.EnableTeamMemberDeprecation {
+		for k, v := range c.pgUsers {
+			if v.Origin == spec.RoleOriginTeamsAPI {
+				c.pgUsersCache[k] = v
+			}
+		}
+	}
+
 	// clear our the previous state of the cluster users (in case we are
 	// running a sync).
 	c.systemUsers = map[string]spec.PgUser{}
@@ -650,7 +664,7 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 	needConnectionPooler := needMasterConnectionPoolerWorker(&newSpec.Spec) ||
 		needReplicaConnectionPoolerWorker(&newSpec.Spec)
 	if !sameUsers || needConnectionPooler {
-		c.logger.Debugf("syncing secrets")
+		c.logger.Debugf("initialize users")
 		if err := c.initUsers(); err != nil {
 			c.logger.Errorf("could not init users: %v", err)
 			updateFailed = true

pkg/cluster/database.go

@@ -198,6 +198,7 @@ func (c *Cluster) readPgUsersFromDatabase(userNames []string) (users spec.PgUser
 			rolname, rolpassword                                          string
 			rolsuper, rolinherit, rolcreaterole, rolcreatedb, rolcanlogin bool
 			roloptions, memberof                                          []string
+			roldeleted                                                    bool
 		)
 		err := rows.Scan(&rolname, &rolpassword, &rolsuper, &rolinherit,
 			&rolcreaterole, &rolcreatedb, &rolcanlogin, pq.Array(&roloptions), pq.Array(&memberof))
@@ -216,7 +217,11 @@ func (c *Cluster) readPgUsersFromDatabase(userNames []string) (users spec.PgUser
 			parameters[fields[0]] = fields[1]
 		}
 
-		users[rolname] = spec.PgUser{Name: rolname, Password: rolpassword, Flags: flags, MemberOf: memberof, Parameters: parameters}
+		if strings.HasSuffix(rolname, c.OpConfig.RoleDeletionSuffix) {
+			roldeleted = true
+		}
+
+		users[rolname] = spec.PgUser{Name: rolname, Password: rolpassword, Flags: flags, MemberOf: memberof, Parameters: parameters, Deleted: roldeleted}
 	}
 
 	return users, nil

pkg/cluster/sync.go

@@ -551,10 +551,29 @@ func (c *Cluster) syncRoles() (err error) {
 		}
 	}()
 
+	// mapping between original role name and with deletion suffix
+	deletedUsers := map[string]string{}
+
+	// create list of database roles to query
 	for _, u := range c.pgUsers {
 		userNames = append(userNames, u.Name)
+		// add team member role name with rename suffix in case we need to rename it back
+		if u.Origin == spec.RoleOriginTeamsAPI && c.OpConfig.EnableTeamMemberDeprecation {
+			deletedUsers[u.Name+c.OpConfig.RoleDeletionSuffix] = u.Name
+			userNames = append(userNames, u.Name+c.OpConfig.RoleDeletionSuffix)
+		}
 	}
 
+	// add team members that exist only in cache
+	// to trigger a rename of the role in ProduceSyncRequests
+	for _, cachedUser := range c.pgUsersCache {
+		if _, exists := c.pgUsers[cachedUser.Name]; !exists {
+			userNames = append(userNames, cachedUser.Name)
+		}
+	}
+
+	// add pooler user to list of pgUsers, too
+	// to check if the pooler user exists or has to be created
 	if needMasterConnectionPooler(&c.Spec) || needReplicaConnectionPooler(&c.Spec) {
 		connectionPoolerUser := c.systemUsers[constants.ConnectionPoolerUserKeyName]
 		userNames = append(userNames, connectionPoolerUser.Name)
@@ -569,6 +588,16 @@ func (c *Cluster) syncRoles() (err error) {
 		return fmt.Errorf("error getting users from the database: %v", err)
 	}
 
+	// update pgUsers where a deleted role was found
+	// so that they are skipped in ProduceSyncRequests
+	for _, dbUser := range dbUsers {
+		if originalUser, exists := deletedUsers[dbUser.Name]; exists {
+			recreatedUser := c.pgUsers[originalUser]
+			recreatedUser.Deleted = true
+			c.pgUsers[originalUser] = recreatedUser
+		}
+	}
+
 	pgSyncRequests := c.userSyncStrategy.ProduceSyncRequests(dbUsers, c.pgUsers)
 	if err = c.userSyncStrategy.ExecuteSyncRequests(pgSyncRequests, c.pgDb); err != nil {
 		return fmt.Errorf("error executing sync statements: %v", err)

pkg/cluster/util.go

@@ -242,7 +242,7 @@ func (c *Cluster) getTeamMembers(teamID string) ([]string, error) {
 		for team, membership := range *c.Config.PgTeamMap {
 			if team == teamID {
 				additionalMembers = membership.AdditionalMembers
-				c.logger.Debugf("found %d additional members for team %q", len(members), teamID)
+				c.logger.Debugf("found %d additional members for team %q", len(additionalMembers), teamID)
 			}
 		}
 
@@ -256,14 +256,12 @@ func (c *Cluster) getTeamMembers(teamID string) ([]string, error) {
 
 	token, err := c.oauthTokenGetter.getOAuthToken()
 	if err != nil {
-		c.logger.Warnf("could not get oauth token to authenticate to team service API, only returning %d members for team %q: %v", len(members), teamID, err)
-		return members, nil
+		return nil, fmt.Errorf("could not get oauth token to authenticate to team service API: %v", err)
 	}
 
 	teamInfo, err := c.teamsAPIClient.TeamInfo(teamID, token)
 	if err != nil {
-		c.logger.Warnf("could not get team info for team %q, only returning %d members: %v", teamID, len(members), err)
-		return members, nil
+		return nil, fmt.Errorf("could not get team info for team %q: %v", teamID, err)
 	}
 
 	for _, member := range teamInfo.Members {