OTA-861: Set Upgradeable=False when there is an upgrade in progress

pkg/cvo/cvo.go

@@ -451,7 +451,9 @@ func (optr *Operator) Run(runContext context.Context, shutdownContext context.Co
 	resultChannelCount++
 	go func() {
 		defer utilruntime.HandleCrash()
-		wait.UntilWithContext(runContext, func(runContext context.Context) { optr.worker(runContext, optr.upgradeableQueue, optr.upgradeableSync) }, time.Second)
+		wait.UntilWithContext(runContext, func(runContext context.Context) {
+			optr.worker(runContext, optr.upgradeableQueue, optr.upgradeableSyncFunc(false))
+		}, time.Second)
 		resultChannel <- asyncResult{name: "upgradeable"}
 	}()
 
@@ -461,6 +463,10 @@ func (optr *Operator) Run(runContext context.Context, shutdownContext context.Co
 		wait.UntilWithContext(runContext, func(runContext context.Context) {
 			// run the worker, then when the queue is closed sync one final time to flush any pending status
 			optr.worker(runContext, optr.queue, func(runContext context.Context, key string) error { return optr.sync(runContext, key) })
+			// This is to ensure upgradeableCondition to be synced and thus to avoid the race caused by the throttle
+			if err := optr.upgradeableSyncFunc(true)(shutdownContext, optr.queueKey()); err != nil {
+				utilruntime.HandleError(fmt.Errorf("unable to perform final upgradeable sync: %v", err))
+			}
 			if err := optr.sync(shutdownContext, optr.queueKey()); err != nil {
 				utilruntime.HandleError(fmt.Errorf("unable to perform final sync: %v", err))
 			}
@@ -738,27 +744,29 @@ func (optr *Operator) availableUpdatesSync(ctx context.Context, key string) erro
 	return optr.syncAvailableUpdates(ctx, config)
 }
 
-// upgradeableSync is triggered on cluster version change (and periodic requeues) to
+// upgradeableSyncFunc returns a function that is triggered on cluster version change (and periodic requeues) to
 // sync upgradeableCondition. It only modifies cluster version.
-func (optr *Operator) upgradeableSync(_ context.Context, key string) error {
-	startTime := time.Now()
-	klog.V(2).Infof("Started syncing upgradeable %q", key)
-	defer func() {
-		klog.V(2).Infof("Finished syncing upgradeable %q (%v)", key, time.Since(startTime))
-	}()
+func (optr *Operator) upgradeableSyncFunc(ignoreThrottlePeriod bool) func(_ context.Context, key string) error {
+	return func(_ context.Context, key string) error {
+		startTime := time.Now()
+		klog.V(2).Infof("Started syncing upgradeable %q", key)
+		defer func() {
+			klog.V(2).Infof("Finished syncing upgradeable %q (%v)", key, time.Since(startTime))
+		}()
 
-	config, err := optr.cvLister.Get(optr.name)
-	if apierrors.IsNotFound(err) {
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-	if errs := validation.ValidateClusterVersion(config); len(errs) > 0 {
-		return nil
-	}
+		config, err := optr.cvLister.Get(optr.name)
+		if apierrors.IsNotFound(err) {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		if errs := validation.ValidateClusterVersion(config); len(errs) > 0 {
+			return nil
+		}
 
-	return optr.syncUpgradeable(config)
+		return optr.syncUpgradeable(config, ignoreThrottlePeriod)
+	}
 }
 
 // isOlderThanLastUpdate returns true if the cluster version is older than

pkg/cvo/cvo_test.go

@@ -3906,7 +3906,7 @@ func TestOperator_upgradeableSync(t *testing.T) {
 					waitForCm(t, cms)
 				}
 
-				err = optr.upgradeableSync(ctx, optr.queueKey())
+				err = optr.upgradeableSyncFunc(false)(ctx, optr.queueKey())
 				if err != nil && tt.wantErr == nil {
 					t.Fatalf("Operator.sync() unexpected error: %v", err)
 				}

pkg/cvo/upgradeable.go

@@ -62,8 +62,8 @@ func defaultUpgradeableCheckIntervals() upgradeableCheckIntervals {
 
 // syncUpgradeable synchronizes the upgradeable status only if the sufficient time passed since its last update. This
 // throttling period is dynamic and is driven by upgradeableCheckIntervals.
-func (optr *Operator) syncUpgradeable(cv *configv1.ClusterVersion) error {
-	if u := optr.getUpgradeable(); u != nil {
+func (optr *Operator) syncUpgradeable(cv *configv1.ClusterVersion, ignoreThrottlePeriod bool) error {
+	if u := optr.getUpgradeable(); u != nil && !ignoreThrottlePeriod {
 		throttleFor := optr.upgradeableCheckIntervals.throttlePeriod(cv)
 		if earliestNext := u.At.Add(throttleFor); time.Now().Before(earliestNext) {
 			klog.V(2).Infof("Upgradeability last checked %s ago, will not re-check until %s", time.Since(u.At), earliestNext.Format(time.RFC3339))
@@ -177,7 +177,7 @@ func (optr *Operator) getUpgradeable() *upgradeable {
 }
 
 type upgradeableCheck interface {
-	// returns a not-nil condition when the check fails.
+	// Check returns a not-nil condition that should be addressed before a minor level upgrade when the check fails.
 	Check() *configv1.ClusterOperatorStatusCondition
 }
 
@@ -263,6 +263,34 @@ func (check *clusterVersionOverridesUpgradeable) Check() *configv1.ClusterOperat
 	return cond
 }
 
+type upgradeInProgressUpgradeable struct {
+	name     string
+	cvLister configlistersv1.ClusterVersionLister
+}
+
+func (check *upgradeInProgressUpgradeable) Check() *configv1.ClusterOperatorStatusCondition {
+	cond := &configv1.ClusterOperatorStatusCondition{
+		Type:   "UpgradeableUpgradeInProgress",
+		Status: configv1.ConditionTrue,
+	}
+
+	cv, err := check.cvLister.Get(check.name)
+	if meta.IsNoMatchError(err) || apierrors.IsNotFound(err) {
+		return nil
+	} else if err != nil {
+		klog.Error(err)
+		return nil
+	}
+
+	if progressingCondition := resourcemerge.FindOperatorStatusCondition(cv.Status.Conditions, configv1.OperatorProgressing); progressingCondition != nil &&
+		progressingCondition.Status == configv1.ConditionTrue {
+		cond.Reason = progressingCondition.Reason
+		cond.Message = progressingCondition.Message
+		return cond
+	}
+	return nil
+}
+
 type clusterManifestDeleteInProgressUpgradeable struct {
 }
 
@@ -421,6 +449,7 @@ func (optr *Operator) defaultUpgradeableChecks() []upgradeableCheck {
 		},
 		&clusterOperatorsUpgradeable{coLister: optr.coLister},
 		&clusterManifestDeleteInProgressUpgradeable{},
+		&upgradeInProgressUpgradeable{name: optr.name, cvLister: optr.cvLister},
 	}
 }
 

pkg/cvo/upgradeable_test.go

@@ -4,7 +4,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
 	configv1 "github.com/openshift/api/config/v1"
+	"github.com/openshift/client-go/config/clientset/versioned/fake"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -57,3 +59,89 @@ func TestUpgradeableCheckIntervalsThrottlePeriod(t *testing.T) {
 		})
 	}
 }
+
+func TestUpgradeInProgressUpgradeable(t *testing.T) {
+	testCases := []struct {
+		name     string
+		cv       *configv1.ClusterVersion
+		expected *configv1.ClusterOperatorStatusCondition
+	}{
+		{
+			name: "empty conditions",
+			cv: &configv1.ClusterVersion{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "unit-test",
+				},
+			},
+		},
+		{
+			name: "progressing is true",
+			cv: &configv1.ClusterVersion{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "unit-test",
+				},
+				Status: configv1.ClusterVersionStatus{
+					Conditions: []configv1.ClusterOperatorStatusCondition{
+						{
+							Type:    configv1.OperatorProgressing,
+							Status:  configv1.ConditionTrue,
+							Reason:  "a",
+							Message: "b",
+						},
+					},
+				},
+			},
+			expected: &configv1.ClusterOperatorStatusCondition{
+				Type:    "UpgradeableUpgradeInProgress",
+				Status:  configv1.ConditionTrue,
+				Reason:  "a",
+				Message: "b",
+			},
+		},
+		{
+			name: "progressing is false",
+			cv: &configv1.ClusterVersion{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "unit-test",
+				},
+				Status: configv1.ClusterVersionStatus{
+					Conditions: []configv1.ClusterOperatorStatusCondition{
+						{
+							Type:    configv1.OperatorProgressing,
+							Status:  configv1.ConditionFalse,
+							Reason:  "a",
+							Message: "b",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "progressing is missing",
+			cv: &configv1.ClusterVersion{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "unit-test",
+				},
+				Status: configv1.ClusterVersionStatus{
+					Conditions: []configv1.ClusterOperatorStatusCondition{
+						{
+							Type:    configv1.OperatorProgressing,
+							Reason:  "a",
+							Message: "b",
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			client := fake.NewSimpleClientset(tc.cv)
+			unit := upgradeInProgressUpgradeable{name: "unit-test", cvLister: &clientCVLister{client: client}}
+			actual := unit.Check()
+			if diff := cmp.Diff(tc.expected, actual); diff != "" {
+				t.Errorf("%s differs from expected:\n%s", tc.name, diff)
+			}
+		})
+	}
+}