Correct the reconciliation status after the initial run after the cluster wide bounce (#2341)

* Correct the reconciliation status after the initial run after the cluster wide bounce

Author: johscheuer

api/v1beta2/foundationdbcluster_types.go

@@ -1805,6 +1805,7 @@ func (cluster *FoundationDBCluster) CheckReconciliation(log logr.Logger) (bool,
 			"reconciledProcessGroups",
 			cluster.Status.ReconciledProcessGroups,
 		)
+		reconciled = false
 	}
 
 	if !cluster.Status.Health.Available {
@@ -1903,6 +1904,19 @@ func (cluster *FoundationDBCluster) CheckReconciliation(log logr.Logger) (bool,
 		}
 	}
 
+	// If the cluster is currently in the process to be upgraded, don't set the reconciled version.
+	if cluster.IsBeingUpgraded() {
+		logger.Info(
+			"Pending cluster upgrade",
+			"runningVersion",
+			cluster.GetRunningVersion(),
+			"desiredVersion",
+			cluster.Spec.Version,
+		)
+
+		reconciled = false
+	}
+
 	if reconciled && cluster.Status.Generations.Reconciled != cluster.Generation {
 		logger.Info(
 			"Update reconciled generation",

controllers/update_sidecar_versions.go

@@ -22,12 +22,11 @@ package controllers
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/go-logr/logr"
 
-	"github.com/FoundationDB/fdb-kubernetes-operator/v2/pkg/podmanager"
-
 	"github.com/FoundationDB/fdb-kubernetes-operator/v2/internal"
 
 	corev1 "k8s.io/api/core/v1"
@@ -53,6 +52,7 @@ func (updateSidecarVersions) reconcile(
 	}
 
 	var upgraded int
+	var errs []error
 	for _, processGroup := range cluster.Status.ProcessGroups {
 		if processGroup.GetConditionTime(fdbv1beta2.ResourcesTerminating) != nil {
 			logger.V(1).Info("Ignore process group that is stuck terminating",
@@ -74,37 +74,13 @@ func (updateSidecarVersions) reconcile(
 			continue
 		}
 
-		image, err := internal.GetSidecarImage(cluster, processGroup.ProcessClass)
+		err = updateSidecarImage(ctx, r, logger, cluster, processGroup, pod)
 		if err != nil {
-			return &requeue{curError: err}
+			errs = append(errs, err)
+			continue
 		}
 
-		for containerIndex, container := range pod.Spec.Containers {
-			if container.Name == fdbv1beta2.SidecarContainerName && container.Image != image {
-				logger.Info(
-					"Upgrading sidecar",
-					"processGroupID",
-					podmanager.GetProcessGroupID(cluster, pod),
-					"oldImage",
-					container.Image,
-					"newImage",
-					image,
-				)
-				err = r.PodLifecycleManager.UpdateImageVersion(
-					logr.NewContext(ctx, logger),
-					r,
-					cluster,
-					pod,
-					containerIndex,
-					image,
-				)
-				if err != nil {
-					return &requeue{curError: err}
-				}
-				upgraded++
-				break
-			}
-		}
+		upgraded++
 	}
 
 	if upgraded > 0 {
@@ -120,5 +96,54 @@ func (updateSidecarVersions) reconcile(
 		)
 	}
 
+	if len(errs) > 0 {
+		return &requeue{
+			curError: errors.Join(errs...),
+		}
+	}
+
+	return nil
+}
+
+func updateSidecarImage(
+	ctx context.Context,
+	r *FoundationDBClusterReconciler,
+	logger logr.Logger,
+	cluster *fdbv1beta2.FoundationDBCluster,
+	processGroup *fdbv1beta2.ProcessGroupStatus,
+	pod *corev1.Pod,
+) error {
+	image, err := internal.GetSidecarImage(cluster, processGroup.ProcessClass)
+	if err != nil {
+		return err
+	}
+
+	for containerIndex, container := range pod.Spec.Containers {
+		if container.Name == fdbv1beta2.SidecarContainerName && container.Image != image {
+			logger.Info(
+				"Upgrading sidecar",
+				"processGroupID",
+				processGroup.ProcessGroupID,
+				"oldImage",
+				container.Image,
+				"newImage",
+				image,
+			)
+			err = r.PodLifecycleManager.UpdateImageVersion(
+				logr.NewContext(ctx, logger),
+				r,
+				cluster,
+				pod,
+				containerIndex,
+				image,
+			)
+			if err != nil {
+				return err
+			}
+
+			break
+		}
+	}
+
 	return nil
 }

controllers/update_status.go

@@ -98,7 +98,7 @@ func (c updateStatus) reconcile(
 	}
 
 	// Update the running version based on the reported version of the FDB processes
-	version, err := getRunningVersion(logger, versionMap, cluster.Status.RunningVersion)
+	version, err := getRunningVersion(logger, versionMap, cluster.GetRunningVersion())
 	if err != nil {
 		return &requeue{
 			curError: fmt.Errorf(
@@ -108,6 +108,14 @@ func (c updateStatus) reconcile(
 		}
 	}
 	clusterStatus.RunningVersion = version
+	// We update the running version of the cluster status here to make sure that the Pod Spec hash and other things
+	// are properly calculated based on the current observed running version. If we don't update the running version
+	// here, we could see a race condition where the updateStatus sub-reconciler runs for the first time after an upgrade
+	// bounce. In this case the running version of the new cluster status (clusterStatus) would be updated but the
+	// status on the cluster struct would be the same. The result of this would be that the operator things the cluster
+	// is reconciled and in the next updateStatus run, it would observe all the required changes after the cluster
+	// wide bounce.
+	cluster.Status.RunningVersion = version
 
 	clusterStatus.HasListenIPsForAllPods = cluster.NeedsExplicitListenAddress()
 	// Update the configuration if the database is available, otherwise the machine-readable status will contain no information

controllers/update_status_test.go

@@ -1266,7 +1266,6 @@ var _ = Describe("update_status", func() {
 					}
 				})
 			})
-
 		})
 	})
 
