MCO-2015: Use an event based queue in the boot image controller

cmd/machine-config-controller/start.go

@@ -9,13 +9,13 @@ import (
 	features "github.com/openshift/api/features"
 	"github.com/openshift/machine-config-operator/cmd/common"
 	"github.com/openshift/machine-config-operator/internal/clients"
+	bootimagecontroller "github.com/openshift/machine-config-operator/pkg/controller/bootimage"
 	certrotationcontroller "github.com/openshift/machine-config-operator/pkg/controller/certrotation"
 	ctrlcommon "github.com/openshift/machine-config-operator/pkg/controller/common"
 	containerruntimeconfig "github.com/openshift/machine-config-operator/pkg/controller/container-runtime-config"
 	"github.com/openshift/machine-config-operator/pkg/controller/drain"
 	"github.com/openshift/machine-config-operator/pkg/controller/internalreleaseimage"
 	kubeletconfig "github.com/openshift/machine-config-operator/pkg/controller/kubelet-config"
-	machinesetbootimage "github.com/openshift/machine-config-operator/pkg/controller/machine-set-boot-image"
 	"github.com/openshift/machine-config-operator/pkg/controller/node"
 	"github.com/openshift/machine-config-operator/pkg/controller/pinnedimageset"
 	"github.com/openshift/machine-config-operator/pkg/controller/render"
@@ -144,7 +144,7 @@ func runStartCmd(_ *cobra.Command, _ []string) {
 		}
 
 		if ctrlcommon.IsBootImageControllerRequired(ctrlctx) {
-			machineSetBootImage := machinesetbootimage.New(
+			bootImageController := bootimagecontroller.New(
 				ctrlctx.ClientBuilder.KubeClientOrDie("machine-set-boot-image-controller"),
 				ctrlctx.ClientBuilder.MachineClientOrDie("machine-set-boot-image-controller"),
 				ctrlctx.KubeNamespacedInformerFactory.Core().V1().ConfigMaps(),
@@ -155,7 +155,7 @@ func runStartCmd(_ *cobra.Command, _ []string) {
 				ctrlctx.OperatorInformerFactory.Operator().V1().MachineConfigurations(),
 				ctrlctx.FeatureGatesHandler,
 			)
-			go machineSetBootImage.Run(ctrlctx.Stop)
+			go bootImageController.Run(ctrlctx.Stop)
 			// start the informers again to enable feature gated types.
 			// see comments in SharedInformerFactory interface.
 			ctrlctx.KubeNamespacedInformerFactory.Start(ctrlctx.Stop)

devex/cmd/mco-sanitize/redactor.go

@@ -155,18 +155,17 @@ func recursiveWalk(node *visitorNode, path []string) (bool, error) {
 				}
 			}
 			return changed, nil
-		} else {
-			// The path provides the index to pick. Transverse only that index
-			pathIndex, err := strconv.Atoi(key)
-			if err != nil {
-				return false, errors.New("redact path uses array indexing at a path level that is not an array")
-			}
-			newNode := node.newArrayChild(pathIndex)
-			if newNode == nil {
-				return false, nil
-			}
-			return recursiveWalk(newNode, path[1:])
 		}
+		// The path provides the index to pick. Transverse only that index
+		pathIndex, err := strconv.Atoi(key)
+		if err != nil {
+			return false, errors.New("redact path uses array indexing at a path level that is not an array")
+		}
+		newNode := node.newArrayChild(pathIndex)
+		if newNode == nil {
+			return false, nil
+		}
+		return recursiveWalk(newNode, path[1:])
 	case map[string]interface{}:
 		// Map case. newObjectChild returns nil if the key doesn't exist which should make us
 		// break the transversing path

pkg/controller/machine-set-boot-image/ami.go → pkg/controller/bootimage/ami.go

@@ -1,4 +1,4 @@
-package machineset
+package bootimage
 
 import (
 	"k8s.io/apimachinery/pkg/util/sets"

pkg/controller/machine-set-boot-image/machine_set_boot_image_controller.go → pkg/controller/bootimage/boot_image_controller.go

@@ -1,10 +1,10 @@
-package machineset
+package bootimage
 
 import (
 	"context"
 	"fmt"
 	"reflect"
-	"sync"
+	"time"
 
 	features "github.com/openshift/api/features"
 	opv1 "github.com/openshift/api/operator/v1"
@@ -14,13 +14,15 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
 	coreinformersv1 "k8s.io/client-go/informers/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	corelisterv1 "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/retry"
+	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 	"k8s.io/kubectl/pkg/scheme"
 
@@ -44,6 +46,8 @@ type Controller struct {
 	mcopClient    mcopclientset.Interface
 	eventRecorder record.EventRecorder
 
+	syncHandler func(event string) error
+
 	mcoCmLister          corelisterv1.ConfigMapLister
 	mapiMachineSetLister machinelistersv1beta1.MachineSetLister
 	cpmsLister           machinelistersv1.ControlPlaneMachineSetLister
@@ -56,15 +60,14 @@ type Controller struct {
 	infraListerSynced          cache.InformerSynced
 	mcopListerSynced           cache.InformerSynced
 
+	queue workqueue.TypedRateLimitingInterface[string]
+
 	mapiStats                  MachineResourceStats
 	cpmsStats                  MachineResourceStats
 	capiMachineSetStats        MachineResourceStats
 	capiMachineDeploymentStats MachineResourceStats
 	mapiBootImageState         map[string]BootImageState
 	cpmsBootImageState         map[string]BootImageState
-	conditionMutex             sync.Mutex
-	mapiSyncMutex              sync.Mutex
-	cpmsSyncMutex              sync.Mutex
 
 	fgHandler ctrlcommon.FeatureGatesHandler
 }
@@ -78,6 +81,7 @@ type MachineResourceStats struct {
 
 // State structure uses for detecting hot loops. Reset when cluster is opted
 // out of boot image updates.
+// nolint: revive
 type BootImageState struct {
 	value        []byte
 	hotLoopCount int
@@ -103,6 +107,9 @@ const (
 
 	// Threshold for hot loop detection
 	HotLoopLimit = 3
+
+	// maxRetries is the number of times a sync will be retried before it is dropped out of the queue.
+	maxRetries = 15
 )
 
 // New returns a new machine-set-boot-image controller.
@@ -126,8 +133,13 @@ func New(
 		machineClient: machineClient,
 		mcopClient:    mcopClient,
 		eventRecorder: ctrlcommon.NamespacedEventRecorder(eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: "machineconfigcontroller-machinesetbootimagecontroller"})),
+		queue: workqueue.NewTypedRateLimitingQueueWithConfig(
+			workqueue.DefaultTypedControllerRateLimiter[string](),
+			workqueue.TypedRateLimitingQueueConfig[string]{Name: "machineconfigcontroller-machinesetbootimagecontroller"}),
 	}
 
+	ctrl.syncHandler = ctrl.syncAll
+
 	ctrl.mcoCmLister = mcoCmInfomer.Lister()
 	ctrl.mapiMachineSetLister = mapiMachineSetInformer.Lister()
 	ctrl.cpmsLister = cpmsInformer.Lister()
@@ -178,6 +190,7 @@ func New(
 // Run executes the machine-set-boot-image controller.
 func (ctrl *Controller) Run(stopCh <-chan struct{}) {
 	defer utilruntime.HandleCrash()
+	defer ctrl.queue.ShutDown()
 
 	if !cache.WaitForCacheSync(stopCh, ctrl.mcoCmListerSynced, ctrl.mapiMachineSetListerSynced, ctrl.infraListerSynced, ctrl.mcopListerSynced) {
 		return
@@ -186,9 +199,58 @@ func (ctrl *Controller) Run(stopCh <-chan struct{}) {
 	klog.Info("Starting MachineConfigController-MachineSetBootImageController")
 	defer klog.Info("Shutting down MachineConfigController-MachineSetBootImageController")
 
+	// This controller needs to run in single thread mode, as the work unit per sync are
+	// the same and shouldn't overlap each other.
+	go wait.Until(ctrl.worker, time.Second, stopCh)
+
 	<-stopCh
 }
 
+// enqueueEvent adds a event to the work queue.
+func (ctrl *Controller) enqueueEvent(event string) {
+	ctrl.queue.Add(event)
+}
+
+// worker runs a worker thread that just dequeues items, processes them, and marks them done.
+// It enforces that the syncHandler is never invoked concurrently with the same key.
+func (ctrl *Controller) worker() {
+	for ctrl.processNextWorkItem() {
+	}
+}
+
+// processNextWorkItem processes the next work item from the queue.
+func (ctrl *Controller) processNextWorkItem() bool {
+	event, quit := ctrl.queue.Get()
+	if quit {
+		return false
+	}
+	defer ctrl.queue.Done(event)
+
+	err := ctrl.syncHandler(event)
+	ctrl.handleErr(err, event)
+
+	return true
+}
+
+// handleErr checks if an error happened and makes sure we will retry later.
+func (ctrl *Controller) handleErr(err error, event string) {
+	if err == nil {
+		ctrl.queue.Forget(event)
+		return
+	}
+
+	if ctrl.queue.NumRequeues(event) < maxRetries {
+		klog.V(2).Infof("Error syncing boot image controller for event %v: %v", event, err)
+		ctrl.queue.AddRateLimited(event)
+		return
+	}
+
+	utilruntime.HandleError(err)
+	klog.V(2).Infof("Dropping event %q out of the queue: %v", event, err)
+	ctrl.queue.Forget(event)
+	ctrl.queue.AddAfter(event, 1*time.Minute)
+}
+
 // addMAPIMachineSet handles the addition of a MAPI MachineSet by triggering
 // a reconciliation of all enrolled MAPI MachineSets.
 func (ctrl *Controller) addMAPIMachineSet(obj interface{}) {
@@ -200,7 +262,7 @@ func (ctrl *Controller) addMAPIMachineSet(obj interface{}) {
 	// Update/Check all machinesets instead of just this one. This prevents needing to maintain a local
 	// store of machineset conditions. As this is using a lister, it is relatively inexpensive to do
 	// this.
-	go func() { ctrl.syncMAPIMachineSets("MAPIMachinesetAdded") }()
+	ctrl.enqueueEvent("MAPIMachineSetAdded")
 }
 
 // updateMAPIMachineSet handles updates to a MAPI MachineSet by triggering
@@ -223,7 +285,7 @@ func (ctrl *Controller) updateMAPIMachineSet(oldMS, newMS interface{}) {
 	// Update all machinesets instead of just this one. This prevents needing to maintain a local
 	// store of machineset conditions. As this is using a lister, it is relatively inexpensive to do
 	// this.
-	go func() { ctrl.syncMAPIMachineSets("MAPIMachinesetUpdated") }()
+	ctrl.enqueueEvent("MAPIMachineSetUpdated")
 }
 
 // deleteMAPIMachineSet handles the deletion of a MAPI MachineSet by triggering
@@ -237,7 +299,7 @@ func (ctrl *Controller) deleteMAPIMachineSet(deletedMS interface{}) {
 	// Update all machinesets. This prevents needing to maintain a local
 	// store of machineset conditions. As this is using a lister, it is relatively inexpensive to do
 	// this.
-	go func() { ctrl.syncMAPIMachineSets("MAPIMachinesetDeleted") }()
+	ctrl.enqueueEvent("MAPIMachineSetDeleted")
 }
 
 // addControlPlaneMachineSet handles the addition of a ControlPlaneMachineSet by triggering
@@ -251,7 +313,7 @@ func (ctrl *Controller) addControlPlaneMachineSet(obj interface{}) {
 	// Update/Check all ControlPlaneMachineSets instead of just this one. This prevents needing to maintain a local
 	// store of machineset conditions. As this is using a lister, it is relatively inexpensive to do
 	// this.
-	go func() { ctrl.syncControlPlaneMachineSets("ControlPlaneMachineSetAdded") }()
+	ctrl.enqueueEvent("ControlPlaneMachineSetAdded")
 }
 
 // updateControlPlaneMachineSet handles updates to a ControlPlaneMachineSet by triggering
@@ -274,21 +336,21 @@ func (ctrl *Controller) updateControlPlaneMachineSet(oldCPMS, newCPMS interface{
 	// Update all ControlPlaneMachineSets instead of just this one. This prevents needing to maintain a local
 	// store of machineset conditions. As this is using a lister, it is relatively inexpensive to do
 	// this.
-	go func() { ctrl.syncControlPlaneMachineSets("ControlPlaneMachineSetUpdated") }()
+	ctrl.enqueueEvent("ControlPlaneMachineSetUpdated")
 }
 
 // deleteControlPlaneMachineSet handles the deletion of a ControlPlaneMachineSet by triggering
 // a reconciliation of all enrolled ControlPlaneMachineSets.
 func (ctrl *Controller) deleteControlPlaneMachineSet(deletedCPMS interface{}) {
 
-	deletedMachineSet := deletedCPMS.(*machinev1beta1.MachineSet)
+	deletedMachineSet := deletedCPMS.(*machinev1.ControlPlaneMachineSet)
 
 	klog.Infof("ControlPlaneMachineSet %s deleted, reconciling enrolled machineset resources", deletedMachineSet.Name)
 
 	// Update all ControlPlaneMachineSets. This prevents needing to maintain a local
 	// store of machineset conditions. As this is using a lister, it is relatively inexpensive to do
 	// this.
-	go func() { ctrl.syncControlPlaneMachineSets("ControlPlaneMachineSetDeleted") }()
+	ctrl.enqueueEvent("ControlPlaneMachineSetDeleted")
 }
 
 // addConfigMap handles the addition of the boot images ConfigMap by triggering
@@ -305,7 +367,7 @@ func (ctrl *Controller) addConfigMap(obj interface{}) {
 	klog.Infof("configMap %s added, reconciling enrolled machine resources", configMap.Name)
 
 	// Update all machinesets since the "golden" configmap has been added
-	go func() { ctrl.syncAll("BootImageConfigMapAdded") }()
+	ctrl.enqueueEvent("BootImageConfigMapAdded")
 }
 
 // updateConfigMap handles updates to the boot images ConfigMap by triggering
@@ -328,7 +390,7 @@ func (ctrl *Controller) updateConfigMap(oldCM, newCM interface{}) {
 	klog.Infof("configMap %s updated, reconciling enrolled machine resources", oldConfigMap.Name)
 
 	// Update all machinesets since the "golden" configmap has been updated
-	go func() { ctrl.syncAll("BootImageConfigMapUpdated") }()
+	ctrl.enqueueEvent("BootImageConfigMapUpdated")
 }
 
 // deleteConfigMap handles the deletion of the boot images ConfigMap by triggering
@@ -345,7 +407,7 @@ func (ctrl *Controller) deleteConfigMap(obj interface{}) {
 	klog.Infof("configMap %s deleted, reconciling enrolled machine resources", configMap.Name)
 
 	// Update all machinesets since the "golden" configmap has been deleted
-	go func() { ctrl.syncAll("BootImageConfigMapDeleted") }()
+	ctrl.enqueueEvent("BootImageConfigMapDeleted")
 }
 
 // addMachineConfiguration handles the addition of the cluster-level MachineConfiguration
@@ -363,7 +425,7 @@ func (ctrl *Controller) addMachineConfiguration(obj interface{}) {
 	klog.Infof("Bootimages management configuration has been added, reconciling enrolled machine resources")
 
 	// Update/Check machinesets since the boot images configuration knob was updated
-	go func() { ctrl.syncAll("BootImageUpdateConfigurationAdded") }()
+	ctrl.enqueueEvent("BootImageUpdateConfigurationAdded")
 }
 
 // updateMachineConfiguration handles updates to the cluster-level MachineConfiguration
@@ -387,7 +449,7 @@ func (ctrl *Controller) updateMachineConfiguration(oldMC, newMC interface{}) {
 	klog.Infof("Bootimages management configuration has been updated, reconciling enrolled machine resources")
 
 	// Update all machinesets since the boot images configuration knob was updated
-	go func() { ctrl.syncAll("BootImageUpdateConfigurationUpdated") }()
+	ctrl.enqueueEvent("BootImageUpdateConfigurationUpdated")
 }
 
 // deleteMachineConfiguration handles the deletion of the cluster-level MachineConfiguration
@@ -405,14 +467,13 @@ func (ctrl *Controller) deleteMachineConfiguration(obj interface{}) {
 	klog.Infof("Bootimages management configuration has been deleted, reconciling enrolled machine resources")
 
 	// Update/Check machinesets since the boot images configuration knob was updated
-	go func() { ctrl.syncAll("BootImageUpdateConfigurationDeleted") }()
+	ctrl.enqueueEvent("BootImageUpdateConfigurationDeleted")
 }
 
 // updateConditions updates the boot image update conditions on the MachineConfiguration status
 // based on the current state of machine resource reconciliation.
 func (ctrl *Controller) updateConditions(newReason string, syncError error, targetConditionType string) {
-	ctrl.conditionMutex.Lock()
-	defer ctrl.conditionMutex.Unlock()
+
 	mcop, err := ctrl.mcopClient.OperatorV1().MachineConfigurations().Get(context.TODO(), ctrlcommon.MCOOperatorKnobsObjectName, metav1.GetOptions{})
 	if err != nil {
 		klog.Errorf("error updating progressing condition: %s", err)
@@ -508,8 +569,17 @@ func getDefaultConditions() []metav1.Condition {
 
 }
 
-// syncAll will attempt to enqueue all supported machine resources
-func (ctrl *Controller) syncAll(reason string) {
-	ctrl.syncControlPlaneMachineSets(reason)
-	ctrl.syncMAPIMachineSets(reason)
+// syncAll will attempt to sync all supported machine resources
+func (ctrl *Controller) syncAll(event string) error {
+	klog.V(4).Infof("Syncing boot image controller for event: %s", event)
+
+	// Wait for MachineConfiguration/cluster to be ready before syncing any machine resources
+	if err := ctrl.waitForMachineConfigurationReady(); err != nil {
+		ctrl.updateConditions(event, fmt.Errorf("MachineConfiguration was not ready: %v", err), opv1.MachineConfigurationBootImageUpdateDegraded)
+		return err
+	}
+
+	ctrl.syncControlPlaneMachineSets(event)
+	ctrl.syncMAPIMachineSets(event)
+	return nil
 }

pkg/controller/machine-set-boot-image/machine_set_boot_image_controller_test.go → pkg/controller/bootimage/boot_image_controller_test.go

@@ -1,4 +1,4 @@
-package machineset
+package bootimage
 
 import (
 	"testing"