config: Introduce configuration provider interface

Introduce a configuration provider interface. The goal of the interface
is to abstract the underlying system of fetching the CVO configuration.
Be it a Kubernetes API server, or a configuration file. For the moment,
the Kubernetes API server is the sole provider.

Refactor the code to adhere to the new logic.

Apply a default configuration on a NotFound error (was previously noted
as a TODO in the source code).

Author: DavidHurta

pkg/cvo/configuration/configuration.go

@@ -5,44 +5,48 @@ import (
 	"fmt"
 	"time"
 
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
 	operatorv1 "github.com/openshift/api/operator/v1"
-	operatorv1alpha1 "github.com/openshift/api/operator/v1alpha1"
 	operatorclientset "github.com/openshift/client-go/operator/clientset/versioned"
-	cvoclientv1alpha1 "github.com/openshift/client-go/operator/clientset/versioned/typed/operator/v1alpha1"
 	operatorexternalversions "github.com/openshift/client-go/operator/informers/externalversions"
-	operatorlistersv1alpha1 "github.com/openshift/client-go/operator/listers/operator/v1alpha1"
 	"github.com/openshift/library-go/pkg/operator/loglevel"
 
 	i "github.com/openshift/cluster-version-operator/pkg/internal"
 )
 
 const ClusterVersionOperatorConfigurationName = "cluster"
 
+const defaultQueueKey = "ClusterVersionOperator/" + ClusterVersionOperatorConfigurationName
+
 type configuration struct {
 	lastObservedGeneration int64
 	desiredLogLevel        operatorv1.LogLevel
 }
 
+func defaultConfiguration() configuration {
+	return configuration{
+		lastObservedGeneration: 0,
+		desiredLogLevel:        operatorv1.Normal,
+	}
+}
+
+type ConfigProvider interface {
+	start(ctx context.Context) error
+	get(context.Context) (configuration, error)
+}
+
 type ClusterVersionOperatorConfiguration 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]
 
-	client  cvoclientv1alpha1.ClusterVersionOperatorInterface
-	lister  operatorlistersv1alpha1.ClusterVersionOperatorLister
-	factory operatorexternalversions.SharedInformerFactory
-
 	started bool
 
 	configuration configuration
+	provider      ConfigProvider
 
 	// Function to handle an update to the internal configuration.
 	//
@@ -57,28 +61,9 @@ func (config *ClusterVersionOperatorConfiguration) Queue() workqueue.TypedRateLi
 	return config.queue
 }
 
-// clusterVersionOperatorEventHandler queues an update for the cluster version operator on any change to the given object.
-// Callers should use this with an informer.
-func (config *ClusterVersionOperatorConfiguration) clusterVersionOperatorEventHandler() cache.ResourceEventHandler {
-	return cache.ResourceEventHandlerFuncs{
-		AddFunc: func(_ interface{}) {
-			config.queue.Add(config.queueKey)
-			klog.V(i.Debug).Infof("ClusterVersionOperator resource was added; queuing a sync")
-		},
-		UpdateFunc: func(_, _ interface{}) {
-			config.queue.Add(config.queueKey)
-			klog.V(i.Debug).Infof("ClusterVersionOperator resource was modified or resync period has passed; queuing a sync")
-		},
-		DeleteFunc: func(_ interface{}) {
-			config.queue.Add(config.queueKey)
-			klog.V(i.Debug).Infof("ClusterVersionOperator resource was deleted; queuing a sync")
-		},
-	}
-}
-
 // NewClusterVersionOperatorConfiguration returns ClusterVersionOperatorConfiguration, which might be used
 // to synchronize with the ClusterVersionOperator resource.
