Add topologySpreadConstraints configuration to pod spec.

charts/postgres-operator/crds/postgresqls.yaml

@@ -589,6 +589,23 @@ spec:
                         - PreferNoSchedule
                     tolerationSeconds:
                       type: integer
+              topologySpreadConstraints:
+                type: array
+                nullable: true
+                items:
+                  type: object
+                  properties:
+                    maxskew:
+                      type: integer
+                      format: int32
+                      minimum: 1
+                    topologyKey:
+                      type: string
+                    whenUnsatisfiable:
+                      type: string
+                      enum:
+                        - DoNotSchedule
+                        - ScheduleAnyway
               useLoadBalancer:
                 type: boolean
                 description: deprecated

docs/user.md

@@ -714,7 +714,7 @@ but Kubernetes will not spin up the pod if the requested HugePages cannot be all
 For more information on HugePages in Kubernetes, see also
 [https://kubernetes.io/docs/tasks/manage-hugepages/scheduling-hugepages/](https://kubernetes.io/docs/tasks/manage-hugepages/scheduling-hugepages/)
 
-## Use taints, tolerations and node affinity for dedicated PostgreSQL nodes
+## Use taints, tolerations, node affinity and topology spread constraint for dedicated PostgreSQL nodes
 
 To ensure Postgres pods are running on nodes without any other application pods,
 you can use [taints and tolerations](https://kubernetes.io/docs/concepts/configuration/taint-and-toleration/)
@@ -755,6 +755,23 @@ spec:
 If you need to define a `nodeAffinity` for all your Postgres clusters use the
 `node_readiness_label` [configuration](administrator.md#node-readiness-labels).
 
+If you need PostgreSQL Pods to run on separate nodes, you can use the
+[topologySpreadConstraints](https://kubernetes.io/docs/concepts/scheduling-eviction/topology-spread-constraints/) to control how they are distributed across your cluster.
+This ensures they are spread among failure domains such as
+regions, zones, nodes, or other user-defined topology domains.
+
+```yaml
+apiVersion: "acid.zalan.do/v1"
+kind: postgresql
+metadata:
+  name: acid-minimal-cluster
+spec:
+  topologySpreadConstraints:
+    - maxskew: 1
+      topologyKey: topology.kubernetes.io/zone
+      whenUnsatisfiable: DoNotSchedule
+```
+
 ## In-place major version upgrade
 
 Starting with Spilo 14, operator supports in-place major version upgrade to a
@@ -1064,7 +1081,7 @@ spec:
     - all
     volumeSource:
       emptyDir: {}
-  sidecars: 
+  sidecars:
   - name: "container-name"
     image: "company/image:tag"
     volumeMounts:

e2e/tests/test_e2e.py

@@ -560,7 +560,7 @@ def compare_config():
 
             pg_patch_config["spec"]["patroni"]["slots"][slot_to_change]["database"] = "bar"
             del pg_patch_config["spec"]["patroni"]["slots"][slot_to_remove]
-            
+
             k8s.api.custom_objects_api.patch_namespaced_custom_object(
                 "acid.zalan.do", "v1", "default", "postgresqls", "acid-minimal-cluster", pg_delete_slot_patch)
 
@@ -577,7 +577,7 @@ def compare_config():
 
             self.eventuallyEqual(lambda: self.query_database(leader.metadata.name, "postgres", get_slot_query%("database", slot_to_change))[0], "bar",
                 "The replication slot cannot be updated", 10, 5)
-            
+
             # make sure slot from Patroni didn't get deleted
             self.eventuallyEqual(lambda: len(self.query_database(leader.metadata.name, "postgres", get_slot_query%("slot_name", patroni_slot))), 1,
                 "The replication slot from Patroni gets deleted", 10, 5)
@@ -933,7 +933,7 @@ def test_ignored_annotations(self):
                     },
                 }
             }
-            
+
             old_sts_creation_timestamp = sts.metadata.creation_timestamp
             k8s.api.apps_v1.patch_namespaced_stateful_set(sts.metadata.name, sts.metadata.namespace, annotation_patch)
             old_svc_creation_timestamp = svc.metadata.creation_timestamp
@@ -1371,7 +1371,7 @@ def test_persistent_volume_claim_retention_policy(self):
         }
         k8s.update_config(patch_scaled_policy_retain)
         self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, "Operator does not get in sync")
