Ensure podAnnotations are removed from pods if reset in the config

pkg/cluster/cluster.go

@@ -105,10 +105,11 @@ type Cluster struct {
 }
 
 type compareStatefulsetResult struct {
-	match         bool
-	replace       bool
-	rollingUpdate bool
-	reasons       []string
+	match                 bool
+	replace               bool
+	rollingUpdate         bool
+	reasons               []string
+	deletedPodAnnotations []string
 }
 
 // New creates a new cluster. This function should be called from a controller.
@@ -431,6 +432,7 @@ func (c *Cluster) Create() (err error) {
 }
 
 func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compareStatefulsetResult {
+	deletedPodAnnotations := []string{}
 	reasons := make([]string, 0)
 	var match, needsRollUpdate, needsReplace bool
 
@@ -445,7 +447,7 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 		needsReplace = true
 		reasons = append(reasons, "new statefulset's ownerReferences do not match")
 	}
-	if changed, reason := c.compareAnnotations(c.Statefulset.Annotations, statefulSet.Annotations); changed {
+	if changed, reason := c.compareAnnotations(c.Statefulset.Annotations, statefulSet.Annotations, nil); changed {
 		match = false
 		needsReplace = true
 		reasons = append(reasons, "new statefulset's annotations do not match: "+reason)
@@ -519,7 +521,7 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 		}
 	}
 