-func NewClusterVersionOperatorConfiguration(client operatorclientset.Interface, factory operatorexternalversions.SharedInformerFactory) *ClusterVersionOperatorConfiguration {
+func NewClusterVersionOperatorConfiguration() *ClusterVersionOperatorConfiguration {
 	var desiredLogLevel operatorv1.LogLevel
 	if currentLogLevel, notFound := loglevel.GetLogLevel(); notFound {
 		klog.Warningf("The current log level could not be found; assuming the 'Normal' level is the currently desired")
@@ -88,34 +73,28 @@ func NewClusterVersionOperatorConfiguration(client operatorclientset.Interface,
 	}
 
 	return &ClusterVersionOperatorConfiguration{
-		queueKey: fmt.Sprintf("ClusterVersionOperator/%s", ClusterVersionOperatorConfigurationName),
 		queue: workqueue.NewTypedRateLimitingQueueWithConfig[any](
 			workqueue.DefaultTypedControllerRateLimiter[any](),
 			workqueue.TypedRateLimitingQueueConfig[any]{Name: "configuration"}),
-		client:        client.OperatorV1alpha1().ClusterVersionOperators(),
-		factory:       factory,
 		configuration: configuration{desiredLogLevel: desiredLogLevel},
 		handler:       func(c configuration) error { return applyLogLevel(c.desiredLogLevel) },
 	}
 }
 
+func (config *ClusterVersionOperatorConfiguration) UsingKubeAPIServer(client operatorclientset.Interface, factory operatorexternalversions.SharedInformerFactory) *ClusterVersionOperatorConfiguration {
+	config.provider = newConfigProviderKubeAPIServer(client, factory, config.queue)
+	return config
+}
+
 // Start initializes and starts the configuration's informers. Must be run before Sync is called.
 // Blocks until informers caches are synchronized or the context is cancelled.
 func (config *ClusterVersionOperatorConfiguration) Start(ctx context.Context) error {
-	informer := config.factory.Operator().V1alpha1().ClusterVersionOperators()
-	if _, err := informer.Informer().AddEventHandler(config.clusterVersionOperatorEventHandler()); err != nil {
-		return err
+	if config.provider == nil {
+		return fmt.Errorf("the configuration provider must be initialized")
 	}
-	config.lister = informer.Lister()
-
-	config.factory.Start(ctx.Done())
-	synced := config.factory.WaitForCacheSync(ctx.Done())
-	for _, ok := range synced {
-		if !ok {
-			return fmt.Errorf("caches failed to sync: %w", ctx.Err())
-		}
+	if err := config.provider.start(ctx); err != nil {
+		return err
 	}
-
 	config.started = true
 	return nil
 }
@@ -130,36 +109,14 @@ func (config *ClusterVersionOperatorConfiguration) Sync(ctx context.Context, key
 		klog.V(i.Normal).Infof("Finished syncing CVO configuration (%v)", time.Since(startTime))
 	}()
 
-	desiredConfig, err := config.lister.Get(ClusterVersionOperatorConfigurationName)
-	if apierrors.IsNotFound(err) {
-		// TODO: Set default values
-		return nil
-	}
+	desiredConfig, err := config.provider.get(ctx)
 	if err != nil {
 		return err
 	}
-	return config.sync(ctx, desiredConfig)
-}
-
-// sync synchronizes the local configuration based on the desired configuration
-// and updates the status of the Kubernetes resource if needed.
-//
-// desiredConfig is a read-only representation of the desired configuration.
-func (config *ClusterVersionOperatorConfiguration) sync(ctx context.Context, desiredConfig *operatorv1alpha1.ClusterVersionOperator) error {
-	if desiredConfig.Status.ObservedGeneration != desiredConfig.Generation {
-		newConfig := desiredConfig.DeepCopy()
-		newConfig.Status.ObservedGeneration = desiredConfig.Generation
-		_, err := config.client.UpdateStatus(ctx, newConfig, metav1.UpdateOptions{})
-		if err != nil {
-			return fmt.Errorf("failed to update the ClusterVersionOperator resource: %w", err)
-		}
-	}
-
-	config.configuration.lastObservedGeneration = desiredConfig.Generation
-	config.configuration.desiredLogLevel = desiredConfig.Spec.OperatorLogLevel
-	if config.configuration.desiredLogLevel == "" {
-		config.configuration.desiredLogLevel = operatorv1.Normal
+	if desiredConfig.desiredLogLevel == "" {
+		desiredConfig.desiredLogLevel = operatorv1.Normal
 	}
+	config.configuration = desiredConfig
 
 	if config.handler != nil {
 		return config.handler(config.configuration)

pkg/cvo/configuration/configuration_test.go

@@ -169,7 +169,7 @@ func TestClusterVersionOperatorConfiguration_sync(t *testing.T) {
 			client := operatorclientsetfake.NewClientset(&tt.config)
 			factory := operatorexternalversions.NewSharedInformerFactoryWithOptions(client, time.Minute)
 
-			configController := NewClusterVersionOperatorConfiguration(client, factory)
+			configController := NewClusterVersionOperatorConfiguration().UsingKubeAPIServer(client, factory)
 
 			called := false
 			configController.handler = func(_ configuration) error {
@@ -265,7 +265,7 @@ func TestClusterVersionOperatorConfiguration_Sync(t *testing.T) {
 			}
 
 			factory := operatorexternalversions.NewSharedInformerFactoryWithOptions(client, time.Minute)
-			cvoConfiguration := NewClusterVersionOperatorConfiguration(client, factory)
+			cvoConfiguration := NewClusterVersionOperatorConfiguration().UsingKubeAPIServer(client, factory)
 			defer cvoConfiguration.queue.ShutDown()
 
 			ctx, cancelFunc := context.WithDeadline(context.Background(), time.Now().Add(time.Minute))

pkg/cvo/configuration/resource.go

@@ -0,0 +1,98 @@
+package configuration
+
+import (
+	"context"
+	"fmt"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+
+	operatorclientset "github.com/openshift/client-go/operator/clientset/versioned"
+	cvoclientv1alpha1 "github.com/openshift/client-go/operator/clientset/versioned/typed/operator/v1alpha1"
+	operatorexternalversions "github.com/openshift/client-go/operator/informers/externalversions"
+	operatorlistersv1alpha1 "github.com/openshift/client-go/operator/listers/operator/v1alpha1"
+
+	i "github.com/openshift/cluster-version-operator/pkg/internal"
+)
+
+type ConfigProviderKubeAPIServer struct {
+	queueAdd func(any)
+	queueKey string
+
+	client  cvoclientv1alpha1.ClusterVersionOperatorInterface
+	lister  operatorlistersv1alpha1.ClusterVersionOperatorLister
+	factory operatorexternalversions.SharedInformerFactory
+}
+
+func newConfigProviderKubeAPIServer(client operatorclientset.Interface, factory operatorexternalversions.SharedInformerFactory, queue workqueue.TypedRateLimitingInterface[any]) *ConfigProviderKubeAPIServer {
+	return &ConfigProviderKubeAPIServer{
+		queueAdd: queue.Add,
+		queueKey: defaultQueueKey,
+		client:   client.OperatorV1alpha1().ClusterVersionOperators(),
+		factory:  factory,
+	}
+}
+
+func (c *ConfigProviderKubeAPIServer) start(ctx context.Context) error {
+	informer := c.factory.Operator().V1alpha1().ClusterVersionOperators()
+	if _, err := informer.Informer().AddEventHandler(c.clusterVersionOperatorEventHandler()); err != nil {
+		return err
+	}
+	c.lister = informer.Lister()
+
+	c.factory.Start(ctx.Done())
+	synced := c.factory.WaitForCacheSync(ctx.Done())
+	for _, ok := range synced {
+		if !ok {
+			return fmt.Errorf("caches failed to sync: %w", ctx.Err())
+		}
+	}
+	return nil
+}
+
+func (c *ConfigProviderKubeAPIServer) get(ctx context.Context) (configuration, error) {
+	config := defaultConfiguration()
+
+	desiredConfig, err := c.lister.Get(ClusterVersionOperatorConfigurationName)
+	if apierrors.IsNotFound(err) {
+		return config, nil
+	}
+	if err != nil {
+		return config, err
+	}
+
+	if desiredConfig.Status.ObservedGeneration != desiredConfig.Generation {
+		newConfig := desiredConfig.DeepCopy()
+		newConfig.Status.ObservedGeneration = desiredConfig.Generation
+		_, err := c.client.UpdateStatus(ctx, newConfig, metav1.UpdateOptions{})
+		if err != nil {
+			return config, fmt.Errorf("failed to update the ClusterVersionOperator resource: %w", err)
+		}
+	}
+
+	config.lastObservedGeneration = desiredConfig.Generation
+	config.desiredLogLevel = desiredConfig.Spec.OperatorLogLevel
+	return config, nil
+}
+
+// clusterVersionOperatorEventHandler queues an update for the cluster version operator on any change to the given object.
+// Callers should use this with an informer.
+func (c *ConfigProviderKubeAPIServer) clusterVersionOperatorEventHandler() cache.ResourceEventHandler {
+	return cache.ResourceEventHandlerFuncs{
+		AddFunc: func(_ interface{}) {
+			c.queueAdd(c.queueKey)
+			klog.V(i.Debug).Infof("ClusterVersionOperator resource was added; queuing a sync")
+		},
+		UpdateFunc: func(_, _ interface{}) {
+			c.queueAdd(c.queueKey)
+			klog.V(i.Debug).Infof("ClusterVersionOperator resource was modified or resync period has passed; queuing a sync")
+		},
+		DeleteFunc: func(_ interface{}) {
+			c.queueAdd(c.queueKey)
+			klog.V(i.Debug).Infof("ClusterVersionOperator resource was deleted; queuing a sync")
+		},
+	}
+}

pkg/cvo/cvo.go

@@ -291,7 +291,7 @@ func New(
 	optr.upgradeableChecks = optr.defaultUpgradeableChecks()
 
 	if shouldReconcileCVOConfiguration(cvoGates, optr.hypershift) {
-		optr.configuration = configuration.NewClusterVersionOperatorConfiguration(operatorClient, operatorInformerFactory)
+		optr.configuration = configuration.NewClusterVersionOperatorConfiguration().UsingKubeAPIServer(operatorClient, operatorInformerFactory)
 	}
 
 	return optr, nil