- 
+
         # decrease the number of instances
         k8s.api.custom_objects_api.patch_namespaced_custom_object(
             'acid.zalan.do', 'v1', 'default', 'postgresqls', 'acid-minimal-cluster', pg_patch_scale_down_instances)
@@ -1648,7 +1648,6 @@ def test_node_readiness_label(self):
         # toggle pod anti affinity to move replica away from master node
         self.assert_distributed_pods(master_nodes)
 
-
     @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
     def test_overwrite_pooler_deployment(self):
         pooler_name = 'acid-minimal-cluster-pooler'
@@ -1801,7 +1800,7 @@ def test_password_rotation(self):
             },
         }
         k8s.api.core_v1.patch_namespaced_secret(
-            name="foo-user.acid-minimal-cluster.credentials.postgresql.acid.zalan.do", 
+            name="foo-user.acid-minimal-cluster.credentials.postgresql.acid.zalan.do",
             namespace="default",
             body=secret_fake_rotation)
 
@@ -1818,7 +1817,7 @@ def test_password_rotation(self):
                 "enable_password_rotation": "true",
                 "inherited_annotations": "environment",
                 "password_rotation_interval": "30",
-                "password_rotation_user_retention": "30",  # should be set to 60 
+                "password_rotation_user_retention": "30",  # should be set to 60
             },
         }
         k8s.update_config(enable_password_rotation)
@@ -1887,7 +1886,7 @@ def test_password_rotation(self):
         self.assertTrue("environment" in db_user_secret.metadata.annotations, "Added annotation was not propagated to secret")
 
         # disable password rotation for all other users (foo_user)
