Add topologySpreadConstraints configuration to pod spec.

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

@@ -91,6 +91,8 @@ type PostgresSpec struct {
 	// deprecated json tags
 	InitContainersOld       []v1.Container `json:"init_containers,omitempty"`
 	PodPriorityClassNameOld string         `json:"pod_priority_class_name,omitempty"`
+
+	TopologySpreadConstraints []v1.TopologySpreadConstraint `json:"topologySpreadConstraints,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/cluster/cluster.go

@@ -453,6 +453,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

pkg/cluster/k8sres.go

@@ -578,6 +578,36 @@ func generatePodAntiAffinity(podAffinityTerm v1.PodAffinityTerm, preferredDuring
 	return podAntiAffinity
 }
 
+func generateTopologySpreadConstraints(labels labels.Set, topologySpreadConstraintObjs []*v1.TopologySpreadConstraint) []v1.TopologySpreadConstraint {
+	var topologySpreadConstraints []v1.TopologySpreadConstraint
+	var nodeAffinityPolicy *v1.NodeInclusionPolicy
+	var nodeTaintsPolicy *v1.NodeInclusionPolicy
+	for _, topologySpreadConstraintObj := range topologySpreadConstraintObjs {
+		if topologySpreadConstraintObj.NodeAffinityPolicy != nil {
+			nodeAffinityPolicy = (*v1.NodeInclusionPolicy)(topologySpreadConstraintObj.NodeAffinityPolicy)
+		}
+		if topologySpreadConstraintObj.NodeTaintsPolicy != nil {
+			nodeTaintsPolicy = (*v1.NodeInclusionPolicy)(topologySpreadConstraintObj.NodeTaintsPolicy)
+		}
+		topologySpreadConstraint := v1.TopologySpreadConstraint{
+			MaxSkew:           topologySpreadConstraintObj.MaxSkew,
+			TopologyKey:       topologySpreadConstraintObj.TopologyKey,
+			WhenUnsatisfiable: v1.UnsatisfiableConstraintAction(topologySpreadConstraintObj.WhenUnsatisfiable),
+			LabelSelector: &metav1.LabelSelector{
+				MatchLabels: labels,
+			},
+			MinDomains:         topologySpreadConstraintObj.MinDomains,
+			NodeAffinityPolicy: nodeAffinityPolicy,
+			NodeTaintsPolicy:   nodeTaintsPolicy,
+			MatchLabelKeys:     topologySpreadConstraintObj.MatchLabelKeys,
+		}
+		topologySpreadConstraints = append(topologySpreadConstraints, topologySpreadConstraint)
+		nodeAffinityPolicy = nil
+		nodeTaintsPolicy = nil
+	}
+	return topologySpreadConstraints
+}
+
 func tolerations(tolerationsSpec *[]v1.Toleration, podToleration map[string]string) []v1.Toleration {
 	// allow to override tolerations by postgresql manifest
 	if len(*tolerationsSpec) > 0 {
@@ -791,6 +821,7 @@ func (c *Cluster) generatePodTemplate(
 	podAntiAffinity bool,
 	podAntiAffinityTopologyKey string,
 	podAntiAffinityPreferredDuringScheduling bool,
+	topologySpreadConstraints *[]v1.TopologySpreadConstraint,
 	additionalSecretMount string,
 	additionalSecretMountPath string,
 	additionalVolumes []acidv1.AdditionalVolume,
@@ -846,6 +877,10 @@ func (c *Cluster) generatePodTemplate(
 		podSpec.PriorityClassName = priorityClassName
 	}
 
+	if len(topologySpreadConstraints) > 0 {
+		podSpec.TopologySpreadConstraints = generateTopologySpreadConstraints(labels, topologySpreadConstraints)
+	}
+
 	if sharePgSocketWithSidecars != nil && *sharePgSocketWithSidecars {
 		addVarRunVolume(&podSpec)
 	}
@@ -1447,6 +1482,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 		c.OpConfig.EnablePodAntiAffinity,
 		c.OpConfig.PodAntiAffinityTopologyKey,
 		c.OpConfig.PodAntiAffinityPreferredDuringScheduling,
+		spec.TopologySpreadConstraints,
 		c.OpConfig.AdditionalSecretMount,
 		c.OpConfig.AdditionalSecretMountPath,
 		additionalVolumes)
@@ -2282,6 +2318,7 @@ func (c *Cluster) generateLogicalBackupJob() (*batchv1.CronJob, error) {
 		false,
 		"",
 		false,
+		&[]v1.TopologySpreadConstraint{},
 		c.OpConfig.AdditionalSecretMount,
 		c.OpConfig.AdditionalSecretMountPath,
 		[]acidv1.AdditionalVolume{}); err != nil {