feat: Add dynamic addon handler application via controller

cmd/main.go

@@ -30,6 +30,7 @@ import (
 	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/api/v1alpha1"
 	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/common/pkg/capi/clustertopology/handlers"
 	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/common/pkg/server"
+	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/pkg/controllers/addons"
 	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/pkg/controllers/namespacesync"
 	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/pkg/handlers/aws"
 	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/pkg/handlers/docker"
@@ -100,6 +101,7 @@ func main() {
 	genericMetaHandlers := generic.New()
 
 	namespacesyncOptions := namespacesync.Options{}
+	addonsOptions := addons.Options{}
 
 	// Initialize and parse command line flags.
 	logs.AddFlags(pflag.CommandLine, logs.SkipLoggingConfigurationFlags())
@@ -111,6 +113,7 @@ func main() {
 	dockerMetaHandlers.AddFlags(pflag.CommandLine)
 	nutanixMetaHandlers.AddFlags(pflag.CommandLine)
 	namespacesyncOptions.AddFlags(pflag.CommandLine)
+	addonsOptions.AddFlags(pflag.CommandLine)
 	pflag.CommandLine.SetNormalizeFunc(cliflag.WordSepNormalizeFunc)
 	pflag.CommandLine.AddGoFlagSet(flag.CommandLine)
 	pflag.Parse()
@@ -134,8 +137,9 @@ func main() {
 		os.Exit(1)
 	}
 
+	lifecycleHandlers := genericLifecycleHandlers.AllHandlers(mgr)
 	var allHandlers []handlers.Named
-	allHandlers = append(allHandlers, genericLifecycleHandlers.AllHandlers(mgr)...)
+	allHandlers = append(allHandlers, lifecycleHandlers...)
 	allHandlers = append(allHandlers, awsMetaHandlers.AllHandlers(mgr)...)
 	allHandlers = append(allHandlers, dockerMetaHandlers.AllHandlers(mgr)...)
 	allHandlers = append(allHandlers, nutanixMetaHandlers.AllHandlers(mgr)...)
@@ -174,6 +178,15 @@ func main() {
 		os.Exit(1)
 	}
 
+	if err := addons.NewController(mgr.GetClient(), lifecycleHandlers).SetupWithManager(
+		signalCtx,
+		mgr,
+		controller.Options{MaxConcurrentReconciles: namespacesyncOptions.Concurrency},
+	); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "addons.Reconciler")
+		os.Exit(1)
+	}
+
 	if err := mgr.Start(signalCtx); err != nil {
 		setupLog.Error(err, "unable to start controller manager")
 		os.Exit(1)

common/pkg/capi/clustertopology/handlers/lifecycle/interfaces.go

@@ -6,6 +6,7 @@ package lifecycle
 import (
 	"context"
 
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	runtimehooksv1 "sigs.k8s.io/cluster-api/exp/runtime/hooks/api/v1alpha1"
 )
 
@@ -44,3 +45,10 @@ type BeforeClusterDelete interface {
 		*runtimehooksv1.BeforeClusterDeleteResponse,
 	)
 }
+
+type OnClusterSpecUpdated interface {
+	OnClusterSpecUpdated(
+		ctx context.Context,
+		cluster *clusterv1.Cluster,
+	) error
+}

pkg/controllers/addons/controller.go

@@ -0,0 +1,106 @@
+// Copyright 2024 Nutanix. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package addons
+
+import (
+	"context"
+	"fmt"
+
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	"sigs.k8s.io/cluster-api/util/predicates"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+
+	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/common/pkg/capi/clustertopology/handlers"
+	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/common/pkg/capi/clustertopology/handlers/lifecycle"
+)
+
+type Reconciler struct {
+	client client.Client
+
+	handlers []lifecycle.OnClusterSpecUpdated
+}
+
+func NewController(cl client.Client, lifecycleHandlers []handlers.Named) *Reconciler {
+	specUpdatedHandlers := []lifecycle.OnClusterSpecUpdated{}
+	for _, h := range lifecycleHandlers {
+		if h, ok := h.(lifecycle.OnClusterSpecUpdated); ok {
+			specUpdatedHandlers = append(specUpdatedHandlers, h)
+		}
+	}
+	return &Reconciler{
+		client:   cl,
+		handlers: specUpdatedHandlers,
+	}
+}
+
+func (r *Reconciler) SetupWithManager(
+	ctx context.Context,
+	mgr ctrl.Manager,
+	options controller.Options,
+) error {
+	hasTopologyPredicate := predicates.ClusterHasTopology(ctrl.LoggerFrom(ctx))
+	generationChangedPredicate := predicate.GenerationChangedPredicate{}
+
+	err := ctrl.NewControllerManagedBy(mgr).
+		For(&clusterv1.Cluster{}, builder.WithPredicates(
+			predicate.Funcs{
+				CreateFunc: func(e event.CreateEvent) bool {
+					return false
+				},
+				UpdateFunc: func(e event.UpdateEvent) bool {
+					// Only reconcile Cluster with topology.
+					if !hasTopologyPredicate.UpdateFunc(e) {
+						return false
+					}
+					if !generationChangedPredicate.Update(e) {
+						return false
+					}
+					cluster, ok := e.ObjectNew.(*clusterv1.Cluster)
+					if !ok {
+						return false
+					}
+
+					return !cluster.Spec.Paused
+				},
+				DeleteFunc: func(e event.DeleteEvent) bool {
+					return false
+				},
+				GenericFunc: func(e event.GenericEvent) bool {
+					return false
+				},
+			},
+		)).
+		Named("addons").
+		WithOptions(options).
+		Complete(r)
+	if err != nil {
+		return fmt.Errorf("failed to set up with controller manager: %w", err)
+	}
+
+	return nil
+}
+
+func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	cluster := &clusterv1.Cluster{}
+	if err := r.client.Get(ctx, req.NamespacedName, cluster); err != nil {
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+
+	for _, h := range r.handlers {
+		if err := h.OnClusterSpecUpdated(ctx, cluster); err != nil {
+			return ctrl.Result{}, fmt.Errorf(
+				"failed to reconcile cluster %s: %w",
+				client.ObjectKeyFromObject(cluster),
+				err,
+			)
+		}
+	}
+
+	return ctrl.Result{}, nil
+}

pkg/controllers/addons/doc.go

@@ -0,0 +1,5 @@
+// Copyright 2024 Nutanix. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+// +kubebuilder:rbac:groups=cluster.x-k8s.io,resources=clusters,verbs=get;list;watch
+package addons

pkg/controllers/addons/flags.go

@@ -0,0 +1,21 @@
+// Copyright 2024 Nutanix. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package addons
+
+import (
+	"github.com/spf13/pflag"
+)
+
+type Options struct {
+	Concurrency int
+}
+
+func (o *Options) AddFlags(flags *pflag.FlagSet) {
+	pflag.CommandLine.IntVar(
+		&o.Concurrency,
+		"addons-clusters-concurrency",
+		10,
+		"Number of clusters to sync concurrently.",
+	)
+}

pkg/handlers/generic/lifecycle/ccm/doc.go

@@ -1,12 +1,6 @@
 // Copyright 2023 Nutanix. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
-// Package calico provides a handler for managing Calico deployments on clusters, configurable via
-// labels and annotations.
-//
-// To enable Calico deployment, a cluster must be labelled with `caren.nutanix.com/cni=calico`.
-// This will ensure the Tigera Configmap and associated ClusterResourceSet.
-//
 // +kubebuilder:rbac:groups=addons.cluster.x-k8s.io,resources=clusterresourcesets,verbs=watch;list;get;create;patch;update;delete
 // +kubebuilder:rbac:groups="",resources=configmaps,verbs=watch;list;get;create;patch;update;delete
 package ccm

pkg/handlers/generic/lifecycle/ccm/handler.go

@@ -68,28 +68,45 @@ func (c *CCMHandler) AfterControlPlaneInitialized(
 	req *runtimehooksv1.AfterControlPlaneInitializedRequest,
 	resp *runtimehooksv1.AfterControlPlaneInitializedResponse,
 ) {
-	commonResponse := &runtimehooksv1.CommonResponse{}
-	c.apply(ctx, &req.Cluster, commonResponse)
-	resp.Status = commonResponse.GetStatus()
-	resp.Message = commonResponse.GetMessage()
+	c.handle(ctx, &req.Cluster, &resp.CommonResponse)
 }
 
 func (c *CCMHandler) BeforeClusterUpgrade(
 	ctx context.Context,
 	req *runtimehooksv1.BeforeClusterUpgradeRequest,
 	resp *runtimehooksv1.BeforeClusterUpgradeResponse,
 ) {
-	commonResponse := &runtimehooksv1.CommonResponse{}
-	c.apply(ctx, &req.Cluster, commonResponse)
-	resp.Status = commonResponse.GetStatus()
-	resp.Message = commonResponse.GetMessage()
+	c.handle(ctx, &req.Cluster, &resp.CommonResponse)
 }
 
-func (c *CCMHandler) apply(
+func (c *CCMHandler) OnClusterSpecUpdated(
+	ctx context.Context,
+	cluster *clusterv1.Cluster,
+) error {
+	if err := c.apply(ctx, cluster); err != nil {
+		return fmt.Errorf("failed to apply CCM: %w", err)
+	}
+
+	return nil
+}
+
+func (c *CCMHandler) handle(
 	ctx context.Context,
 	cluster *clusterv1.Cluster,
 	resp *runtimehooksv1.CommonResponse,
 ) {
+	if err := c.apply(ctx, cluster); err != nil {
+		resp.SetStatus(runtimehooksv1.ResponseStatusFailure)
+		resp.SetMessage(err.Error())
+	}
+
+	resp.SetStatus(runtimehooksv1.ResponseStatusSuccess)
+}
+
+func (c *CCMHandler) apply(
+	ctx context.Context,
+	cluster *clusterv1.Cluster,
+) error {
 	clusterKey := ctrlclient.ObjectKeyFromObject(cluster)
 
 	log := ctrl.LoggerFrom(ctx).WithValues(
@@ -103,37 +120,17 @@ func (c *CCMHandler) apply(
 	if err != nil {
 		if variables.IsNotFoundError(err) {
 			log.V(5).Info("Skipping CCM handler.")
-			return
+			return nil
 		}
-		log.Error(
-			err,
-			"failed to read CCM from cluster definition",
-		)
-		resp.SetStatus(runtimehooksv1.ResponseStatusFailure)
-		resp.SetMessage(
-			fmt.Sprintf("failed to read CCM from cluster definition: %v",
-				err,
-			),
-		)
-		return
+		return fmt.Errorf("failed to read CCM from cluster definition: %w", err)
 	}
 
 	clusterConfigVar, err := variables.Get[apivariables.ClusterConfigSpec](
 		varMap,
 		v1alpha1.ClusterConfigVariableName,
 	)
 	if err != nil {
-		log.Error(
-			err,
-			"failed to read clusterConfig variable from cluster definition",
-		)
-		resp.SetStatus(runtimehooksv1.ResponseStatusFailure)
-		resp.SetMessage(
-			fmt.Sprintf("failed to read clusterConfig variable from cluster definition: %v",
-				err,
-			),
-		)
-		return
+		return fmt.Errorf("failed to read clusterConfig variable from cluster definition: %w", err)
 	}
 
 	// There's a 1:1 mapping of infra to CCM provider. We derive the CCM provider from the infra.
@@ -147,21 +144,13 @@ func (c *CCMHandler) apply(
 		handler = c.ProviderHandler[v1alpha1.CCMProviderNutanix]
 	default:
 		log.Info(fmt.Sprintf("No CCM handler provided for infra kind %s", infraKind))
-		return
+		return nil
 	}
 
 	err = handler.Apply(ctx, cluster, &clusterConfigVar, log)
 	if err != nil {
-		log.Error(
-			err,
-			"failed to deploy CCM for cluster",
-		)
-		resp.SetStatus(runtimehooksv1.ResponseStatusFailure)
-		resp.SetMessage(
-			fmt.Sprintf("failed to deploy CCM for cluster: %v",
-				err,
-			),
-		)
-		return
+		return fmt.Errorf("failed to deploy CCM for cluster: %w", err)
 	}
+
+	return nil
 }