-        # and pick smaller intervals to see if the third fake rotation user is dropped 
+        # and pick smaller intervals to see if the third fake rotation user is dropped
         enable_password_rotation = {
             "data": {
                 "enable_password_rotation": "false",
@@ -2387,6 +2386,90 @@ def test_taint_based_eviction(self):
         # toggle pod anti affinity to move replica away from master node
         self.assert_distributed_pods(master_nodes)
 
+    @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
+    def test_topology_spread_constraints(self):
+        '''
+            Enable topologySpreadConstraints for pods
+        '''
+        k8s = self.k8s
+        cluster_labels = "application=spilo,cluster-name=acid-minimal-cluster"
+
+        # Verify we are in good state from potential previous tests
+        self.eventuallyEqual(lambda: k8s.count_running_pods(), 2, "No 2 pods running")
+
+        # patch the pvc retention policy to enable delete when scale down
+        patch_scaled_policy_delete = {
+            "data": {
+                "persistent_volume_claim_retention_policy": "when_deleted:retain,when_scaled:delete"
+            }
+        }
+        k8s.update_config(patch_scaled_policy_delete)
+        self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, "Operator does not get in sync")
+
+        master_nodes, replica_nodes = k8s.get_cluster_nodes()
+        self.assertNotEqual(master_nodes, [])
+        self.assertNotEqual(replica_nodes, [])
+
+        # Patch label to nodes for topologySpreadConstraints
+        patch_node_label = {
+            "metadata": {
+                "labels": {
+                    "topology.kubernetes.io/zone": "zalando"
+                }
+            }
+        }
+        k8s.api.core_v1.patch_node(master_nodes[0], patch_node_label)
+        k8s.api.core_v1.patch_node(replica_nodes[0], patch_node_label)
+
+        # Patch topologySpreadConstraint and scale-out postgresql pods to postgresqls manifest.
+        patch_topologySpreadConstraint_config = {
+            "spec": {
+                "numberOfInstances": 6,
+                "topologySpreadConstraint": [
+                    {
+                        "maxskew": 1,
+                        "topologyKey": "topology.kubernetes.io/zone",
+                        "whenUnsatisfiable": "DoNotSchedule"
+                    }
+                ]
+            }
+        }
+        k8s.api.custom_objects_api.patch_namespaced_custom_object(
+            "acid.zalan.do", "v1", "default",
+            "postgresqls", "acid-minimal-cluster",
+            patch_topologySpreadConstraint_config)
+        self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, "Operator does not get in sync")
+        self.eventuallyEqual(lambda: k8s.count_pods_with_label(cluster_labels), 6, "Postgresql StatefulSet are scale to 6")
+        self.eventuallyEqual(lambda: k8s.count_running_pods(), 6, "All pods are running")
+
+        worker_node_1 = 0
+        worker_node_2 = 0
+        pods = k8s.api.core_v1.list_namespaced_pod('default', label_selector=cluster_labels)
+        for pod in pods.items:
+            if pod.spec.node_name == 'postgres-operator-e2e-tests-worker':
+                worker_node_1 += 1
+            elif pod.spec.node_name == 'postgres-operator-e2e-tests-worker2':
+                worker_node_2 += 1
+
+        self.assertEqual(worker_node_1, worker_node_2)
+        self.assertEqual(worker_node_1, 3)
+        self.assertEqual(worker_node_2, 3)
+
+        # Reset configurations
+        patch_topologySpreadConstraint_config = {
+            "spec": {
+                "numberOfInstances": 2,
+                "topologySpreadConstraint": []
+            }
+        }
+        k8s.api.custom_objects_api.patch_namespaced_custom_object(
+            "acid.zalan.do", "v1", "default",
+            "postgresqls", "acid-minimal-cluster",
+            patch_topologySpreadConstraint_config)
+        self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, "Operator does not get in sync")
+        self.eventuallyEqual(lambda: k8s.count_pods_with_label(cluster_labels), 2, "Postgresql StatefulSet are scale to 2")
+        self.eventuallyEqual(lambda: k8s.count_running_pods(), 2, "All pods are running")
+
     @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
     def test_zz_cluster_deletion(self):
         '''
@@ -2462,7 +2545,7 @@ def test_zz_cluster_deletion(self):
             self.eventuallyEqual(lambda: k8s.count_deployments_with_label(cluster_label), 0, "Deployments not deleted")
             self.eventuallyEqual(lambda: k8s.count_pdbs_with_label(cluster_label), 0, "Pod disruption budget not deleted")
             self.eventuallyEqual(lambda: k8s.count_secrets_with_label(cluster_label), 8, "Secrets were deleted although disabled in config")
-            self.eventuallyEqual(lambda: k8s.count_pvcs_with_label(cluster_label), 3, "PVCs were deleted although disabled in config")
+            self.eventuallyEqual(lambda: k8s.count_pvcs_with_label(cluster_label), 2, "PVCs were deleted although disabled in config")
 
         except timeout_decorator.TimeoutError:
             print('Operator log: {}'.format(k8s.get_operator_log()))
@@ -2504,7 +2587,7 @@ def assert_distributed_pods(self, target_nodes, cluster_labels='cluster-name=aci
 
         # if nodes are different we can quit here
         if master_nodes[0] not in replica_nodes:
-            return True             
+            return True
 
         # enable pod anti affintiy in config map which should trigger movement of replica
         patch_enable_antiaffinity = {
@@ -2528,7 +2611,7 @@ def assert_distributed_pods(self, target_nodes, cluster_labels='cluster-name=aci
             }
             k8s.update_config(patch_disable_antiaffinity, "disable antiaffinity")
             self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, "Operator does not get in sync")
-            
+
             k8s.wait_for_pod_start('spilo-role=replica,' + cluster_labels)
             k8s.wait_for_running_pods(cluster_labels, 2)
 
@@ -2539,7 +2622,7 @@ def assert_distributed_pods(self, target_nodes, cluster_labels='cluster-name=aci
             # if nodes are different we can quit here
             for target_node in target_nodes:
                 if (target_node not in master_nodes or target_node not in replica_nodes) and master_nodes[0] in replica_nodes:
-                    print('Pods run on the same node') 
+                    print('Pods run on the same node')
                     return False
 
         except timeout_decorator.TimeoutError:

manifests/complete-postgres-manifest.yaml

@@ -232,6 +232,12 @@ spec:
 #              values:
 #                - enabled
 
+# Add topology spread constraint to distribute PostgreSQL pods across all nodes labeled with "topology.kubernetes.io/zone".
+#   topologySpreadConstraint:
+#     - maxSkew: 1
+#       topologyKey: topology.kubernetes.io/zone
+#       whenUnsatisfiable: DoNotSchedule
+
 # Enables change data capture streams for defined database tables
 #  streams:
 #  - applicationId: test-app

manifests/postgresql.crd.yaml

@@ -4056,6 +4056,22 @@ spec:
                       type: string
                   type: object
                 type: array
+              topologySpreadConstraints:
+                type: array
+                nullable: true
+                items:
+                  type: object
+                  properties:
+                    maxSkew:
+                      type: integer
+                      format: int32
+                    topologyKey:
+                      type: string
+                    whenUnsatisfiable:
+                      type: string
+                      enum:
+                        - DoNotSchedule
+                        - ScheduleAnyway
               useLoadBalancer:
                 description: |-
                   deprecated load balancer settings maintained for backward compatibility

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

@@ -56,6 +56,7 @@ var OperatorConfigCRDResourceColumns = []apiextv1.CustomResourceColumnDefinition
 }
 
 var min1 = 1.0
+var minLength1 int64 = 1
 var minDisable = -1.0
 
 // OperatorConfigCRDResourceValidation to check applied manifest parameters

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

@@ -92,17 +92,18 @@ type PostgresSpec struct {
 	Clone              *CloneDescription   `json:"clone,omitempty"`
 	// Note: usernames specified here as database owners must be declared
 	// in the users key of the spec key.
-	Databases              map[string]string           `json:"databases,omitempty"`
-	PreparedDatabases      map[string]PreparedDatabase `json:"preparedDatabases,omitempty"`
-	SchedulerName          *string                     `json:"schedulerName,omitempty"`
-	NodeAffinity           *v1.NodeAffinity            `json:"nodeAffinity,omitempty"`
-	Tolerations            []v1.Toleration             `json:"tolerations,omitempty"`
-	Sidecars               []Sidecar                   `json:"sidecars,omitempty"`
-	InitContainers         []v1.Container              `json:"initContainers,omitempty"`
-	PodPriorityClassName   string                      `json:"podPriorityClassName,omitempty"`
-	ShmVolume              *bool                       `json:"enableShmVolume,omitempty"`
-	EnableLogicalBackup    bool                        `json:"enableLogicalBackup,omitempty"`
-	LogicalBackupRetention string                      `json:"logicalBackupRetention,omitempty"`
+	Databases                 map[string]string             `json:"databases,omitempty"`
+	PreparedDatabases         map[string]PreparedDatabase   `json:"preparedDatabases,omitempty"`
+	SchedulerName             *string                       `json:"schedulerName,omitempty"`
+	NodeAffinity              *v1.NodeAffinity              `json:"nodeAffinity,omitempty"`
+	TopologySpreadConstraints []v1.TopologySpreadConstraint `json:"topologySpreadConstraints,omitempty"`
+	Tolerations               []v1.Toleration               `json:"tolerations,omitempty"`
+	Sidecars                  []Sidecar                     `json:"sidecars,omitempty"`
+	InitContainers            []v1.Container                `json:"initContainers,omitempty"`
+	PodPriorityClassName      string                        `json:"podPriorityClassName,omitempty"`
+	ShmVolume                 *bool                         `json:"enableShmVolume,omitempty"`
+	EnableLogicalBackup       bool                          `json:"enableLogicalBackup,omitempty"`
+	LogicalBackupRetention    string                        `json:"logicalBackupRetention,omitempty"`
 	// +kubebuilder:validation:Pattern=`^(\d+|\*)(/\d+)?(\s+(\d+|\*)(/\d+)?){4}$`
 	LogicalBackupSchedule string              `json:"logicalBackupSchedule,omitempty"`
 	StandbyCluster        *StandbyDescription `json:"standby,omitempty"`

pkg/cluster/cluster.go

@@ -504,6 +504,11 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compa
 		needsRollUpdate = true
 		reasons = append(reasons, "new statefulset's pod affinity does not match the current one")
 	}
+	if !reflect.DeepEqual(c.Statefulset.Spec.Template.Spec.TopologySpreadConstraints, statefulSet.Spec.Template.Spec.TopologySpreadConstraints) {
+		needsReplace = true
+		needsRollUpdate = true
+		reasons = append(reasons, "new statefulset's pod topologySpreadConstraints does not match the current one")
+	}
 	if len(c.Statefulset.Spec.Template.Spec.Tolerations) != len(statefulSet.Spec.Template.Spec.Tolerations) {
 		needsReplace = true
 		needsRollUpdate = true