Propagate annotations to the StatefulSet (#932)

* Initial commit

* Corrections

- set the type of the new  configuration parameter to be array of
  strings
- propagate the annotations to statefulset at sync

* Enable regular expression matching

* Improvements

-handle rollingUpdate flag
-modularize code
-rename config parameter name

* fix merge error

* Pass annotations to connection pooler deployment

* update code-gen

* Add documentation and update manifests

* add e2e test and introduce option in configmap

* fix service annotations test

* Add unit test

* fix e2e tests

* better key lookup of annotations tests

* add debug message for annotation tests

* Fix typos

* minor fix for looping

* Handle update path and renaming

- handle the update path to update sts and connection pooler deployment.
  This way no need to wait for sync
- rename the parameter to downscaler_annotations
- handle other review comments

* another try to fix python loops

* Avoid unneccessary update events

* Update manifests

* some final polishing

* fix cluster_test after polishing

Co-authored-by: Rafia Sabih <rafia.sabih@zalando.de>
Co-authored-by: Felix Kunde <felix-kunde@gmx.de>

pkg/cluster/cluster.go

@@ -711,8 +711,7 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 			updateFailed = true
 			return
 		}
-
-		if !reflect.DeepEqual(oldSs, newSs) {
+		if !reflect.DeepEqual(oldSs, newSs) || !reflect.DeepEqual(oldSpec.Annotations, newSpec.Annotations) {
 			c.logger.Debugf("syncing statefulsets")
 			// TODO: avoid generating the StatefulSet object twice by passing it to syncStatefulSet
 			if err := c.syncStatefulSet(); err != nil {

pkg/cluster/cluster_test.go

@@ -28,19 +28,48 @@ var eventRecorder = record.NewFakeRecorder(1)
 var cl = New(
 	Config{
 		OpConfig: config.Config{
-			ProtectedRoles: []string{"admin"},
+			PodManagementPolicy: "ordered_ready",
+			ProtectedRoles:      []string{"admin"},
 			Auth: config.Auth{
 				SuperUsername:       superUserName,
 				ReplicationUsername: replicationUserName,
 			},
+			Resources: config.Resources{
+				DownscalerAnnotations: []string{"downscaler/*"},
+			},
 		},
 	},
 	k8sutil.NewMockKubernetesClient(),
-	acidv1.Postgresql{ObjectMeta: metav1.ObjectMeta{Name: "acid-test", Namespace: "test"}},
+	acidv1.Postgresql{ObjectMeta: metav1.ObjectMeta{Name: "acid-test", Namespace: "test", Annotations: map[string]string{"downscaler/downtime_replicas": "0"}}},
 	logger,
 	eventRecorder,
 )
 
+func TestStatefulSetAnnotations(t *testing.T) {
+	testName := "CheckStatefulsetAnnotations"
+	spec := acidv1.PostgresSpec{
+		TeamID: "myapp", NumberOfInstances: 1,
+		Resources: acidv1.Resources{
+			ResourceRequests: acidv1.ResourceDescription{CPU: "1", Memory: "10"},
+			ResourceLimits:   acidv1.ResourceDescription{CPU: "1", Memory: "10"},
+		},
+		Volume: acidv1.Volume{
+			Size: "1G",
+		},
+	}
+	ss, err := cl.generateStatefulSet(&spec)
+	if err != nil {
+		t.Errorf("in %s no statefulset created %v", testName, err)
+	}
+	if ss != nil {
+		annotation := ss.ObjectMeta.GetAnnotations()
+		if _, ok := annotation["downscaler/downtime_replicas"]; !ok {
+			t.Errorf("in %s respective annotation not found on sts", testName)
+		}
+	}
+
+}
+
 func TestInitRobotUsers(t *testing.T) {
 	testName := "TestInitRobotUsers"
 	tests := []struct {

pkg/cluster/k8sres.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"path"
 	"sort"
+	"strconv"
 
 	"github.com/sirupsen/logrus"
 
@@ -1182,12 +1183,15 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 		return nil, fmt.Errorf("could not set the pod management policy to the unknown value: %v", c.OpConfig.PodManagementPolicy)
 	}
 
+	annotations = make(map[string]string)
+	annotations[rollingUpdateStatefulsetAnnotationKey] = strconv.FormatBool(false)
+
 	statefulSet := &appsv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        c.statefulSetName(),
 			Namespace:   c.Namespace,
 			Labels:      c.labelsSet(true),
-			Annotations: map[string]string{rollingUpdateStatefulsetAnnotationKey: "false"},
+			Annotations: c.AnnotationsToPropagate(annotations),
 		},
 		Spec: appsv1.StatefulSetSpec{
 			Replicas:             &numberOfInstances,

pkg/cluster/resources.go

@@ -853,3 +853,24 @@ func (c *Cluster) updateConnectionPoolerDeployment(oldDeploymentSpec, newDeploym
 
 	return deployment, nil
 }
+
+//updateConnectionPoolerAnnotations updates the annotations of connection pooler deployment
+func (c *Cluster) updateConnectionPoolerAnnotations(annotations map[string]string) (*appsv1.Deployment, error) {
+	c.logger.Debugf("updating connection pooler annotations")
+	patchData, err := metaAnnotationsPatch(annotations)
+	if err != nil {
+		return nil, fmt.Errorf("could not form patch for the deployment metadata: %v", err)
+	}
+	result, err := c.KubeClient.Deployments(c.ConnectionPooler.Deployment.Namespace).Patch(
+		context.TODO(),
+		c.ConnectionPooler.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
+
+}

pkg/cluster/sync.go

@@ -3,6 +3,7 @@ package cluster
 import (
 	"context"
 	"fmt"
+	"regexp"
 	"strings"
 
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
@@ -358,6 +359,8 @@ func (c *Cluster) syncStatefulSet() error {
 				}
 			}
 		}
+		annotations := c.AnnotationsToPropagate(c.Statefulset.Annotations)
+		c.updateStatefulSetAnnotations(annotations)
 
 		if !podsRollingUpdateRequired && !c.OpConfig.EnableLazySpiloUpgrade {
 			// even if desired and actual statefulsets match
@@ -397,6 +400,30 @@ func (c *Cluster) syncStatefulSet() error {
 	return nil
 }
 
+// AnnotationsToPropagate get the annotations to update if required
+// based on the annotations in postgres CRD
+func (c *Cluster) AnnotationsToPropagate(annotations map[string]string) map[string]string {
+	toPropagateAnnotations := c.OpConfig.DownscalerAnnotations
+	pgCRDAnnotations := c.Postgresql.ObjectMeta.GetAnnotations()
+
+	if toPropagateAnnotations != nil && pgCRDAnnotations != nil {
+		for _, anno := range toPropagateAnnotations {
+			for k, v := range pgCRDAnnotations {
+				matched, err := regexp.MatchString(anno, k)
+				if err != nil {
+					c.logger.Errorf("annotations matching issue: %v", err)
+					return nil
+				}
+				if matched {
+					annotations[k] = v
+				}
+			}
+		}
+	}
+
+	return annotations
+}
+
 // checkAndSetGlobalPostgreSQLConfiguration checks whether cluster-wide API parameters
 // (like max_connections) has changed and if necessary sets it via the Patroni API
 func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration() error {
@@ -939,6 +966,11 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 		}
 	}
 
+	newAnnotations := c.AnnotationsToPropagate(c.ConnectionPooler.Deployment.Annotations)
+	if newAnnotations != nil {
+		c.updateConnectionPoolerAnnotations(newAnnotations)
+	}
+
 	service, err := c.KubeClient.
 		Services(c.Namespace).
 		Get(context.TODO(), c.connectionPoolerName(), metav1.GetOptions{})