Repository: kubernetes/kubernetes
File: staging/src/k8s.io/cloud-provider/controllers/route/route_controller.go (Go)

Code before:
	if rc.broadcaster != nil {
		rc.broadcaster.StartStructuredLogging(0)
		rc.broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: rc.kubeClient.CoreV1().Events("")})
		defer rc.broadcaster.Shutdown()
	}

	klog.Info("Starting route controller")
	defer klog.Info("Shutting down route controller")
	controllerManagerMetrics.ControllerStarted("route")
	defer controllerManagerMetrics.ControllerStopped("route")

	if !cache.WaitForNamedCacheSync("route", ctx.Done(), rc.nodeListerSynced) {
		return
	}

	if utilfeature.DefaultFeatureGate.Enabled(features.CloudControllerManagerWatchBasedRoutesReconciliation) {
		// Process any node events that may change the routes.
		// It does not make sense to add concurrency here, as the routes
		// controller always processes the whole cluster.
		go wait.UntilWithContext(ctx, rc.runWorker, time.Second)

		// We should still regularly run the reconcile, even if no events come in. Usually controllers use the
		// `syncPeriod` for this, but for the route controller the default period is very low (5s) and not useful for
		// watch-based reconciliation.
		// Because of our event filters the usual resync period from the informer does not trigger reconciles.
		// TODO: What time should we use?
	} else {
		go wait.NonSlidingUntil(func() {
			if err := rc.reconcileNodeRoutes(ctx); err != nil {
				klog.Errorf("Couldn't reconcile node routes: %v", err)
			}
		}, syncPeriod, ctx.Done())
	}

	<-ctx.Done()
}

func (rc *RouteController) runWorker(ctx context.Context) {
	for rc.processNextWorkItem(ctx) {
	}
}

// processNextWorkItem will read a single work item off the workqueue and
// attempt to process it, by calling reconcileNodeRoutes.
func (rc *RouteController) processNextWorkItem(ctx context.Context) bool {
	obj, shutdown := rc.workqueue.Get()
	if shutdown {
		return false
	}

	// We wrap this block in a func so we can defer rc.workqueue.Done.

Review comment (suggestion):
if we do not filter on the event handlers, is not possible to use the resync period then? doing watch based and periodic at the same time is racy and we are worse than before

Code after:
func (rc *RouteController) Run(ctx context.Context, syncPeriod time.Duration, controllerManagerMetrics *controllersmetrics.ControllerManagerMetrics) {
	defer utilruntime.HandleCrash()

	if utilfeature.DefaultFeatureGate.Enabled(features.CloudControllerManagerWatchBasedRoutesReconciliation) && rc.workqueue != nil {
		defer rc.workqueue.ShutDown()
	}

	rc.broadcaster = record.NewBroadcaster(record.WithContext(ctx))
	rc.recorder = rc.broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "route_controller"})

	// Start event processing pipeline.
	if rc.broadcaster != nil {
		rc.broadcaster.StartStructuredLogging(0)
		rc.broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: rc.kubeClient.CoreV1().Events("")})
		defer rc.broadcaster.Shutdown()
	}

	klog.Info("Starting route controller")
	defer klog.Info("Shutting down route controller")
	controllerManagerMetrics.ControllerStarted("route")
	defer controllerManagerMetrics.ControllerStopped("route")

	if !cache.WaitForNamedCacheSyncWithContext(ctx, rc.nodeListerSynced) {
		return
	}

	if utilfeature.DefaultFeatureGate.Enabled(features.CloudControllerManagerWatchBasedRoutesReconciliation) {
		// Process any node events that may change the routes.
		// It does not make sense to add concurrency here, as the routes
		// controller always processes the whole cluster.
		go wait.UntilWithContext(ctx, rc.runWorker, time.Second)
	} else {
		go wait.NonSlidingUntil(func() {
			if err := rc.reconcileNodeRoutes(ctx); err != nil {
				klog.Errorf("Couldn't reconcile node routes: %v", err)
			}
		}, syncPeriod, ctx.Done())
	}

	<-ctx.Done()
}

func (rc *RouteController) runWorker(ctx context.Context) {
	for rc.processNextWorkItem(ctx) {
	}
}

// processNextWorkItem will read a single work item off the workqueue and
// attempt to process it, by calling reconcileNodeRoutes.
func (rc *RouteController) processNextWorkItem(ctx context.Context) bool {
	obj, shutdown := rc.workqueue.Get()