chore(controllers): Don't run controlplane specific controllers on workers

Signed-off-by: Homayoon (Hue) Alimohammadi <[email protected]>

Author: HomayoonAlimohammadi

src/k8s/pkg/k8sd/app/app.go

@@ -189,29 +189,31 @@ func New(cfg Config) (*App, error) {
 	app.controllerCoordinator = controllers.NewCoordinator(
 		cfg.Snap,
 		app.readyWg.Wait,
-		cfg.DisableUpgradeController,
-		upgrade.ControllerOptions{
-			FeatureControllerReadyCh:   app.featureController.ReadyCh(),
-			NotifyNetworkFeature:       app.NotifyNetwork,
-			NotifyGatewayFeature:       app.NotifyGateway,
-			NotifyIngressFeature:       app.NotifyIngress,
-			NotifyLoadBalancerFeature:  app.NotifyLoadBalancer,
-			NotifyLocalStorageFeature:  app.NotifyLocalStorage,
-			NotifyMetricsServerFeature: app.NotifyMetricsServer,
-			NotifyDNSFeature:           app.NotifyDNS,
-			FeatureToReconciledCh: map[types.FeatureName]<-chan struct{}{
-				features.Network:       app.featureController.ReconciledNetworkCh(),
-				features.Gateway:       app.featureController.ReconciledGatewayCh(),
-				features.Ingress:       app.featureController.ReconciledIngressCh(),
-				features.DNS:           app.featureController.ReconciledDNSCh(),
-				features.LoadBalancer:  app.featureController.ReconciledLoadBalancerCh(),
-				features.LocalStorage:  app.featureController.ReconciledLocalStorageCh(),
-				features.MetricsServer: app.featureController.ReconciledMetricsServerCh(),
+		controllers.UpgradeControllerOptions{
+			Disable: cfg.DisableUpgradeController,
+			ControllerOptions: upgrade.ControllerOptions{
+				FeatureControllerReadyCh:   app.featureController.ReadyCh(),
+				NotifyNetworkFeature:       app.NotifyNetwork,
+				NotifyGatewayFeature:       app.NotifyGateway,
+				NotifyIngressFeature:       app.NotifyIngress,
+				NotifyLoadBalancerFeature:  app.NotifyLoadBalancer,
+				NotifyLocalStorageFeature:  app.NotifyLocalStorage,
+				NotifyMetricsServerFeature: app.NotifyMetricsServer,
+				NotifyDNSFeature:           app.NotifyDNS,
+				FeatureToReconciledCh: map[types.FeatureName]<-chan struct{}{
+					features.Network:       app.featureController.ReconciledNetworkCh(),
+					features.Gateway:       app.featureController.ReconciledGatewayCh(),
+					features.Ingress:       app.featureController.ReconciledIngressCh(),
+					features.DNS:           app.featureController.ReconciledDNSCh(),
+					features.LoadBalancer:  app.featureController.ReconciledLoadBalancerCh(),
+					features.LocalStorage:  app.featureController.ReconciledLocalStorageCh(),
+					features.MetricsServer: app.featureController.ReconciledMetricsServerCh(),
+				},
 			},
-			FeatureControllerReadyTimeout:     10 * time.Minute,
-			FeatureControllerReconcileTimeout: 2 * time.Minute,
 		},
-		cfg.DisableCSRSigningController,
+		controllers.CSRSigningControllerOptions{
+			Disable: cfg.DisableCSRSigningController,
+		},
 	)
 
 	return app, nil

src/k8s/pkg/k8sd/app/hooks_start.go

@@ -124,16 +124,18 @@ func (a *App) onStart(ctx context.Context, s state.State) error {
 	}
 
 	// start controller coordinator
-	go func() {
-		if err := a.controllerCoordinator.Run(
-			ctx,
-			func(ctx context.Context) (types.ClusterConfig, error) {
-				return databaseutil.GetClusterConfig(ctx, s)
-			},
-		); err != nil {
-			log.FromContext(ctx).Error(err, "Failed to start controller coordinator")
-		}
-	}()
+	if a.controllerCoordinator != nil {
+		go func() {
+			if err := a.controllerCoordinator.Run(
+				ctx,
+				func(ctx context.Context) (types.ClusterConfig, error) {
+					return databaseutil.GetClusterConfig(ctx, s)
+				},
+			); err != nil {
+				log.FromContext(ctx).Error(err, "Failed to start controller coordinator")
+			}
+		}()
+	}
 
 	return nil
 }

src/k8s/pkg/k8sd/controllers/coordinator.go

@@ -12,6 +12,7 @@ import (
 	"github.com/canonical/k8s/pkg/k8sd/types"
 	"github.com/canonical/k8s/pkg/log"
 	"github.com/canonical/k8s/pkg/snap"
+	snaputil "github.com/canonical/k8s/pkg/snap/util"
 	"github.com/canonical/k8s/pkg/utils"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
@@ -25,28 +26,31 @@ type Coordinator struct {
 	snap      snap.Snap
 	waitReady func()
 
-	// Upgrade controller
-	disableUpgradeController bool
-	upgradeControllerOpts    upgrade.ControllerOptions
+	upgradeControllerOptions    UpgradeControllerOptions
+	csrSigningControllerOptions CSRSigningControllerOptions
+}
+
+type UpgradeControllerOptions struct {
+	upgrade.ControllerOptions
+	Disable bool
+}
 
-	// CSR signing controller
-	disableCSRSigningController bool
+type CSRSigningControllerOptions struct {
+	Disable bool
 }
 
 // NewCoordinator creates a new Coordinator instance.
 func NewCoordinator(
 	snap snap.Snap,
 	waitReady func(),
-	disableUpgradeController bool,
-	upgradeControllerOpts upgrade.ControllerOptions,
-	disableCSRSiningController bool,
+	upgradeControllerOptions UpgradeControllerOptions,
+	csrSigningControllerOptions CSRSigningControllerOptions,
 ) *Coordinator {
 	return &Coordinator{
 		snap:                        snap,
 		waitReady:                   waitReady,
-		disableUpgradeController:    disableUpgradeController,
-		upgradeControllerOpts:       upgradeControllerOpts,
-		disableCSRSigningController: disableCSRSiningController,
+		upgradeControllerOptions:    upgradeControllerOptions,
+		csrSigningControllerOptions: csrSigningControllerOptions,
 	}
 }
 
@@ -56,6 +60,16 @@ func (c *Coordinator) Run(
 	getClusterConfig func(context.Context) (types.ClusterConfig, error),
 ) error {
 	logger := log.FromContext(ctx).WithName("controller-coordinator")
+
+	isWorker, err := snaputil.IsWorker(c.snap)
+	if err != nil {
+		return fmt.Errorf("failed to determine if snap is running as worker: %w", err)
+	}
+	if isWorker {
+		logger.Info("Skipping controller coordinator on worker node")
+		return nil
+	}
+
 	ctrllog.SetLogger(logger)
 	ctx = log.NewContext(ctx, logger)
 
@@ -132,7 +146,7 @@ func (c *Coordinator) setupUpgradeController(
 ) error {
 	logger := mgr.GetLogger()
 
-	if c.disableUpgradeController {
+	if c.upgradeControllerOptions.Disable {
 		logger.Info("Upgrade controller is disabled. Skipping setup.")
 		return nil
 	}
@@ -150,7 +164,7 @@ func (c *Coordinator) setupUpgradeController(
 	upgradeController := upgrade.NewController(
 		logger,
 		mgr.GetClient(),
-		c.upgradeControllerOpts,
+		c.upgradeControllerOptions.ControllerOptions,
 	)
 
 	if err := upgradeController.SetupWithManager(mgr); err != nil {
@@ -166,7 +180,7 @@ func (c *Coordinator) setupCSRSigningController(
 ) error {
 	logger := mgr.GetLogger()
 
-	if c.disableCSRSigningController {
+	if c.csrSigningControllerOptions.Disable {
 		logger.Info("CSR signing controller is disabled. Skipping setup.")
 		return nil
 	}

src/k8s/pkg/k8sd/controllers/feature.go

@@ -12,6 +12,7 @@ import (
 	"github.com/canonical/k8s/pkg/k8sd/types"
 	"github.com/canonical/k8s/pkg/log"
 	"github.com/canonical/k8s/pkg/snap"
+	snaputil "github.com/canonical/k8s/pkg/snap/util"
 	"github.com/canonical/k8s/pkg/utils"
 	timeutils "github.com/canonical/k8s/pkg/utils/time"
 	"github.com/canonical/microcluster/v2/state"
@@ -142,9 +143,19 @@ func (c *FeatureController) Run(
 ) {
 	ctx = log.NewContext(ctx, log.FromContext(ctx).WithValues("controller", "feature"))
 	log := log.FromContext(ctx)
-	log.Info("Starting feature controller")
 	c.waitReady()
 
+	isWorker, err := snaputil.IsWorker(c.snap)
+	if err != nil {
+		log.Error(err, "Failed to determine if snap is running as worker")
+	}
+	if isWorker {
+		log.Info("Skipping feature controller on worker node")
+		return
+	}
+
+	log.Info("Starting feature controller")
+
 	s := getState()
 
 	go c.reconcileLoop(ctx, getClusterConfig, setFeatureStatus, features.Network, c.triggerNetworkCh, c.reconciledNetworkCh, func(cfg types.ClusterConfig) (types.FeatureStatus, error) {

src/k8s/pkg/k8sd/controllers/node_label.go

@@ -172,8 +172,8 @@ func (c *NodeLabelController) reconcile(ctx context.Context, client *kubernetes.
 
 	var errs []error
 
-	log.FromContext(ctx).Info("Reconciling node failure domain", "nodeName", node.Name)
 	if !isWorker {
+		log.FromContext(ctx).Info("Reconciling node failure domain", "nodeName", node.Name)
 		if err := c.reconcileFailureDomain(ctx, node, getDatastoreType); err != nil {
 			errs = append(errs, fmt.Errorf("failed to reconcile failure domain: %w", err))
 		}