Merge branch 'master' into update-ebs-tags-inherit-labels

pkg/cluster/cluster.go

@@ -1140,6 +1140,12 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 		c.logger.Infof("Storage resize is disabled (storage_resize_mode is off). Skipping volume size sync.")
 	}
 
+	// Pod service account (IRSA annotation sync)
+	if err := c.syncPodServiceAccount(); err != nil {
+		c.logger.Errorf("could not sync pod service account: %v", err)
+		updateFailed = true
+	}
+
 	// Statefulset
 	func() {
 		if err := c.syncStatefulSet(); err != nil {

pkg/cluster/k8sres.go

@@ -2278,6 +2278,12 @@ func (c *Cluster) generatePrimaryPodDisruptionBudget() *policyv1.PodDisruptionBu
 		labels[c.OpConfig.PodRoleLabel] = string(Master)
 	}
 
+	// When master selector is disabled and synchronous_mode_strict is on, require
+	// master + synchronous_node_count (default 1) healthy pods for write quorum.
+	if pdbMasterLabelSelector != nil && !*pdbMasterLabelSelector && minAvailable.IntVal > 0 && c.Spec.SynchronousModeStrict {
+		minAvailable = intstr.FromInt32(int32(c.Spec.SynchronousNodeCount + 1))
+	}
+
 	return &policyv1.PodDisruptionBudget{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:            c.PrimaryPodDisruptionBudgetName(),

pkg/cluster/k8sres_test.go

@@ -2691,13 +2691,13 @@ func TestGeneratePodDisruptionBudget(t *testing.T) {
 				k8sutil.KubernetesClient{},
 				acidv1.Postgresql{
 					ObjectMeta: metav1.ObjectMeta{Name: "myapp-database", Namespace: "myapp"},
-					Spec:       acidv1.PostgresSpec{TeamID: "myapp", NumberOfInstances: 3}},
+					Spec:       acidv1.PostgresSpec{TeamID: "myapp", NumberOfInstances: 3, Patroni: acidv1.Patroni{SynchronousModeStrict: true, SynchronousNodeCount: 1}}},
 				logger,
 				eventRecorder),
 			check: []func(cluster *Cluster, podDisruptionBudget *policyv1.PodDisruptionBudget) error{
 				testPodDisruptionBudgetOwnerReference,
 				hasName("postgres-myapp-database-pdb"),
-				hasMinAvailable(1),
+				hasMinAvailable(2),
 				testLabelsAndSelectors(true),
 			},
 		},

pkg/cluster/sync.go

@@ -107,6 +107,10 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 		}
 	}
 
+	if err = c.syncPodServiceAccount(); err != nil {
+		c.logger.Errorf("could not sync pod service account: %v", err)
+	}
+
 	if err = c.syncStatefulSet(); err != nil {
 		if !k8sutil.ResourceAlreadyExists(err) {
 			err = fmt.Errorf("could not sync statefulsets: %v", err)
@@ -630,6 +634,10 @@ func (c *Cluster) syncStatefulSet() error {
 		if !cmp.rollingUpdate {
 			updatedPodAnnotations := map[string]*string{}
 			for _, anno := range cmp.deletedPodAnnotations {
+				// during IRSA migration let kube2iam annotation drain naturally via pod rotation
+				if c.OpConfig.IRSARoleARN != "" && anno == constants.KubeIAmAnnotation {
+					continue
+				}
 				updatedPodAnnotations[anno] = nil
 			}
 			for anno, val := range desiredSts.Spec.Template.Annotations {
@@ -1803,6 +1811,7 @@ func (c *Cluster) syncLogicalBackupJob() error {
 	// no existing logical backup job, create new one
 	c.logger.Info("could not find the cluster's logical backup job")
 
+
 	if err = c.createLogicalBackupJob(); err == nil {
 		c.logger.Infof("created missing logical backup job %s", jobName)
 	} else {
@@ -1817,3 +1826,62 @@ func (c *Cluster) syncLogicalBackupJob() error {
 
 	return nil
 }
+
+func (c *Cluster) syncPodServiceAccount() error {
+	sa, err := c.KubeClient.ServiceAccounts(c.Namespace).Get(context.TODO(), c.OpConfig.PodServiceAccountName, metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("could not get pod service account %q: %v", c.OpConfig.PodServiceAccountName, err)
+	}
+
+	changed := false
+
+	if c.OpConfig.IRSARoleARN != "" {
+		if val, ok := sa.Annotations[constants.IRSAAnnotation]; !ok || val != c.OpConfig.IRSARoleARN {
+			if sa.Annotations == nil {
+				sa.Annotations = make(map[string]string)
+			}
+			sa.Annotations[constants.IRSAAnnotation] = c.OpConfig.IRSARoleARN
+			changed = true
+		}
+	} else {
+		if _, ok := sa.Annotations[constants.IRSAAnnotation]; ok {
+			delete(sa.Annotations, constants.IRSAAnnotation)
+			changed = true
+		}
+	}
+
+	if changed {
+		if _, err = c.KubeClient.ServiceAccounts(c.Namespace).Update(context.TODO(), sa, metav1.UpdateOptions{}); err != nil {
+			return fmt.Errorf("could not update pod service account %q: %v", sa.Name, err)
+		}
+		c.logger.Infof("synced annotations on pod service account %q", sa.Name)
+	}
+
+	if c.OpConfig.IRSARoleARN != "" {
+		c.logIRSAMigrationProgress()
+	}
+
+	return nil
+}
+
+func (c *Cluster) logIRSAMigrationProgress() {
+	pods, err := c.listPods()
+	if err != nil {
+		c.logger.Warnf("IRSA migration: could not list pods: %v", err)
+		return
+	}
+
+	total := len(pods)
+	remaining := 0
+	for _, pod := range pods {
+		if _, ok := pod.Annotations[constants.KubeIAmAnnotation]; ok {
+			remaining++
+		}
+	}
+
+	if remaining > 0 {
+		c.logger.Infof("IRSA migration in progress: %d/%d pods still carry kube2iam annotation, will be removed on next rotation", remaining, total)
+	} else {
+		c.logger.Infof("IRSA migration complete: all %d pods have rotated, kube2iam annotation fully drained", total)
+	}
+}

pkg/cluster/util_test.go

@@ -53,6 +53,7 @@ func newFakeK8sAnnotationsClient() (k8sutil.KubernetesClient, *k8sFake.Clientset
 		EndpointsGetter:              clientSet.CoreV1(),
 		ConfigMapsGetter:             clientSet.CoreV1(),
 		PodsGetter:                   clientSet.CoreV1(),
+		ServiceAccountsGetter:        clientSet.CoreV1(),
 		DeploymentsGetter:            clientSet.AppsV1(),
 		CronJobsGetter:               clientSet.BatchV1(),
 	}, clientSet