Initial inherited annotations fix implementation

Author: hughcapet

manifests/operator-service-account-rbac.yaml

@@ -173,6 +173,7 @@ rules:
   - get
   - list
   - patch
+  - update
 # to CRUD cron jobs for logical backups
 - apiGroups:
   - batch

pkg/cluster/cluster.go

@@ -30,6 +30,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	v1 "k8s.io/api/core/v1"
+	apipolicyv1 "k8s.io/api/policy/v1"
 	policyv1 "k8s.io/api/policy/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -400,7 +401,7 @@ func (c *Cluster) Create() (err error) {
 
 	if len(c.Spec.Streams) > 0 {
 		// creating streams requires syncing the statefulset first
-		err = c.syncStatefulSet()
+		err = c.syncStatefulSet(true)
 		if err != nil {
 			return fmt.Errorf("could not sync statefulset: %v", err)
 		}
@@ -493,7 +494,6 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 	if changed, reason := c.compareAnnotations(c.Statefulset.Spec.Template.Annotations, statefulSet.Spec.Template.Annotations); changed {
 		match = false
 		needsReplace = true
-		needsRollUpdate = true
 		reasons = append(reasons, "new statefulset's pod template metadata annotations does not match "+reason)
 	}
 	if !reflect.DeepEqual(c.Statefulset.Spec.Template.Spec.SecurityContext, statefulSet.Spec.Template.Spec.SecurityContext) {
@@ -513,9 +513,9 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 				reasons = append(reasons, fmt.Sprintf("new statefulset's name for volume %d does not match the current one", i))
 				continue
 			}
-			if !reflect.DeepEqual(c.Statefulset.Spec.VolumeClaimTemplates[i].Annotations, statefulSet.Spec.VolumeClaimTemplates[i].Annotations) {
+			if changed, reason := c.compareAnnotations(c.Statefulset.Spec.VolumeClaimTemplates[i].Annotations, statefulSet.Spec.VolumeClaimTemplates[i].Annotations); changed {
 				needsReplace = true
-				reasons = append(reasons, fmt.Sprintf("new statefulset's annotations for volume %q does not match the current one", name))
+				reasons = append(reasons, fmt.Sprintf("new statefulset's annotations for volume %q does not match the current one: ", name)+reason)
 			}
 			if !reflect.DeepEqual(c.Statefulset.Spec.VolumeClaimTemplates[i].Spec, statefulSet.Spec.VolumeClaimTemplates[i].Spec) {
 				name := c.Statefulset.Spec.VolumeClaimTemplates[i].Name
@@ -764,6 +764,16 @@ func (c *Cluster) compareAnnotations(old, new map[string]string) (bool, string)
 
 }
 
+func (c *Cluster) extractIgnoredAnnotations(annoList map[string]string) map[string]string {
+	result := make(map[string]string)
+	for _, ignore := range c.OpConfig.IgnoredAnnotations {
+		if _, ok := annoList[ignore]; ok {
+			result[ignore] = annoList[ignore]
+		}
+	}
+	return result
+}
+
 func (c *Cluster) compareServices(old, new *v1.Service) (bool, string) {
 	if old.Spec.Type != new.Spec.Type {
 		return false, fmt.Sprintf("new service's type %q does not match the current one %q",
@@ -818,6 +828,17 @@ func (c *Cluster) compareLogicalBackupJob(cur, new *batchv1.CronJob) (match bool
 	return true, ""
 }
 
+func (c *Cluster) ComparePodDisruptionBudget(cur, new *apipolicyv1.PodDisruptionBudget) (bool, string) {
+	//TODO: improve comparison
+	if match := reflect.DeepEqual(new.Spec, cur.Spec); !match {
+		return false, "new PDB spec does not match the current one"
+	}
+	if changed, reason := c.compareAnnotations(cur.Annotations, new.Annotations); changed {
+		return false, "new PDB's annotations does not match the current one:" + reason
+	}
+	return true, ""
+}
+
 func getPgVersion(cronJob *batchv1.CronJob) string {
 	envs := cronJob.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Env
 	for _, env := range envs {
@@ -922,12 +943,9 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 	}
 
 	// Service
-	if !reflect.DeepEqual(c.generateService(Master, &oldSpec.Spec), c.generateService(Master, &newSpec.Spec)) ||
-		!reflect.DeepEqual(c.generateService(Replica, &oldSpec.Spec), c.generateService(Replica, &newSpec.Spec)) {
-		if err := c.syncServices(); err != nil {
-			c.logger.Errorf("could not sync services: %v", err)
-			updateFailed = true
-		}
+	if err := c.syncServices(); err != nil {
+		c.logger.Errorf("could not sync services: %v", err)
+		updateFailed = true
 	}
 
 	// Users
@@ -946,15 +964,19 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 		// only when streams were not specified in oldSpec but in newSpec
 		needStreamUser := len(oldSpec.Spec.Streams) == 0 && len(newSpec.Spec.Streams) > 0
 
-		if !sameUsers || !sameRotatedUsers || needPoolerUser || needStreamUser {
+		annotationsChanged, _ := c.compareAnnotations(oldSpec.Annotations, newSpec.Annotations)
+
+		initUsers := !sameUsers || !sameRotatedUsers || needPoolerUser || needStreamUser
+		if initUsers {
 			c.logger.Debugf("initialize users")
 			if err := c.initUsers(); err != nil {
 				c.logger.Errorf("could not init users - skipping sync of secrets and databases: %v", err)
 				userInitFailed = true
 				updateFailed = true
 				return
 			}
-
+		}
+		if initUsers || annotationsChanged {
 			c.logger.Debugf("syncing secrets")
 			//TODO: mind the secrets of the deleted/new users
 			if err := c.syncSecrets(); err != nil {
@@ -968,7 +990,7 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 	if c.OpConfig.StorageResizeMode != "off" {
 		c.syncVolumes()
 	} else {
-		c.logger.Infof("Storage resize is disabled (storage_resize_mode is off). Skipping volume sync.")
+		c.logger.Infof("Storage resize is disabled (storage_resize_mode is off). Skipping volume size sync.")
 	}
 
 	// streams configuration
@@ -978,29 +1000,11 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 
 	// Statefulset
 	func() {
-		oldSs, err := c.generateStatefulSet(&oldSpec.Spec)
-		if err != nil {
-			c.logger.Errorf("could not generate old statefulset spec: %v", err)
-			updateFailed = true
-			return
-		}
-
-		newSs, err := c.generateStatefulSet(&newSpec.Spec)
-		if err != nil {
-			c.logger.Errorf("could not generate new statefulset spec: %v", err)
+		if err := c.syncStatefulSet(syncStatefulSet); err != nil {
+			c.logger.Errorf("could not sync statefulsets: %v", err)
 			updateFailed = true
-			return
-		}
-
-		if syncStatefulSet || !reflect.DeepEqual(oldSs, newSs) {
-			c.logger.Debugf("syncing statefulsets")
-			syncStatefulSet = false
-			// TODO: avoid generating the StatefulSet object twice by passing it to syncStatefulSet
-			if err := c.syncStatefulSet(); err != nil {
-				c.logger.Errorf("could not sync statefulsets: %v", err)
-				updateFailed = true
-			}
 		}
+		syncStatefulSet = false
 	}()
 
 	// add or remove standby_cluster section from Patroni config depending on changes in standby section
@@ -1011,12 +1015,9 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 	}
 
 	// pod disruption budget
-	if oldSpec.Spec.NumberOfInstances != newSpec.Spec.NumberOfInstances {
-		c.logger.Debug("syncing pod disruption budgets")
-		if err := c.syncPodDisruptionBudget(true); err != nil {
-			c.logger.Errorf("could not sync pod disruption budget: %v", err)
-			updateFailed = true
-		}
+	if err := c.syncPodDisruptionBudget(true); err != nil {
+		c.logger.Errorf("could not sync pod disruption budget: %v", err)
+		updateFailed = true
 	}
 
 	// logical backup job

pkg/cluster/connection_pooler.go

@@ -10,6 +10,7 @@ import (
 	"github.com/sirupsen/logrus"
 	acidzalando "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do"
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
+	"golang.org/x/exp/maps"
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -691,26 +692,6 @@ func updateConnectionPoolerDeployment(KubeClient k8sutil.KubernetesClient, newDe
 	return deployment, nil
 }
 
-// updateConnectionPoolerAnnotations updates the annotations of connection pooler deployment
-func updateConnectionPoolerAnnotations(KubeClient k8sutil.KubernetesClient, deployment *appsv1.Deployment, annotations map[string]string) (*appsv1.Deployment, error) {
-	patchData, err := metaAnnotationsPatch(annotations)
-	if err != nil {
-		return nil, fmt.Errorf("could not form patch for the connection pooler deployment metadata: %v", err)
-	}
-	result, err := KubeClient.Deployments(deployment.Namespace).Patch(
-		context.TODO(),
-		deployment.Name,
-		types.MergePatchType,
-		[]byte(patchData),
-		metav1.PatchOptions{},
-		"")
-	if err != nil {
-		return nil, fmt.Errorf("could not patch connection pooler annotations %q: %v", patchData, err)
-	}
-	return result, nil
-
-}
-
 // Test if two connection pooler configuration needs to be synced. For simplicity
 // compare not the actual K8S objects, but the configuration itself and request
 // sync if there is any difference.
@@ -1022,6 +1003,25 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 			syncReason = append(syncReason, specReason...)
 		}
 
+		newPodAnnotations := c.annotationsSet(c.generatePodAnnotations(&c.Spec))
+		if changed, reason := c.compareAnnotations(deployment.Spec.Template.Annotations, newPodAnnotations); changed {
+			specSync = true
+			syncReason = append(syncReason, []string{"new connection pooler's pod template annotations do not match the current one: " + reason}...)
+			deployment.Spec.Template.Annotations = newPodAnnotations
+		}
+		newAnnotations := c.AnnotationsToPropagate(c.annotationsSet(nil)) // including the downscaling annotations
+		ignoredAnnotations := c.extractIgnoredAnnotations(deployment.Annotations)
+		maps.Copy(newAnnotations, ignoredAnnotations)
+		if changed, reason := c.compareAnnotations(deployment.Annotations, newAnnotations); changed {
+			deployment.Annotations = newAnnotations
+			c.logger.Infof("new connection pooler deployments's annotations does not match the current one:" + reason)
+			c.logger.Debug("updating connection pooler deployments's annotations")
+			deployment, err = c.KubeClient.Deployments(deployment.Namespace).Update(context.TODO(), deployment, metav1.UpdateOptions{})
+			if err != nil {
+				return nil, fmt.Errorf("could not update connection pooler annotations %q: %v", deployment.Name, err)
+			}
+		}
+
 		defaultsSync, defaultsReason := c.needSyncConnectionPoolerDefaults(&c.Config, newConnectionPooler, deployment)
 		syncReason = append(syncReason, defaultsReason...)
 
@@ -1042,15 +1042,6 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 		}
 	}
 
-	newAnnotations := c.AnnotationsToPropagate(c.annotationsSet(c.ConnectionPooler[role].Deployment.Annotations))
-	if newAnnotations != nil {
-		deployment, err = updateConnectionPoolerAnnotations(c.KubeClient, c.ConnectionPooler[role].Deployment, newAnnotations)
-		if err != nil {
-			return nil, err
-		}
-		c.ConnectionPooler[role].Deployment = deployment
-	}
-
 	// check if pooler pods must be replaced due to secret update
 	listOptions := metav1.ListOptions{
 		LabelSelector: labels.Set(c.connectionPoolerLabels(role, true).MatchLabels).String(),
@@ -1076,22 +1067,27 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 			if err != nil {
 				return nil, fmt.Errorf("could not delete pooler pod: %v", err)
 			}
+		} else if changed, _ := c.compareAnnotations(pod.Annotations, deployment.Spec.Template.Annotations); changed {
+			newAnnotations := c.extractIgnoredAnnotations(pod.Annotations)
+			maps.Copy(newAnnotations, deployment.Spec.Template.Annotations)
+			pod.Annotations = newAnnotations
+			c.logger.Debugf("updating annotations for connection pooler's pod %q", pod.Name)
+			_, err := c.KubeClient.Pods(pod.Namespace).Update(context.TODO(), &pod, metav1.UpdateOptions{})
+			if err != nil {
+				return nil, fmt.Errorf("could not update annotations for pod %q: %v", pod.Name, err)
+			}
 		}
 	}
 
 	if service, err = c.KubeClient.Services(c.Namespace).Get(context.TODO(), c.connectionPoolerName(role), metav1.GetOptions{}); err == nil {
 		c.ConnectionPooler[role].Service = service
 		desiredSvc := c.generateConnectionPoolerService(c.ConnectionPooler[role])
-		if match, reason := c.compareServices(service, desiredSvc); !match {
-			syncReason = append(syncReason, reason)
-			c.logServiceChanges(role, service, desiredSvc, false, reason)
-			newService, err = c.updateService(role, service, desiredSvc)
-			if err != nil {
-				return syncReason, fmt.Errorf("could not update %s service to match desired state: %v", role, err)
-			}
-			c.ConnectionPooler[role].Service = newService
-			c.logger.Infof("%s service %q is in the desired state now", role, util.NameFromMeta(desiredSvc.ObjectMeta))
+		newService, err = c.updateService(role, service, desiredSvc)
+		if err != nil {
+			return syncReason, fmt.Errorf("could not update %s service to match desired state: %v", role, err)
 		}
+		c.ConnectionPooler[role].Service = newService
+		c.logger.Infof("%s service %q is in the desired state now", role, util.NameFromMeta(desiredSvc.ObjectMeta))
 		return NoSync, nil
 	}
 

pkg/cluster/k8sres_test.go

@@ -3306,7 +3306,7 @@ func TestGenerateResourceRequirements(t *testing.T) {
 		cluster.Namespace = namespace
 		_, err := cluster.createStatefulSet()
 		if k8sutil.ResourceAlreadyExists(err) {
-			err = cluster.syncStatefulSet()
+			err = cluster.syncStatefulSet(true)
 		}
 		assert.NoError(t, err)
 

pkg/cluster/resources.go

@@ -291,51 +291,27 @@ func (c *Cluster) updateService(role PostgresRole, oldService *v1.Service, newSe
 		err error
 	)
 
+	match, reason := c.compareServices(oldService, newService)
+	if match {
+		return oldService, nil
+	}
+
+	c.logServiceChanges(role, oldService, newService, false, reason)
 	c.setProcessName("updating %v service", role)
 
 	serviceName := util.NameFromMeta(oldService.ObjectMeta)
 
-	// update the service annotation in order to propagate ELB notation.
-	if len(newService.ObjectMeta.Annotations) > 0 {
-		if annotationsPatchData, err := metaAnnotationsPatch(newService.ObjectMeta.Annotations); err == nil {
-			_, err = c.KubeClient.Services(serviceName.Namespace).Patch(
-				context.TODO(),
-				serviceName.Name,
-				types.MergePatchType,
-				[]byte(annotationsPatchData),
-				metav1.PatchOptions{},
-				"")
-
-			if err != nil {
-				return nil, fmt.Errorf("could not replace annotations for the service %q: %v", serviceName, err)
-			}
-		} else {
-			return nil, fmt.Errorf("could not form patch for the service metadata: %v", err)
-		}
-	}
-
 	// now, patch the service spec, but when disabling LoadBalancers do update instead
 	// patch does not work because of LoadBalancerSourceRanges field (even if set to nil)
 	oldServiceType := oldService.Spec.Type
 	newServiceType := newService.Spec.Type
 	if newServiceType == "ClusterIP" && newServiceType != oldServiceType {
 		newService.ResourceVersion = oldService.ResourceVersion
 		newService.Spec.ClusterIP = oldService.Spec.ClusterIP
-		svc, err = c.KubeClient.Services(serviceName.Namespace).Update(context.TODO(), newService, metav1.UpdateOptions{})
-		if err != nil {
-			return nil, fmt.Errorf("could not update service %q: %v", serviceName, err)
-		}
-	} else {
-		patchData, err := specPatch(newService.Spec)
-		if err != nil {
-			return nil, fmt.Errorf("could not form patch for the service %q: %v", serviceName, err)
-		}
-
-		svc, err = c.KubeClient.Services(serviceName.Namespace).Patch(
-			context.TODO(), serviceName.Name, types.MergePatchType, patchData, metav1.PatchOptions{}, "")
-		if err != nil {
-			return nil, fmt.Errorf("could not patch service %q: %v", serviceName, err)
-		}
+	}
+	svc, err = c.KubeClient.Services(serviceName.Namespace).Update(context.TODO(), newService, metav1.UpdateOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("could not update service %q: %v", serviceName, err)
 	}
 
 	return svc, nil