OTA-1966: Init the Proposal Lifecycle Controller

Author: hongkailiu

pkg/cvo/availableupdates.go

@@ -182,6 +182,8 @@ func (optr *Operator) syncAvailableUpdates(ctx context.Context, config *configv1
 
 	// queue optr.sync() to update ClusterVersion status
 	optr.queue.Add(queueKey)
+	// queue optr.proposalController.Sync() to manage proposals
+	optr.proposalController.Queue().Add(optr.proposalController.QueueKey())
 	return nil
 }
 

pkg/cvo/cvo.go

@@ -53,6 +53,7 @@ import (
 	"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/proposal"
 	"github.com/openshift/cluster-version-operator/pkg/risk"
 	"github.com/openshift/cluster-version-operator/pkg/risk/alert"
 )
@@ -213,6 +214,9 @@ type Operator struct {
 	// risks holds update-risk source (in-cluster alerts, etc.)
 	// that will be aggregated into conditional update risks.
 	risks risk.Source
+
+	// proposalController, if enabled, watch available updates and manage proposals
+	proposalController *proposal.Controller
 }
 
 // New returns a new cluster version operator.
@@ -317,6 +321,13 @@ func New(
 
 	optr.configuration = configuration.NewClusterVersionOperatorConfiguration(operatorClient, operatorInformerFactory)
 
+	optr.proposalController = proposal.NewController(func() ([]configv1.Release, error) {
+		if optr.availableUpdates == nil {
+			return nil, nil
+		}
+		return optr.availableUpdates.Updates, nil
+	})
+
 	return optr, nil
 }
 
@@ -470,6 +481,7 @@ func (optr *Operator) Run(runContext context.Context, shutdownContext context.Co
 	defer optr.availableUpdatesQueue.ShutDown()
 	defer optr.upgradeableQueue.ShutDown()
 	defer optr.configuration.Queue().ShutDown()
+	defer optr.proposalController.Queue().ShutDown()
 	stopCh := runContext.Done()
 
 	klog.Infof("Starting ClusterVersionOperator with minimum reconcile period %s", optr.minimumUpdateCheckInterval)
@@ -525,6 +537,19 @@ func (optr *Operator) Run(runContext context.Context, shutdownContext context.Co
 		klog.Infof("The ClusterVersionOperatorConfiguration feature gate is disabled or HyperShift is detected; the configuration sync routine will not run.")
 	}
 
+	if optr.shouldEnableProposalController() {
+		resultChannelCount++
+		go func() {
+			defer utilruntime.HandleCrash()
+			wait.UntilWithContext(runContext, func(runContext context.Context) {
+				optr.worker(runContext, optr.proposalController.Queue(), optr.proposalController.Sync)
+			}, time.Second)
+			resultChannel <- asyncResult{name: "proposal controller"}
+		}()
+	} else {
+		klog.Infof("The proposal controller is disabled.")
+	}
+
 	resultChannelCount++
 	go func() {
 		defer utilruntime.HandleCrash()
@@ -595,6 +620,7 @@ func (optr *Operator) Run(runContext context.Context, shutdownContext context.Co
 			optr.availableUpdatesQueue.ShutDown()
 			optr.upgradeableQueue.ShutDown()
 			optr.configuration.Queue().ShutDown()
+			optr.proposalController.Queue().ShutDown()
 		}
 	}
 
@@ -1191,3 +1217,9 @@ func (optr *Operator) shouldReconcileAcceptRisks() bool {
 	// HyperShift will be supported later if needed
 	return optr.enabledCVOFeatureGates.AcceptRisks() && !optr.hypershift
 }
+
+// shouldEnableProposalController returns whether the CVO should enable the proposal controller
+func (optr *Operator) shouldEnableProposalController() bool {
+	// TODO: guarded by some feature gate?
+	return true
+}

pkg/proposal/controller.go

@@ -0,0 +1,65 @@
+package proposal
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+
+	configv1 "github.com/openshift/api/config/v1"
+
+	i "github.com/openshift/cluster-version-operator/pkg/internal"
+)
+
+type Controller struct {
+	queueKey string
+	// queue tracks checking for the CVO configuration.
+	//
+	// The type any is used to comply with the worker method of the cvo.Operator struct.
+	queue workqueue.TypedRateLimitingInterface[any]
+
+	availableUpdatesGetterFunc AvailableUpdatesGetterFunc
+}
+
+const controllerName = "proposal-lifecycle-controller"
+
+type AvailableUpdatesGetterFunc func() ([]configv1.Release, error)
+
+// NewController returns Controller to manage Proposals.
+func NewController(availableUpdatesGetterFunc AvailableUpdatesGetterFunc) *Controller {
+	return &Controller{
+		queueKey: fmt.Sprintf("ClusterVersionOperator/%s", controllerName),
+		queue: workqueue.NewTypedRateLimitingQueueWithConfig[any](
+			workqueue.DefaultTypedControllerRateLimiter[any](),
+			workqueue.TypedRateLimitingQueueConfig[any]{Name: controllerName}),
+		availableUpdatesGetterFunc: availableUpdatesGetterFunc,
+	}
+}
+
+func (c *Controller) Queue() workqueue.TypedRateLimitingInterface[any] {
+	return c.queue
+}
+
+func (c *Controller) QueueKey() string {
+	return c.queueKey
+}
+
+// Sync ensures:
+// TODO:
+func (c *Controller) Sync(_ context.Context, key string) error {
+	startTime := time.Now()
+	klog.V(i.Normal).Infof("Started syncing CVO configuration %q", key)
+	defer func() {
+		klog.V(i.Normal).Infof("Finished syncing CVO configuration (%v)", time.Since(startTime))
+	}()
+
+	updates, err := c.availableUpdatesGetterFunc()
+	if err != nil {
+		klog.Errorf("Error getting available updates: %v", err)
+		return err
+	}
+	klog.Infof("Got available updates: %#v", updates)
+	return nil
+}