Merge pull request #243 from abhinavdahiya/feature-prevent-upgrade

Bug 1730401: prevent automatic upgrades for clusters with cluster operators reporting Upgradeable false

Author: openshift-merge-robot

pkg/cvo/cvo.go

@@ -37,6 +37,8 @@ import (
 	"github.com/openshift/cluster-version-operator/pkg/cvo/internal/dynamicclient"
 	"github.com/openshift/cluster-version-operator/pkg/internal"
 	"github.com/openshift/cluster-version-operator/pkg/payload"
+	"github.com/openshift/cluster-version-operator/pkg/payload/precondition"
+	preconditioncv "github.com/openshift/cluster-version-operator/pkg/payload/precondition/clusterversion"
 	"github.com/openshift/cluster-version-operator/pkg/verify"
 )
 
@@ -113,11 +115,18 @@ type Operator struct {
 	queue workqueue.RateLimitingInterface
 	// availableUpdatesQueue tracks checking for updates from the update server.
 	availableUpdatesQueue workqueue.RateLimitingInterface
+	// upgradeableQueue tracks checking for upgradeable.
+	upgradeableQueue workqueue.RateLimitingInterface
 
 	// statusLock guards access to modifying available updates
 	statusLock       sync.Mutex
 	availableUpdates *availableUpdates
 
+	// upgradeableStatusLock guards access to modifying Upgradeable conditions
+	upgradeableStatusLock sync.Mutex
+	upgradeable           *upgradeable
+	upgradeableChecks     []upgradeableCheck
+
 	// verifier, if provided, will be used to check an update before it is executed.
 	// Any error will prevent an update payload from being accessed.
 	verifier verify.Interface
@@ -171,11 +180,9 @@ func New(
 
 		queue:                 workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "clusterversion"),
 		availableUpdatesQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "availableupdates"),
+		upgradeableQueue:      workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "upgradeable"),
 	}
 
-	optr.proxyLister = proxyInformer.Lister()
-	proxyInformer.Informer().AddEventHandler(optr.eventHandler())
-
 	cvInformer.Informer().AddEventHandler(optr.eventHandler())
 
 	optr.coLister = coInformer.Lister()
@@ -184,8 +191,12 @@ func New(
 	optr.cvLister = cvInformer.Lister()
 	optr.cacheSynced = append(optr.cacheSynced, cvInformer.Informer().HasSynced)
 
+	optr.proxyLister = proxyInformer.Lister()
 	optr.cmConfigLister = cmConfigInformer.Lister().ConfigMaps(internal.ConfigNamespace)
 
+	// make sure this is initialized after all the listers are initialized
+	optr.upgradeableChecks = optr.defaultUpgradeableChecks()
+
 	if enableMetrics {
 		if err := optr.registerMetrics(coInformer.Informer()); err != nil {
 			panic(err)
@@ -225,9 +236,10 @@ func (optr *Operator) InitializeFromPayload(restConfig *rest.Config, burstRestCo
 
 	// after the verifier has been loaded, initialize the sync worker with a payload retriever
 	// which will consume the verifier
-	optr.configSync = NewSyncWorker(
+	optr.configSync = NewSyncWorkerWithPreconditions(
 		optr.defaultPayloadRetriever(),
 		NewResourceBuilder(restConfig, burstRestConfig, optr.coLister),
+		optr.defaultPreconditionChecks(),
 		optr.minimumUpdateCheckInterval,
 		wait.Backoff{
 			Duration: time.Second * 10,
@@ -263,6 +275,7 @@ func (optr *Operator) Run(ctx context.Context, workers int) {
 	go runThrottledStatusNotifier(stopCh, optr.statusInterval, 2, optr.configSync.StatusCh(), func() { optr.queue.Add(optr.queueKey()) })
 	go optr.configSync.Start(ctx, 16)
 	go wait.Until(func() { optr.worker(optr.availableUpdatesQueue, optr.availableUpdatesSync) }, time.Second, stopCh)
+	go wait.Until(func() { optr.worker(optr.upgradeableQueue, optr.upgradeableSync) }, time.Second, stopCh)
 	go wait.Until(func() {
 		defer close(workerStopCh)
 
@@ -292,10 +305,12 @@ func (optr *Operator) eventHandler() cache.ResourceEventHandler {
 		AddFunc: func(obj interface{}) {
 			optr.queue.Add(workQueueKey)
 			optr.availableUpdatesQueue.Add(workQueueKey)
+			optr.upgradeableQueue.Add(workQueueKey)
 		},
 		UpdateFunc: func(old, new interface{}) {
 			optr.queue.Add(workQueueKey)
 			optr.availableUpdatesQueue.Add(workQueueKey)
+			optr.upgradeableQueue.Add(workQueueKey)
 		},
 		DeleteFunc: func(obj interface{}) {
 			optr.queue.Add(workQueueKey)
@@ -432,6 +447,29 @@ func (optr *Operator) availableUpdatesSync(key string) error {
 	return optr.syncAvailableUpdates(config)
 }
 
+// upgradeableSync is triggered on cluster version change (and periodic requeues) to
+// sync upgradeableCondition. It only modifies cluster version.
+func (optr *Operator) upgradeableSync(key string) error {
+	startTime := time.Now()
+	klog.V(4).Infof("Started syncing upgradeable %q (%v)", key, startTime)
+	defer func() {
+		klog.V(4).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
+	}
+
+	return optr.syncUpgradeable(config)
+}
+
 // isOlderThanLastUpdate returns true if the cluster version is older than
 // the last update we saw.
 func (optr *Operator) isOlderThanLastUpdate(config *configv1.ClusterVersion) bool {
@@ -608,3 +646,9 @@ func hasReachedLevel(cv *configv1.ClusterVersion, desired configv1.Update) bool
 	}
 	return desired.Image == cv.Status.History[0].Image
 }
+
+func (optr *Operator) defaultPreconditionChecks() precondition.List {
+	return []precondition.Precondition{
+		preconditioncv.NewUpgradeable(optr.cvLister),
+	}
+}