pkg: Use risk.Source framework to feed Upgradeable

This simplifies the complex cvo package by shifting some logic into
small, narrowly scoped helper packages.  That makes for easier unit
testing of that logic, and less complication when reading the
remaining cvo package code.

Using the risk.Source framework with the new aggregate implementation
allows Upgradeable risks to feed into the conditionalUpdateRisks
framework.  This should help reduce confusion some users experience
when the see Upgradeable=False and think it applies to all updates
(when in reality it applies to major and minor bumps, but does not
apply to patch updates, especially since 6f8f984, OTA-1860: Stop
blocking patch updates when cluster version overrides are set,
2026-02-10, #1314).  It will also allow structured access to
individual risks that are reported via this pipeline, while
Upgradeable had an aggregated message string only consumable by
humans.

I'm also removing internal.UpgradeableAdminAckRequired and similar
from ClusterVersion status.conditions.  While these were visible by
end users, they were mostly a mechanism for passing data from one part
of the CVO to another.  There are no known consumers outside of the
CVO.

With the refactor providing clearer abstraction layers, I could drop
the entire upgradeableCheckIntervals structure.  Major touches in the
history of that structure were bd05174 (pkg/cvo/upgradeable.go:
Sync Upgradeable status of the CVO more often, 2022-08-02, #808) and
cc94c95 (pkg/cvo/upgradeable: refactor throttling, 2023-01-11, #882).
The new approach is event-driven via the informer handlers for almost
all of the new risk.Source implementations.  The one source that still
relies on polling is the deletion source, where it's watching a
CVO-side global variable, and not a Kube-side informer.  It will still
sync whenever ClusterVersion status is updated though, which should be
often enough for this usually-transient situation.  If folks want
lower latency, follow up work could add informer-style handler
callbacks to the deletion global.

The addition of '.' and '-' to validName in pkg/risk allows for names
like ClusterOperatorUpgradeable-test-operator in the Upgradeable test,
composed with the ClusterOperator name.  That requires us to extend
the regular expression to allow the characters supported by
ClusterOperator names [2].  We could alternatively build a system that
CamelCased ClusterOperator names, but that seemed difficult, and not
worth the effort.

The changeCallback in unit tests is unbuffered, because I want to know
exactly how often that callback is being called.  If a logic change in
the packages causes additional callback calls, the unbuffered channel
will block them from sending the message, deadlock the informer for
that handler [1], and break later test-cases.

[1]: https://github.com/kubernetes/client-go/blob/v0.35.3/tools/cache/shared_informer.go#L1052-L1086
[2]: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#dns-subdomain-names

Author: wking

pkg/cvo/cvo.go

@@ -57,7 +57,13 @@ import (
 	preconditioncv "github.com/openshift/cluster-version-operator/pkg/payload/precondition/clusterversion"
 	"github.com/openshift/cluster-version-operator/pkg/proposal"
 	"github.com/openshift/cluster-version-operator/pkg/risk"
+	"github.com/openshift/cluster-version-operator/pkg/risk/adminack"
+	"github.com/openshift/cluster-version-operator/pkg/risk/aggregate"
 	"github.com/openshift/cluster-version-operator/pkg/risk/alert"
+	deletionrisk "github.com/openshift/cluster-version-operator/pkg/risk/deletion"
+	"github.com/openshift/cluster-version-operator/pkg/risk/overrides"
+	updatingrisk "github.com/openshift/cluster-version-operator/pkg/risk/updating"
+	upgradeablerisk "github.com/openshift/cluster-version-operator/pkg/risk/upgradeable"
 )
 
 const (
@@ -133,21 +139,13 @@ type Operator struct {
 	queue workqueue.TypedRateLimitingInterface[any]
 	// availableUpdatesQueue tracks checking for updates from the update server.
 	availableUpdatesQueue workqueue.TypedRateLimitingInterface[any]
-	// upgradeableQueue tracks checking for upgradeable.
-	upgradeableQueue workqueue.TypedRateLimitingInterface[any]
 
 	// 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
-
-	// upgradeableCheckIntervals drives minimal intervals between Upgradeable status
-	// synchronization
-	upgradeableCheckIntervals upgradeableCheckIntervals
+	// upgradeable tracks the risks that feed the ClusterVersion Upgradeable condition.
+	upgradeable risk.Source
 
 	// conditionRegistry is used to evaluate whether a particular condition is risky or not.
 	conditionRegistry clusterconditions.ConditionRegistry
@@ -264,7 +262,6 @@ func New(
 
 		statusInterval:             15 * time.Second,
 		minimumUpdateCheckInterval: minimumInterval,
-		upgradeableCheckIntervals:  defaultUpgradeableCheckIntervals(),
 		payloadDir:                 overridePayloadDir,
 		updateService:              updateService,
 
@@ -274,13 +271,11 @@ func New(
 		eventRecorder:         eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: namespace}),
 		queue:                 workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[any](), workqueue.TypedRateLimitingQueueConfig[any]{Name: "clusterversion"}),
 		availableUpdatesQueue: workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[any](), workqueue.TypedRateLimitingQueueConfig[any]{Name: "availableupdates"}),
-		upgradeableQueue:      workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[any](), workqueue.TypedRateLimitingQueueConfig[any]{Name: "upgradeable"}),
 
 		hypershift:                hypershift,
 		exclude:                   exclude,
 		clusterProfile:            clusterProfile,
 		conditionRegistry:         standard.NewConditionRegistry(promqlTarget),
-		risks:                     alert.New("Alert", promqlTarget),
 		injectClusterIdIntoPromQL: injectClusterIdIntoPromQL,
 
 		requiredFeatureSet:          featureSet,
@@ -293,12 +288,6 @@ func New(
 	if _, err := cvInformer.Informer().AddEventHandler(optr.clusterVersionEventHandler()); err != nil {
 		return nil, err
 	}
-	if _, err := cmConfigInformer.Informer().AddEventHandler(optr.adminAcksEventHandler()); err != nil {
-		return nil, err
-	}
-	if _, err := cmConfigManagedInformer.Informer().AddEventHandler(optr.adminGatesEventHandler()); err != nil {
-		return nil, err
-	}
 	if _, err := coInformer.Informer().AddEventHandler(optr.clusterOperatorEventHandler()); err != nil {
 		return nil, err
 	}
@@ -320,7 +309,19 @@ func New(
 	optr.cacheSynced = append(optr.cacheSynced, featureGateInformer.Informer().HasSynced)
 
 	// make sure this is initialized after all the listers are initialized
-	optr.upgradeableChecks = optr.defaultUpgradeableChecks()
+	riskSourceCallback := func() { optr.availableUpdatesQueue.Add(optr.queueKey()) }
+	optr.upgradeable = aggregate.New(
+		updatingrisk.New("ClusterVersionUpdating", optr.name, cvInformer, riskSourceCallback),
+		overrides.New("ClusterVersionOverrides", optr.name, cvInformer, riskSourceCallback),
+		deletionrisk.New("ResourceDeletionInProgress", optr.currentVersion),
+		adminack.New("AdminAck", optr.currentVersion, cmConfigManagedInformer, cmConfigInformer, riskSourceCallback),
+		upgradeablerisk.New("ClusterOperatorUpgradeable", optr.currentVersion, coInformer, riskSourceCallback),
+	)
+
+	optr.risks = aggregate.New(
+		alert.New("Alert", promqlTarget),
+		optr.upgradeable,
+	)
 
 	optr.configuration = configuration.NewClusterVersionOperatorConfiguration(operatorClient, operatorInformerFactory)
 
@@ -483,7 +484,6 @@ func loadConfigMapVerifierDataFromUpdate(update *payload.Update, clientBuilder s
 func (optr *Operator) Run(runContext context.Context, shutdownContext context.Context) error {
 	defer optr.queue.ShutDown()
 	defer optr.availableUpdatesQueue.ShutDown()
-	defer optr.upgradeableQueue.ShutDown()
 	defer optr.configuration.Queue().ShutDown()
 	defer optr.proposalController.Queue().ShutDown()
 	stopCh := runContext.Done()
@@ -554,25 +554,12 @@ func (optr *Operator) Run(runContext context.Context, shutdownContext context.Co
 		klog.Infof("The proposal controller is disabled.")
 	}
 
-	resultChannelCount++
-	go func() {
-		defer utilruntime.HandleCrash()
-		wait.UntilWithContext(runContext, func(runContext context.Context) {
-			optr.worker(runContext, optr.upgradeableQueue, optr.upgradeableSyncFunc(false))
-		}, time.Second)
-		resultChannel <- asyncResult{name: "upgradeable"}
-	}()
-
 	resultChannelCount++
 	go func() {
 		defer utilruntime.HandleCrash()
 		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))
 			}
@@ -622,7 +609,6 @@ func (optr *Operator) Run(runContext context.Context, shutdownContext context.Co
 			shutdown = true
 			optr.queue.ShutDown()
 			optr.availableUpdatesQueue.ShutDown()
-			optr.upgradeableQueue.ShutDown()
 			optr.configuration.Queue().ShutDown()
 			optr.proposalController.Queue().ShutDown()
 		}
@@ -643,12 +629,10 @@ func (optr *Operator) clusterVersionEventHandler() cache.ResourceEventHandler {
 		AddFunc: func(_ interface{}) {
 			optr.queue.Add(workQueueKey)
 			optr.availableUpdatesQueue.Add(workQueueKey)
-			optr.upgradeableQueue.Add(workQueueKey)
 		},
 		UpdateFunc: func(_, _ interface{}) {
 			optr.queue.Add(workQueueKey)
 			optr.availableUpdatesQueue.Add(workQueueKey)
-			optr.upgradeableQueue.Add(workQueueKey)
 		},
 		DeleteFunc: func(_ interface{}) {
 			optr.queue.Add(workQueueKey)
@@ -859,31 +843,6 @@ func (optr *Operator) availableUpdatesSync(ctx context.Context, key string) erro
 	return optr.syncAvailableUpdates(ctx, config)
 }
 
-// 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) 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, optr.shouldReconcileAcceptRisks()); len(errs) > 0 {
-			return nil
-		}
-
-		return optr.syncUpgradeable(config, ignoreThrottlePeriod)
-	}
-}
-
 // isOlderThanLastUpdate returns true if the cluster version is older than
 // the last update we saw.
 func (optr *Operator) isOlderThanLastUpdate(config *configv1.ClusterVersion) bool {

pkg/cvo/cvo_scenarios_test.go

@@ -29,6 +29,8 @@ import (
 	"github.com/openshift/client-go/config/clientset/versioned/fake"
 	"github.com/openshift/library-go/pkg/manifest"
 
+	"github.com/openshift/cluster-version-operator/pkg/clusterconditions"
+	"github.com/openshift/cluster-version-operator/pkg/clusterconditions/always"
 	"github.com/openshift/cluster-version-operator/pkg/featuregates"
 	"github.com/openshift/cluster-version-operator/pkg/internal"
 	"github.com/openshift/cluster-version-operator/pkg/payload"
@@ -114,13 +116,16 @@ func setupCVOTest(payloadDir string) (*Operator, map[string]apiruntime.Object, *
 		fmt.Printf("Cannot create cluster version object, err: %#v\n", err)
 	}
 
+	registry := clusterconditions.NewConditionRegistry()
+	registry.Register("Always", &always.Always{})
 	o := &Operator{
 		namespace:              "test",
 		name:                   "version",
 		queue:                  workqueue.NewTypedRateLimitingQueueWithConfig[any](workqueue.DefaultTypedControllerRateLimiter[any](), workqueue.TypedRateLimitingQueueConfig[any]{Name: "cvo-loop-test"}),
 		client:                 client,
 		enabledCVOFeatureGates: featuregates.DefaultCvoGates("version"),
 		cvLister:               &clientCVLister{client: client},
+		conditionRegistry:      registry,
 		exclude:                "exclude-test",
 		eventRecorder:          record.NewFakeRecorder(100),
 		clusterProfile:         payload.DefaultClusterProfile,