Merge pull request #2290 from iamniting/conditions

Consider all storage clusters status for operator upgradability

controllers/storagecluster/reconcile.go

@@ -377,8 +377,14 @@ func (r *StorageClusterReconciler) reconcilePhases(
 			}
 		}
 		r.Log.Info("StorageCluster is terminated, skipping reconciliation.", "StorageCluster", klog.KRef(instance.Namespace, instance.Name))
-		returnErr := r.SetOperatorConditions("Skipping StorageCluster reconciliation", "Terminated", metav1.ConditionTrue, nil)
-		return reconcile.Result{}, returnErr
+
+		// mark operator upgradeable if and only if all other storageclusters are ready or current cluster is the last cluster
+		if r.clusters.AreOtherStorageClustersReady(instance) {
+			returnErr := r.SetOperatorConditions("Skipping StorageCluster reconciliation", "Terminated", metav1.ConditionTrue, nil)
+			return reconcile.Result{}, returnErr
+		}
+
+		return reconcile.Result{}, nil
 	}
 
 	// in-memory conditions should start off empty. It will only ever hold
@@ -487,10 +493,16 @@ func (r *StorageClusterReconciler) reconcilePhases(
 		// to set upgradeable to true.
 		if instance.Status.Phase != statusutil.PhaseClusterExpanding {
 			instance.Status.Phase = statusutil.PhaseReady
-			returnErr := r.SetOperatorConditions(message, reason, metav1.ConditionTrue, nil)
-			if returnErr != nil {
-				return reconcile.Result{}, returnErr
+
+			// mark operator upgradeable if and only if all storageclusters are ready
+			if r.clusters.AreOtherStorageClustersReady(instance) {
+				returnErr := r.SetOperatorConditions(message, reason, metav1.ConditionTrue, nil)
+				if returnErr != nil {
+					return reconcile.Result{}, returnErr
+				}
 			}
+			return reconcile.Result{}, nil
+
 		}
 	} else {
 		// If any component operator reports negatively we want to write that to

controllers/util/clusters.go

@@ -113,3 +113,18 @@ func GetClusters(ctx context.Context, cli client.Client) (*Clusters, error) {
 		namespacedNames:         namespacedNames,
 	}, nil
 }
+
+// AreOtherStorageClustersReady checks if all other storage clusters (internal and external) are ready.
+func (c *Clusters) AreOtherStorageClustersReady(instance *ocsv1.StorageCluster) bool {
+
+	for _, sc := range append(c.internalStorageClusters, c.externalStorageClusters...) {
+		// ignore the current recociling storage cluster as its status will set in the current reconcile
+		if sc.Name != instance.Name && sc.Namespace != instance.Namespace {
+			if sc.Status.Phase != PhaseReady {
+				return false
+			}
+		}
+	}
+
+	return true
+}