merge with master

charts/postgres-operator/crds/operatorconfigurations.yaml

@@ -117,6 +117,10 @@ spec:
                   type: object
                   additionalProperties:
                     type: string
+                downscaler_annotations:
+                  type: array
+                  items:
+                    type: string
                 enable_init_containers:
                   type: boolean
                 enable_pod_antiaffinity:

charts/postgres-operator/values-crd.yaml

@@ -67,6 +67,11 @@ configKubernetes:
   #   keya: valuea
   #   keyb: valueb
 
+  # list of annotations propagated from cluster manifest to statefulset and deployment 
+  # downscaler_annotations:
+  #   - deployment-time
+  #   - downscaler/*
+
   # enables initContainers to run actions before Spilo is started
   enable_init_containers: true
   # toggles pod anti affinity on the Postgres pods
@@ -262,11 +267,11 @@ configConnectionPooler:
   # docker image
   connection_pooler_image: "registry.opensource.zalan.do/acid/pgbouncer"
   # max db connections the pooler should hold
-  connection_pooler_max_db_connections: "60"
+  connection_pooler_max_db_connections: 60
   # default pooling mode
   connection_pooler_mode: "transaction"
   # number of pooler instances
-  connection_pooler_number_of_instances: "2"
+  connection_pooler_number_of_instances: 2
   # default resources
   connection_pooler_default_cpu_request: 500m
   connection_pooler_default_memory_request: 100Mi

charts/postgres-operator/values.yaml

@@ -63,6 +63,9 @@ configKubernetes:
   # annotations attached to each database pod
   # custom_pod_annotations: "keya:valuea,keyb:valueb"
 
+  # list of annotations propagated from cluster manifest to statefulset and deployment
+  # downscaler_annotations: "deployment-time,downscaler/*"
+
   # enables initContainers to run actions before Spilo is started
   enable_init_containers: "true"
   # toggles pod anti affinity on the Postgres pods

docs/reference/operator_parameters.md

@@ -200,6 +200,12 @@ configuration they are grouped under the `kubernetes` key.
   of a database created by the operator. If the annotation key is also provided
   by the database definition, the database definition value is used.
 
+* **downscaler_annotations**
+  An array of annotations that should be passed from Postgres CRD on to the
+  statefulset and, if exists, to the connection pooler deployment as well.
+  Regular expressions like `downscaler/*` etc. are also accepted. Can be used
+  with [kube-downscaler](https://github.com/hjacobs/kube-downscaler).
+
 * **watched_namespace**
   The operator watches for Postgres objects in the given namespace. If not
   specified, the value is taken from the operator namespace. A special `*`

e2e/tests/test_e2e.py

@@ -428,7 +428,7 @@ class EndToEndTestCase(unittest.TestCase):
         k8s.api.core_v1.patch_node(current_master_node, patch_readiness_label)
 
         # wait a little before proceeding with the pod distribution test
-        time.sleep(k8s.RETRY_TIMEOUT_SEC)
+        time.sleep(30)
 
         # toggle pod anti affinity to move replica away from master node
         self.assert_distributed_pods(new_master_node, new_replica_nodes, cluster_label)
@@ -465,21 +465,24 @@ class EndToEndTestCase(unittest.TestCase):
         pg_patch_custom_annotations = {
             "spec": {
                 "serviceAnnotations": {
-                    "annotation.key": "value"
+                    "annotation.key": "value",
+                    "foo": "bar",
                 }
             }
         }
         k8s.api.custom_objects_api.patch_namespaced_custom_object(
             "acid.zalan.do", "v1", "default", "postgresqls", "acid-minimal-cluster", pg_patch_custom_annotations)
 
+        # wait a little before proceeding
+        time.sleep(30)
         annotations = {
             "annotation.key": "value",
             "foo": "bar",
         }
         self.assertTrue(k8s.check_service_annotations(
-            "cluster-name=acid-service-annotations,spilo-role=master", annotations))
+            "cluster-name=acid-minimal-cluster,spilo-role=master", annotations))
         self.assertTrue(k8s.check_service_annotations(
-            "cluster-name=acid-service-annotations,spilo-role=replica", annotations))
+            "cluster-name=acid-minimal-cluster,spilo-role=replica", annotations))
 
         # clean up
         unpatch_custom_service_annotations = {
@@ -489,6 +492,40 @@ class EndToEndTestCase(unittest.TestCase):
         }
         k8s.update_config(unpatch_custom_service_annotations)
 