-	if changed, reason := c.compareAnnotations(c.Statefulset.Spec.Template.Annotations, statefulSet.Spec.Template.Annotations); changed {
+	if changed, reason := c.compareAnnotations(c.Statefulset.Spec.Template.Annotations, statefulSet.Spec.Template.Annotations, &deletedPodAnnotations); changed {
 		match = false
 		needsReplace = true
 		reasons = append(reasons, "new statefulset's pod template metadata annotations does not match "+reason)
@@ -541,7 +543,7 @@ 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 changed, reason := c.compareAnnotations(c.Statefulset.Spec.VolumeClaimTemplates[i].Annotations, statefulSet.Spec.VolumeClaimTemplates[i].Annotations); changed {
+			if changed, reason := c.compareAnnotations(c.Statefulset.Spec.VolumeClaimTemplates[i].Annotations, statefulSet.Spec.VolumeClaimTemplates[i].Annotations, nil); changed {
 				needsReplace = true
 				reasons = append(reasons, fmt.Sprintf("new statefulset's annotations for volume %q do not match the current ones: %s", name, reason))
 			}
@@ -579,7 +581,7 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 		match = false
 	}
 
-	return &compareStatefulsetResult{match: match, reasons: reasons, rollingUpdate: needsRollUpdate, replace: needsReplace}
+	return &compareStatefulsetResult{match: match, reasons: reasons, rollingUpdate: needsRollUpdate, replace: needsReplace, deletedPodAnnotations: deletedPodAnnotations}
 }
 
 type containerCondition func(a, b v1.Container) bool
@@ -781,7 +783,7 @@ func volumeMountExists(mount v1.VolumeMount, mounts []v1.VolumeMount) bool {
 	return false
 }
 
-func (c *Cluster) compareAnnotations(old, new map[string]string) (bool, string) {
+func (c *Cluster) compareAnnotations(old, new map[string]string, removedList *[]string) (bool, string) {
 	reason := ""
 	ignoredAnnotations := make(map[string]bool)
 	for _, ignore := range c.OpConfig.IgnoredAnnotations {
@@ -794,6 +796,9 @@ func (c *Cluster) compareAnnotations(old, new map[string]string) (bool, string)
 		}
 		if _, ok := new[key]; !ok {
 			reason += fmt.Sprintf(" Removed %q.", key)
+			if removedList != nil {
+				*removedList = append(*removedList, key)
+			}
 		}
 	}
 
@@ -836,7 +841,7 @@ func (c *Cluster) compareServices(old, new *v1.Service) (bool, string) {
 	return true, ""
 }
 
-func (c *Cluster) compareLogicalBackupJob(cur, new *batchv1.CronJob) (match bool, reason string) {
+func (c *Cluster) compareLogicalBackupJob(cur, new *batchv1.CronJob, deletedPodAnnotations *[]string) (match bool, reason string) {
 
 	if cur.Spec.Schedule != new.Spec.Schedule {
 		return false, fmt.Sprintf("new job's schedule %q does not match the current one %q",
@@ -852,8 +857,8 @@ func (c *Cluster) compareLogicalBackupJob(cur, new *batchv1.CronJob) (match bool
 
 	newPodAnnotation := new.Spec.JobTemplate.Spec.Template.Annotations
 	curPodAnnotation := cur.Spec.JobTemplate.Spec.Template.Annotations
-	if changed, reason := c.compareAnnotations(curPodAnnotation, newPodAnnotation); changed {
-		return false, fmt.Sprintf("new job's pod template metadata annotations does not match " + reason)
+	if changed, reason := c.compareAnnotations(curPodAnnotation, newPodAnnotation, deletedPodAnnotations); changed {
+		return false, fmt.Sprint("new job's pod template metadata annotations do not match " + reason)
 	}
 
 	newPgVersion := getPgVersion(new)
@@ -881,7 +886,7 @@ func (c *Cluster) comparePodDisruptionBudget(cur, new *policyv1.PodDisruptionBud
 	if !reflect.DeepEqual(new.ObjectMeta.OwnerReferences, cur.ObjectMeta.OwnerReferences) {
 		return false, "new PDB's owner references do not match the current ones"
 	}
-	if changed, reason := c.compareAnnotations(cur.Annotations, new.Annotations); changed {
+	if changed, reason := c.compareAnnotations(cur.Annotations, new.Annotations, nil); changed {
 		return false, "new PDB's annotations do not match the current ones:" + reason
 	}
 	return true, ""
@@ -1016,7 +1021,7 @@ 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
 
-		annotationsChanged, _ := c.compareAnnotations(oldSpec.Annotations, newSpec.Annotations)
+		annotationsChanged, _ := c.compareAnnotations(oldSpec.Annotations, newSpec.Annotations, nil)
 
 		initUsers := !sameUsers || !sameRotatedUsers || needPoolerUser || needStreamUser
 		if initUsers {

pkg/cluster/cluster_test.go

@@ -1680,7 +1680,7 @@ func TestCompareLogicalBackupJob(t *testing.T) {
 				}
 			}
 
-			match, reason := cluster.compareLogicalBackupJob(currentCronJob, desiredCronJob)
+			match, reason := cluster.compareLogicalBackupJob(currentCronJob, desiredCronJob, nil)
 			if match != tt.match {
 				t.Errorf("%s - unexpected match result %t when comparing cronjobs %#v and %#v", t.Name(), match, currentCronJob, desiredCronJob)
 			} else {

pkg/cluster/connection_pooler.go

@@ -2,6 +2,7 @@ package cluster
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"reflect"
 	"strings"
@@ -977,6 +978,7 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 		err           error
 	)
 
+	updatedPodAnnotations := map[string]*string{}
 	syncReason := make([]string, 0)
 	deployment, err = c.KubeClient.
 		Deployments(c.Namespace).
@@ -1038,9 +1040,27 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 		}
 
 		newPodAnnotations := c.annotationsSet(c.generatePodAnnotations(&c.Spec))
-		if changed, reason := c.compareAnnotations(deployment.Spec.Template.Annotations, newPodAnnotations); changed {
+		deletedPodAnnotations := []string{}
+		if changed, reason := c.compareAnnotations(deployment.Spec.Template.Annotations, newPodAnnotations, &deletedPodAnnotations); changed {
 			specSync = true
 			syncReason = append(syncReason, []string{"new connection pooler's pod template annotations do not match the current ones: " + reason}...)
+
+			for _, anno := range deletedPodAnnotations {
+				updatedPodAnnotations[anno] = nil
+			}
+			templateMetadataReq := map[string]map[string]map[string]map[string]map[string]*string{
+				"spec": {"template": {"metadata": {"annotations": updatedPodAnnotations}}}}
+			patch, err := json.Marshal(templateMetadataReq)
+			if err != nil {
+				return nil, fmt.Errorf("could not marshal ObjectMeta for %s connection pooler's pod template: %v", role, err)
+			}
+			deployment, err = c.KubeClient.Deployments(c.Namespace).Patch(context.TODO(),
+				deployment.Name, types.StrategicMergePatchType, patch, metav1.PatchOptions{}, "")
+			if err != nil {
+				c.logger.Errorf("failed to patch %s connection pooler's pod template: %v", role, err)
+				return nil, err
+			}
+
 			deployment.Spec.Template.Annotations = newPodAnnotations
 		}
 
@@ -1064,7 +1084,7 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 		}
 
 		newAnnotations := c.AnnotationsToPropagate(c.annotationsSet(nil)) // including the downscaling annotations
-		if changed, _ := c.compareAnnotations(deployment.Annotations, newAnnotations); changed {
+		if changed, _ := c.compareAnnotations(deployment.Annotations, newAnnotations, nil); changed {
 			deployment, err = patchConnectionPoolerAnnotations(c.KubeClient, deployment, newAnnotations)
 			if err != nil {
 				return nil, err
@@ -1098,14 +1118,20 @@ 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 {
-			patchData, err := metaAnnotationsPatch(deployment.Spec.Template.Annotations)
+		} else if changed, _ := c.compareAnnotations(pod.Annotations, deployment.Spec.Template.Annotations, nil); changed {
+			metadataReq := map[string]map[string]map[string]*string{"metadata": {}}
+
+			for anno, val := range deployment.Spec.Template.Annotations {
+				updatedPodAnnotations[anno] = &val
+			}
+			metadataReq["metadata"]["annotations"] = updatedPodAnnotations
+			patch, err := json.Marshal(metadataReq)
 			if err != nil {
-				return nil, fmt.Errorf("could not form patch for pooler's pod annotations: %v", err)
+				return nil, fmt.Errorf("could not marshal ObjectMeta for %s connection pooler's pods: %v", role, err)
 			}
-			_, err = c.KubeClient.Pods(pod.Namespace).Patch(context.TODO(), pod.Name, types.MergePatchType, []byte(patchData), metav1.PatchOptions{})
+			_, err = c.KubeClient.Pods(pod.Namespace).Patch(context.TODO(), pod.Name, types.StrategicMergePatchType, patch, metav1.PatchOptions{})
 			if err != nil {
-				return nil, fmt.Errorf("could not patch annotations for pooler's pod %q: %v", pod.Name, err)
+				return nil, fmt.Errorf("could not patch annotations for %s connection pooler's pod %q: %v", role, pod.Name, err)
 			}
 		}
 	}

pkg/cluster/resources.go

@@ -329,7 +329,7 @@ func (c *Cluster) updateService(role PostgresRole, oldService *v1.Service, newSe
 		}
 	}
 
-	if changed, _ := c.compareAnnotations(oldService.Annotations, newService.Annotations); changed {
+	if changed, _ := c.compareAnnotations(oldService.Annotations, newService.Annotations, nil); changed {
 		patchData, err := metaAnnotationsPatch(newService.Annotations)
 		if err != nil {
 			return nil, fmt.Errorf("could not form patch for service %q annotations: %v", oldService.Name, err)

pkg/cluster/streams.go

@@ -545,7 +545,7 @@ func (c *Cluster) compareStreams(curEventStreams, newEventStreams *zalandov1.Fab
 	for newKey, newValue := range newEventStreams.Annotations {
 		desiredAnnotations[newKey] = newValue
 	}
-	if changed, reason := c.compareAnnotations(curEventStreams.ObjectMeta.Annotations, desiredAnnotations); changed {
+	if changed, reason := c.compareAnnotations(curEventStreams.ObjectMeta.Annotations, desiredAnnotations, nil); changed {
 		match = false
 		reasons = append(reasons, fmt.Sprintf("new streams annotations do not match: %s", reason))
 	}

pkg/cluster/sync.go

@@ -230,7 +230,7 @@ func (c *Cluster) syncPatroniConfigMap(suffix string) error {
 		maps.Copy(annotations, cm.Annotations)
 		// Patroni can add extra annotations so incl. current annotations in desired annotations
 		desiredAnnotations := c.annotationsSet(cm.Annotations)
-		if changed, _ := c.compareAnnotations(annotations, desiredAnnotations); changed {
+		if changed, _ := c.compareAnnotations(annotations, desiredAnnotations, nil); changed {
 			patchData, err := metaAnnotationsPatch(desiredAnnotations)
 			if err != nil {
 				return fmt.Errorf("could not form patch for %s config map: %v", configMapName, err)
@@ -275,7 +275,7 @@ func (c *Cluster) syncPatroniEndpoint(suffix string) error {
 		maps.Copy(annotations, ep.Annotations)
 		// Patroni can add extra annotations so incl. current annotations in desired annotations
 		desiredAnnotations := c.annotationsSet(ep.Annotations)
-		if changed, _ := c.compareAnnotations(annotations, desiredAnnotations); changed {
+		if changed, _ := c.compareAnnotations(annotations, desiredAnnotations, nil); changed {
 			patchData, err := metaAnnotationsPatch(desiredAnnotations)
 			if err != nil {
 				return fmt.Errorf("could not form patch for %s endpoint: %v", endpointName, err)
@@ -320,7 +320,7 @@ func (c *Cluster) syncPatroniService() error {
 		maps.Copy(annotations, svc.Annotations)
 		// Patroni can add extra annotations so incl. current annotations in desired annotations
 		desiredAnnotations := c.annotationsSet(svc.Annotations)
-		if changed, _ := c.compareAnnotations(annotations, desiredAnnotations); changed {
+		if changed, _ := c.compareAnnotations(annotations, desiredAnnotations, nil); changed {
 			patchData, err := metaAnnotationsPatch(desiredAnnotations)
 			if err != nil {
 				return fmt.Errorf("could not form patch for %s service: %v", serviceName, err)
@@ -412,7 +412,7 @@ func (c *Cluster) syncEndpoint(role PostgresRole) error {
 				return fmt.Errorf("could not update %s endpoint: %v", role, err)
 			}
 		} else {
-			if changed, _ := c.compareAnnotations(ep.Annotations, desiredEp.Annotations); changed {
+			if changed, _ := c.compareAnnotations(ep.Annotations, desiredEp.Annotations, nil); changed {
 				patchData, err := metaAnnotationsPatch(desiredEp.Annotations)
 				if err != nil {
 					return fmt.Errorf("could not form patch for %s endpoint: %v", role, err)
@@ -561,13 +561,22 @@ func (c *Cluster) syncStatefulSet() error {
 
 		cmp := c.compareStatefulSetWith(desiredSts)
 		if !cmp.rollingUpdate {
+			updatedPodAnnotations := map[string]*string{}
+			for _, anno := range cmp.deletedPodAnnotations {
+				updatedPodAnnotations[anno] = nil
+			}
+			for anno, val := range desiredSts.Spec.Template.Annotations {
+				updatedPodAnnotations[anno] = &val
+			}
+			metadataReq := map[string]map[string]map[string]*string{"metadata": {"annotations": updatedPodAnnotations}}
+			patch, err := json.Marshal(metadataReq)
+			if err != nil {
+				return fmt.Errorf("could not form patch for pod annotations: %v", err)
+			}
+
 			for _, pod := range pods {
-				if changed, _ := c.compareAnnotations(pod.Annotations, desiredSts.Spec.Template.Annotations); changed {
-					patchData, err := metaAnnotationsPatch(desiredSts.Spec.Template.Annotations)
-					if err != nil {
-						return fmt.Errorf("could not form patch for pod %q annotations: %v", pod.Name, err)
-					}
-					_, err = c.KubeClient.Pods(pod.Namespace).Patch(context.TODO(), pod.Name, types.MergePatchType, []byte(patchData), metav1.PatchOptions{})
+				if changed, _ := c.compareAnnotations(pod.Annotations, desiredSts.Spec.Template.Annotations, nil); changed {
+					_, err = c.KubeClient.Pods(c.Namespace).Patch(context.TODO(), pod.Name, types.StrategicMergePatchType, patch, metav1.PatchOptions{})
 					if err != nil {
 						return fmt.Errorf("could not patch annotations for pod %q: %v", pod.Name, err)
 					}
@@ -1142,7 +1151,7 @@ func (c *Cluster) updateSecret(
 		c.Secrets[secret.UID] = secret
 	}
 
-	if changed, _ := c.compareAnnotations(secret.Annotations, generatedSecret.Annotations); changed {
+	if changed, _ := c.compareAnnotations(secret.Annotations, generatedSecret.Annotations, nil); changed {
 		patchData, err := metaAnnotationsPatch(generatedSecret.Annotations)
 		if err != nil {
 			return fmt.Errorf("could not form patch for secret %q annotations: %v", secret.Name, err)
@@ -1587,19 +1596,37 @@ func (c *Cluster) syncLogicalBackupJob() error {
 			}
 			c.logger.Infof("logical backup job %s updated", c.getLogicalBackupJobName())
 		}
-		if match, reason := c.compareLogicalBackupJob(job, desiredJob); !match {
+		deletedPodAnnotations := []string{}
+		if match, reason := c.compareLogicalBackupJob(job, desiredJob, &deletedPodAnnotations); !match {
 			c.logger.Infof("logical job %s is not in the desired state and needs to be updated",
 				c.getLogicalBackupJobName(),
 			)
 			if reason != "" {
 				c.logger.Infof("reason: %s", reason)
 			}
+			if len(deletedPodAnnotations) == 0 {
+				templateMetadataReq := map[string]map[string]map[string]map[string]map[string]map[string]map[string]*string{
+					"spec": {"jobTemplate": {"spec": {"template": {"metadata": {"annotations": {}}}}}}}
+				for _, anno := range deletedPodAnnotations {
+					templateMetadataReq["spec"]["jobTemplate"]["spec"]["template"]["metadata"]["annotations"][anno] = nil
+				}
+				patch, err := json.Marshal(templateMetadataReq)
+				if err != nil {
+					return fmt.Errorf("could not marshal ObjectMeta for logical backup job %q pod template: %v", jobName, err)
+				}
+
+				job, err = c.KubeClient.CronJobs(c.Namespace).Patch(context.TODO(), jobName, types.StrategicMergePatchType, patch, metav1.PatchOptions{}, "")
+				if err != nil {
+					c.logger.Errorf("failed to remove annotations from the logical backup job %q pod template: %v", jobName, err)
+					return err
+				}
+			}
 			if err = c.patchLogicalBackupJob(desiredJob); err != nil {
 				return fmt.Errorf("could not update logical backup job to match desired state: %v", err)
 			}
 			c.logger.Info("the logical backup job is synced")
 		}
-		if changed, _ := c.compareAnnotations(job.Annotations, desiredJob.Annotations); changed {
+		if changed, _ := c.compareAnnotations(job.Annotations, desiredJob.Annotations, nil); changed {
 			patchData, err := metaAnnotationsPatch(desiredJob.Annotations)
 			if err != nil {
 				return fmt.Errorf("could not form patch for the logical backup job %q: %v", jobName, err)