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

@@ -8,6 +8,9 @@ import (
 	"sync"
 	"time"
 
+	runtimeclient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
+
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -53,6 +56,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 +217,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 +324,17 @@ func New(
 
 	optr.configuration = configuration.NewClusterVersionOperatorConfiguration(operatorClient, operatorInformerFactory)
 
+	c, err := runtimeclient.New(config.GetConfigOrDie(), runtimeclient.Options{})
+	if err != nil {
+		return nil, err
+	}
+	optr.proposalController = proposal.NewController(func() ([]configv1.Release, []configv1.ConditionalUpdate, error) {
+		if optr.availableUpdates == nil {
+			return nil, nil, nil
+		}
+		return optr.availableUpdates.Updates, optr.availableUpdates.ConditionalUpdates, nil
+	}, c)
+
 	return optr, nil
 }
 
@@ -470,6 +488,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 +544,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 +627,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 +1224,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,107 @@
+package proposal
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	corev1 "k8s.io/api/core/v1"
+	kapierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kutilerrors "k8s.io/apimachinery/pkg/util/errors"
+	"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"
+	proposalv1alpha1 "github.com/openshift/cluster-version-operator/pkg/proposal/api/v1alpha1"
+)
+
+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]
+
+	updatesGetterFunc UpdatesGetterFunc
+
+	client client.Client
+}
+
+const controllerName = "proposal-lifecycle-controller"
+
+type UpdatesGetterFunc func() ([]configv1.Release, []configv1.ConditionalUpdate, error)
+
+// NewController returns Controller to manage Proposals.
+func NewController(updatesGetterFunc UpdatesGetterFunc, client client.Client) *Controller {
+	return &Controller{
+		queueKey: fmt.Sprintf("ClusterVersionOperator/%s", controllerName),
+		queue: workqueue.NewTypedRateLimitingQueueWithConfig[any](
+			workqueue.DefaultTypedControllerRateLimiter[any](),
+			workqueue.TypedRateLimitingQueueConfig[any]{Name: controllerName}),
+		updatesGetterFunc: updatesGetterFunc,
+		client:            client,
+	}
+}
+
+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(ctx 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, conditionalUpdates, err := c.updatesGetterFunc()
+	if err != nil {
+		klog.Errorf("Error getting available updates: %v", err)
+		return err
+	}
+	klog.Infof("Got available updates: %#v", updates)
+	klog.Infof("Got available updates: %#v", conditionalUpdates)
+
+	proposals := getProposals(updates, conditionalUpdates)
+
+	var errs []error
+	for _, proposal := range proposals {
+		err := c.client.Create(ctx, proposal)
+		if err != nil {
+			if !kapierrors.IsAlreadyExists(err) {
+				errs = append(errs, err)
+			}
+		}
+	}
+
+	return kutilerrors.NewAggregate(errs)
+}
+
+// TODO: make it real
+func getProposals(updates []configv1.Release, conditionalUpdates []configv1.ConditionalUpdate) []*proposalv1alpha1.Proposal {
+	return []*proposalv1alpha1.Proposal{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      controllerName,
+				Namespace: metav1.NamespaceAll,
+			},
+			Spec: proposalv1alpha1.ProposalSpec{
+				Request: "some-request",
+				WorkflowRef: corev1.LocalObjectReference{
+					Name: "ota-advisory",
+				},
+			},
+		},
+	}
+}