[RHSTOR-5073] Stop provider upgrade if client becomes unsupported

- Provider supports client operator at X.Y and X.(Y-1) versions
- If any client is already at X.(Y-1) version and provider gets upgraded
then the lagging client goes out of support
- This PR prohibits such provider upgrades

Signed-off-by: Leela Venkaiah G <[email protected]>

controllers/storagecluster/provider_server.go

@@ -8,6 +8,7 @@ import (
 	"sort"
 	"time"
 
+	"github.com/blang/semver/v4"
 	"go.uber.org/multierr"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -19,8 +20,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	ocsv1 "github.com/red-hat-storage/ocs-operator/v4/api/v1"
+	ocsv1alpha1 "github.com/red-hat-storage/ocs-operator/v4/api/v1alpha1"
 	"github.com/red-hat-storage/ocs-operator/v4/controllers/util"
 	"github.com/red-hat-storage/ocs-operator/v4/services/provider/server"
+	ocsversion "github.com/red-hat-storage/ocs-operator/v4/version"
 )
 
 const (
@@ -31,6 +34,8 @@ const (
 	ocsProviderServiceNodePort = int32(31659)
 
 	ocsProviderCertSecretName = ocsProviderServerName + "-cert"
+
+	ocsProviderConsumerSupport = "ConsumerClientOperatorVersion"
 )
 
 type ocsProviderServer struct{}
@@ -63,6 +68,10 @@ func (o *ocsProviderServer) ensureCreated(r *StorageClusterReconciler, instance
 		return res, nil
 	}
 
+	if err := o.setUpgradeableCondition(r, instance); err != nil {
+		return reconcile.Result{}, err
+	}
+
 	return reconcile.Result{}, nil
 }
 
@@ -302,6 +311,37 @@ func (o *ocsProviderServer) ensureDeploymentReplica(actual, desired *appsv1.Depl
 	return nil
 }
 
+func (o *ocsProviderServer) setUpgradeableCondition(r *StorageClusterReconciler, instance *ocsv1.StorageCluster) error {
+
+	scList := &ocsv1alpha1.StorageConsumerList{}
+	err := r.Client.List(r.ctx, scList, &client.ListOptions{Namespace: instance.Namespace})
+	if err != nil {
+		r.Log.Error(err, "Failed to list StorageConsumers")
+		return err
+	}
+
+	providerVersion, err := semver.Make(ocsversion.Version)
+	if err != nil {
+		r.Log.Error(err, "Failed to parse OCS Version")
+		return err
+	}
+
+	for idx := range scList.Items {
+		clientVersion, err := semver.Make(scList.Items[idx].Status.Client.OperatorVersion)
+		if err != nil {
+			r.Log.Error(err, "Failed to parse Client operator version")
+			return err
+		}
+		// we only onboard client that are of same provider version and support one less minor client version
+		if providerVersion.Minor-1 == clientVersion.Minor {
+			message := fmt.Sprintf("StorageConsumer %q will become unsupported if upgraded", scList.Items[idx].GetName())
+			return r.SetOperatorConditions(message, ocsProviderConsumerSupport, metav1.ConditionFalse, nil)
+		}
+	}
+
+	return r.SetOperatorConditions("All StorageConsumers are within support versions", ocsProviderConsumerSupport, metav1.ConditionTrue, nil)
+}
+
 func GetProviderAPIServerDeployment(instance *ocsv1.StorageCluster) *appsv1.Deployment {
 
 	return &appsv1.Deployment{

controllers/storagecluster/reconcile.go

@@ -496,6 +496,17 @@ func (r *StorageClusterReconciler) reconcilePhases(
 
 			// mark operator upgradeable if and only if all storageclusters are ready
 			if r.clusters.AreOtherStorageClustersReady(instance) {
+
+				if instance.Spec.AllowRemoteStorageConsumers {
+					condition, err := r.OperatorCondition.Get(r.ctx)
+					if err != nil {
+						return reconcile.Result{}, err
+					}
+					// in provider mode we already set operatorcondition if not upgradeable, skip overwriting the condition
+					if condition.Reason == ocsProviderConsumerSupport && condition.Status == metav1.ConditionFalse {
+						return reconcile.Result{}, nil
+					}
+				}
 				returnErr := r.SetOperatorConditions(message, reason, metav1.ConditionTrue, nil)
 				if returnErr != nil {
 					return reconcile.Result{}, returnErr

controllers/storagecluster/storagecluster_controller.go

@@ -10,6 +10,7 @@ import (
 	conditionsv1 "github.com/openshift/custom-resource-status/conditions/v1"
 	"github.com/operator-framework/operator-lib/conditions"
 	ocsv1 "github.com/red-hat-storage/ocs-operator/v4/api/v1"
+	ocsv1alpha1 "github.com/red-hat-storage/ocs-operator/v4/api/v1alpha1"
 	"github.com/red-hat-storage/ocs-operator/v4/controllers/util"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	appsv1 "k8s.io/api/apps/v1"
@@ -188,6 +189,43 @@ func (r *StorageClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		},
 	)
 
+	storageConsumerPredicate := predicate.Funcs{
+		CreateFunc: func(e event.CreateEvent) bool {
+			return false
+		},
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			if e.ObjectOld == nil {
+				return true
+			}
+			old := e.ObjectOld.(*ocsv1alpha1.StorageConsumer)
+			new := e.ObjectOld.(*ocsv1alpha1.StorageConsumer)
+			return old.Status.Client.OperatorVersion != new.Status.Client.OperatorVersion
+		},
+	}
+	enqueueFromStorageConsumer := handler.EnqueueRequestsFromMapFunc(
+		func(_ context.Context, obj client.Object) []reconcile.Request {
+			_ = obj.(*ocsv1alpha1.StorageConsumer)
+			scList := &ocsv1.StorageClusterList{}
+			err := r.Client.List(r.ctx, scList, client.InNamespace(obj.GetNamespace()), client.Limit(1))
+			if err != nil {
+				r.Log.Error(err, "Unable to list StorageCluster objects")
+				return []reconcile.Request{}
+			}
+			if len(scList.Items) == 0 {
+				return []reconcile.Request{}
+			}
+			sc := scList.Items[0]
+			return []reconcile.Request{
+				{
+					NamespacedName: types.NamespacedName{
+						Name:      sc.Name,
+						Namespace: sc.Namespace,
+					},
+				},
+			}
+		},
+	)
+
 	builder := ctrl.NewControllerManagedBy(mgr).
 		For(&ocsv1.StorageCluster{}, builder.WithPredicates(scPredicate)).
 		Owns(&cephv1.CephCluster{}).
@@ -204,7 +242,8 @@ func (r *StorageClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
 				},
 			},
 			enqueueStorageClusterRequest,
-		)
+		).
+		Watches(&ocsv1alpha1.StorageConsumer{}, enqueueFromStorageConsumer, builder.WithPredicates(storageConsumerPredicate))
 	if os.Getenv("SKIP_NOOBAA_CRD_WATCH") != "true" {
 		builder.Owns(&nbv1.NooBaa{})
 	}