+    @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
+    def test_statefulset_annotation_propagation(self):
+        '''
+           Inject annotation to Postgresql CRD and check it's propagation to stateful set
+        '''
+        k8s = self.k8s
+        cluster_label = 'application=spilo,cluster-name=acid-minimal-cluster'
+
+        patch_sset_propagate_annotations = {
+            "data": {
+                "downscaler_annotations": "deployment-time,downscaler/*",
+            }
+        }
+        k8s.update_config(patch_sset_propagate_annotations)
+
+        pg_crd_annotations = {
+            "metadata": {
+                "annotations": {
+                    "deployment-time": "2020-04-30 12:00:00",
+                    "downscaler/downtime_replicas": "0",
+                },
+            }
+        }
+        k8s.api.custom_objects_api.patch_namespaced_custom_object(
+            "acid.zalan.do", "v1", "default", "postgresqls", "acid-minimal-cluster", pg_crd_annotations)
+
+        # wait a little before proceeding
+        time.sleep(60)
+        annotations = {
+            "deployment-time": "2020-04-30 12:00:00",
+            "downscaler/downtime_replicas": "0",
+        }
+        self.assertTrue(k8s.check_statefulset_annotations(cluster_label, annotations))
+
     @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
     def test_taint_based_eviction(self):
         '''
@@ -528,7 +565,7 @@ class EndToEndTestCase(unittest.TestCase):
         k8s.update_config(patch_toleration_config)
 
         # wait a little before proceeding with the pod distribution test
-        time.sleep(k8s.RETRY_TIMEOUT_SEC)
+        time.sleep(30)
 
         # toggle pod anti affinity to move replica away from master node
         self.assert_distributed_pods(new_master_node, new_replica_nodes, cluster_label)
@@ -694,10 +731,18 @@ class K8s:
     def check_service_annotations(self, svc_labels, annotations, namespace='default'):
         svcs = self.api.core_v1.list_namespaced_service(namespace, label_selector=svc_labels, limit=1).items
         for svc in svcs:
-            if len(svc.metadata.annotations) != len(annotations):
+            for key, value in annotations.items():
+                if key not in svc.metadata.annotations or svc.metadata.annotations[key] != value:
+                    print("Expected key {} not found in annotations {}".format(key, svc.metadata.annotation))
                     return False
-            for key in svc.metadata.annotations:
+        return True
-                if svc.metadata.annotations[key] != annotations[key]:
+
+    def check_statefulset_annotations(self, sset_labels, annotations, namespace='default'):
+        ssets = self.api.apps_v1.list_namespaced_stateful_set(namespace, label_selector=sset_labels, limit=1).items
+        for sset in ssets:
+            for key, value in annotations.items():
+                if key not in sset.metadata.annotations or sset.metadata.annotations[key] != value:
+                    print("Expected key {} not found in annotations {}".format(key, sset.metadata.annotation))
                     return False
         return True
 

manifests/configmap.yaml

@@ -30,6 +30,7 @@ data:
   # default_memory_limit: 500Mi
   # default_memory_request: 100Mi
   docker_image: registry.opensource.zalan.do/acid/spilo-cdp-12:1.6-p115
+  # downscaler_annotations: "deployment-time,downscaler/*"
   # enable_admin_role_for_users: "true"
   # enable_crd_validation: "true"
   # enable_database_access: "true"

manifests/operatorconfiguration.crd.yaml

@@ -93,6 +93,10 @@ spec:
                   type: object
                   additionalProperties:
                     type: string
+                downscaler_annotations:
+                  type: array
+                  items:
+                    type: string
                 enable_init_containers:
                   type: boolean
                 enable_pod_antiaffinity:

manifests/postgresql-operator-default-configuration.yaml

@@ -31,6 +31,9 @@ configuration:
     # custom_pod_annotations:
     #   keya: valuea
     #   keyb: valueb
+    # downscaler_annotations:
+    # - deployment-time
+    # - downscaler/*
     enable_init_containers: true
     enable_pod_antiaffinity: false
     enable_pod_disruption_budget: true

pkg/apis/acid.zalan.do/v1/crds.go

@@ -21,48 +21,48 @@ const (
 
 // PostgresCRDResourceColumns definition of AdditionalPrinterColumns for postgresql CRD
 var PostgresCRDResourceColumns = []apiextv1beta1.CustomResourceColumnDefinition{
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "Team",
 		Type:        "string",
 		Description: "Team responsible for Postgres cluster",
 		JSONPath:    ".spec.teamId",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "Version",
 		Type:        "string",
 		Description: "PostgreSQL version",
 		JSONPath:    ".spec.postgresql.version",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "Pods",
 		Type:        "integer",
 		Description: "Number of Pods per Postgres cluster",
 		JSONPath:    ".spec.numberOfInstances",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "Volume",
 		Type:        "string",
 		Description: "Size of the bound volume",
 		JSONPath:    ".spec.volume.size",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "CPU-Request",
 		Type:        "string",
 		Description: "Requested CPU for Postgres containers",
 		JSONPath:    ".spec.resources.requests.cpu",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "Memory-Request",
 		Type:        "string",
 		Description: "Requested memory for Postgres containers",
 		JSONPath:    ".spec.resources.requests.memory",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:     "Age",
 		Type:     "date",
 		JSONPath: ".metadata.creationTimestamp",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "Status",
 		Type:        "string",
 		Description: "Current sync status of postgresql resource",
@@ -72,31 +72,31 @@ var PostgresCRDResourceColumns = []apiextv1beta1.CustomResourceColumnDefinition{
 
 // OperatorConfigCRDResourceColumns definition of AdditionalPrinterColumns for OperatorConfiguration CRD
 var OperatorConfigCRDResourceColumns = []apiextv1beta1.CustomResourceColumnDefinition{
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "Image",
 		Type:        "string",
 		Description: "Spilo image to be used for Pods",
 		JSONPath:    ".configuration.docker_image",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "Cluster-Label",
 		Type:        "string",
 		Description: "Label for K8s resources created by operator",
 		JSONPath:    ".configuration.kubernetes.cluster_name_label",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "Service-Account",
 		Type:        "string",
 		Description: "Name of service account to be used",
 		JSONPath:    ".configuration.kubernetes.pod_service_account_name",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:        "Min-Instances",
 		Type:        "integer",
 		Description: "Minimum number of instances per Postgres cluster",
 		JSONPath:    ".configuration.min_instances",
 	},
-	apiextv1beta1.CustomResourceColumnDefinition{
+	{
 		Name:     "Age",
 		Type:     "date",
 		JSONPath: ".metadata.creationTimestamp",
@@ -888,6 +888,14 @@ var OperatorConfigCRDResourceValidation = apiextv1beta1.CustomResourceValidation
 									},
 								},
 							},
+							"downscaler_annotations": {
+								Type: "array",
+								Items: &apiextv1beta1.JSONSchemaPropsOrArray{
+									Schema: &apiextv1beta1.JSONSchemaProps{
+										Type: "string",
+									},
+								},
+							},
 							"enable_init_containers": {
 								Type: "boolean",
 							},

pkg/apis/acid.zalan.do/v1/operator_configuration_type.go

@@ -62,6 +62,7 @@ type KubernetesMetaConfiguration struct {
 	PodRoleLabel                           string                `json:"pod_role_label,omitempty"`
 	ClusterLabels                          map[string]string     `json:"cluster_labels,omitempty"`
 	InheritedLabels                        []string              `json:"inherited_labels,omitempty"`
+	DownscalerAnnotations                  []string              `json:"downscaler_annotations,omitempty"`
 	ClusterNameLabel                       string                `json:"cluster_name_label,omitempty"`
 	NodeReadinessLabel                     map[string]string     `json:"node_readiness_label,omitempty"`
 	CustomPodAnnotations                   map[string]string     `json:"custom_pod_annotations,omitempty"`

pkg/apis/acid.zalan.do/v1/zz_generated.deepcopy.go

@@ -180,6 +180,11 @@ func (in *KubernetesMetaConfiguration) DeepCopyInto(out *KubernetesMetaConfigura
 		*out = make([]string, len(*in))
 		copy(*out, *in)
 	}
+	if in.DownscalerAnnotations != nil {
+		in, out := &in.DownscalerAnnotations, &out.DownscalerAnnotations
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
 	if in.NodeReadinessLabel != nil {
 		in, out := &in.NodeReadinessLabel, &out.NodeReadinessLabel
 		*out = make(map[string]string, len(*in))

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) || !reflect.DeepEqual(oldSpec.Annotations, newSpec.Annotations) {
-		if !reflect.DeepEqual(oldSs, newSs) {
 			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{
+			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{})

pkg/controller/operator_config.go

@@ -73,6 +73,7 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
 	result.PodRoleLabel = util.Coalesce(fromCRD.Kubernetes.PodRoleLabel, "spilo-role")
 	result.ClusterLabels = fromCRD.Kubernetes.ClusterLabels
 	result.InheritedLabels = fromCRD.Kubernetes.InheritedLabels
+	result.DownscalerAnnotations = fromCRD.Kubernetes.DownscalerAnnotations
 	result.ClusterNameLabel = util.Coalesce(fromCRD.Kubernetes.ClusterNameLabel, "cluster-name")
 	result.NodeReadinessLabel = fromCRD.Kubernetes.NodeReadinessLabel
 	result.PodPriorityClassName = fromCRD.Kubernetes.PodPriorityClassName

pkg/controller/postgresql.go

@@ -487,8 +487,10 @@ func (c *Controller) postgresqlUpdate(prev, cur interface{}) {
 	if pgOld != nil && pgNew != nil {
 		// Avoid the inifinite recursion for status updates
 		if reflect.DeepEqual(pgOld.Spec, pgNew.Spec) {
+			if reflect.DeepEqual(pgNew.Annotations, pgOld.Annotations) {
 				return
 			}
+		}
 		c.queueClusterEvent(pgOld, pgNew, EventUpdate)
 	}
 

pkg/util/config/config.go

@@ -34,6 +34,7 @@ type Resources struct {
 	SpiloPrivileged         bool                `name:"spilo_privileged" default:"false"`
 	ClusterLabels           map[string]string   `name:"cluster_labels" default:"application:spilo"`
 	InheritedLabels         []string            `name:"inherited_labels" default:""`
+	DownscalerAnnotations   []string            `name:"downscaler_annotations"`
 	ClusterNameLabel        string              `name:"cluster_name_label" default:"cluster-name"`
 	PodRoleLabel            string              `name:"pod_role_label" default:"spilo-role"`
 	PodToleration           map[string]string   `name:"toleration" default:""`