@@ -1277,8 +1276,7 @@ var _ = Describe("update_status", func() {
 
 		BeforeEach(func() {
 			cluster = internal.CreateDefaultCluster()
-			err = k8sClient.Create(context.TODO(), cluster)
-			Expect(err).NotTo(HaveOccurred())
+			Expect(k8sClient.Create(context.TODO(), cluster)).To(Succeed())
 
 			result, err := reconcileCluster(cluster)
 			Expect(err).NotTo(HaveOccurred())
@@ -1290,8 +1288,9 @@ var _ = Describe("update_status", func() {
 		})
 
 		JustBeforeEach(func() {
-			err = internal.NormalizeClusterSpec(cluster, internal.DeprecationOptions{})
-			Expect(err).NotTo(HaveOccurred())
+			Expect(
+				internal.NormalizeClusterSpec(cluster, internal.DeprecationOptions{}),
+			).To(Succeed())
 			requeue = updateStatus{}.reconcile(
 				context.TODO(),
 				clusterReconciler,
@@ -1351,6 +1350,61 @@ var _ = Describe("update_status", func() {
 				}
 			})
 		})
+
+		When("the cluster was bounced for an upgrade", func() {
+			BeforeEach(func() {
+				version, err := fdbv1beta2.ParseFdbVersion(cluster.Spec.Version)
+				Expect(err).NotTo(HaveOccurred())
+				cluster.Spec.Version = version.NextMajorVersion().String()
+				Expect(k8sClient.Update(context.Background(), cluster)).To(Succeed())
+
+				adminClient, err := mock.NewMockAdminClientUncast(cluster, k8sClient)
+				Expect(err).NotTo(HaveOccurred())
+
+				Expect(
+					internal.NormalizeClusterSpec(cluster, internal.DeprecationOptions{}),
+				).To(Succeed())
+				// Prepare the sidecar versions and make sure that we bounce the cluster.
+				var addresses []fdbv1beta2.ProcessAddress
+				versions := make(map[fdbv1beta2.ProcessGroupID]string)
+				for _, processGroup := range cluster.Status.ProcessGroups {
+					for _, addr := range processGroup.Addresses {
+						parsedAddr, err := fdbv1beta2.ParseProcessAddress(addr)
+						Expect(err).NotTo(HaveOccurred())
+						addresses = append(addresses, parsedAddr)
+					}
+					versions[processGroup.ProcessGroupID] = cluster.Spec.Version
+
+					pod, err := internal.GetPod(cluster, processGroup)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(
+						updateSidecarImage(
+							context.Background(),
+							clusterReconciler,
+							logger,
+							cluster,
+							processGroup,
+							pod,
+						),
+					).To(Succeed())
+				}
+
+				Expect(adminClient.KillProcesses(addresses)).NotTo(HaveOccurred())
+				adminClient.VersionProcessGroups = versions
+
+				_, err = reloadCluster(cluster)
+				Expect(err).NotTo(HaveOccurred())
+			})
+
+			It("should update the incorrect sidecar image condition", func() {
+				Expect(requeue).NotTo(BeNil())
+				Expect(cluster.Generation).NotTo(Equal(cluster.Status.Generations.Reconciled))
+
+				for _, processGroup := range cluster.Status.ProcessGroups {
+					Expect(processGroup.ProcessGroupConditions).NotTo(HaveLen(0))
+				}
+			})
+		})
 	})
 
 	DescribeTable(

e2e/test_operator_upgrades_variations/operator_upgrades_variations_test.go

@@ -192,7 +192,7 @@ func performUpgrade(config testConfig, preUpgradeFunction func(cluster *fixtures
 	// replacements during an upgrade.
 	expectedProcessCounts := (processCounts.Total()-processCounts.Storage)*2 + 5
 	Expect(len(transactionSystemProcessGroups)).To(BeNumerically("<=", expectedProcessCounts))
-	// Ensure we have not data loss.
+	// Ensure we have no data loss.
 	fdbCluster.EnsureTeamTrackersAreHealthy()
 	fdbCluster.EnsureTeamTrackersHaveMinReplicas()
 }