feat(multicluster): implement multicluster lifecycle management and reconcile logic

Author: nexus49

controller/lifecycle/controllerruntime/lifecycle.go

@@ -64,27 +64,11 @@ func (l *LifecycleManager) Spreader() *spread.Spreader {
 }
 
 func (l *LifecycleManager) Reconcile(ctx context.Context, req ctrl.Request, instance runtimeobject.RuntimeObject) (ctrl.Result, error) {
-	return lifecycle.Reconcile(ctx, req, instance, l.client, l)
-}
-
-func (l *LifecycleManager) validateInterfaces(instance runtimeobject.RuntimeObject, log *logger.Logger) error {
-	if l.Spreader() != nil {
-		_, err := l.Spreader().ToRuntimeObjectSpreadReconcileStatusInterface(instance, log)
-		if err != nil {
-			return err
-		}
-	}
-	if l.ConditionsManager() != nil {
-		_, err := l.ConditionsManager().ToRuntimeObjectConditionsInterface(instance, log)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
+	return lifecycle.Reconcile(ctx, req.NamespacedName, instance, l.client, l)
 }
 
 func (l *LifecycleManager) SetupWithManagerBuilder(mgr ctrl.Manager, maxReconciles int, reconcilerName string, instance runtimeobject.RuntimeObject, debugLabelValue string, log *logger.Logger, eventPredicates ...predicate.Predicate) (*builder.Builder, error) {
-	if err := l.validateInterfaces(instance, log); err != nil {
+	if err := lifecycle.ValidateInterfaces(instance, log, l); err != nil {
 		return nil, err
 	}
 

controller/lifecycle/lifecycle.go

@@ -13,6 +13,7 @@ import (
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -43,21 +44,23 @@ type Config struct {
 
 type PrepareContextFunc func(ctx context.Context, instance runtimeobject.RuntimeObject) (context.Context, errors.OperatorError)
 
-func Reconcile(ctx context.Context, req ctrl.Request, instance runtimeobject.RuntimeObject, cl client.Client, l Lifecycle) (ctrl.Result, error) {
+func Reconcile(ctx context.Context, nName types.NamespacedName, instance runtimeobject.RuntimeObject, cl client.Client, l Lifecycle) (ctrl.Result, error) {
 	ctx, span := otel.Tracer(l.Config().OperatorName).Start(ctx, fmt.Sprintf("%s.Reconcile", l.Config().ControllerName))
 	defer span.End()
 
 	result := ctrl.Result{}
 	reconcileId := uuid.New().String()
 
-	log := l.Log().MustChildLoggerWithAttributes("name", req.Name, "namespace", req.Namespace, "reconcile_id", reconcileId)
-	sentryTags := sentry.Tags{"namespace": req.Namespace, "name": req.Name}
+	log := l.Log().MustChildLoggerWithAttributes("name", nName.Name, "namespace", nName.Namespace, "reconcile_id", reconcileId)
+	sentryTags := sentry.Tags{"namespace": nName.Namespace, "name": nName.Name}
 
 	ctx = logger.SetLoggerInContext(ctx, log)
 	ctx = sentry.ContextWithSentryTags(ctx, sentryTags)
 
 	log.Info().Msg("start reconcile")
-	err := cl.Get(ctx, req.NamespacedName, instance)
+
+	nn := types.NamespacedName{Namespace: nName.Namespace, Name: nName.Name}
+	err := cl.Get(ctx, nn, instance)
 	if err != nil {
 		if kerrors.IsNotFound(err) {
 			log.Info().Msg("instance not found. It was likely deleted")
@@ -380,3 +383,19 @@ func HandleOperatorError(ctx context.Context, operatorError errors.OperatorError
 
 	return ctrl.Result{}, nil
 }
+
+func ValidateInterfaces(instance runtimeobject.RuntimeObject, log *logger.Logger, l Lifecycle) error {
+	if l.Spreader() != nil {
+		_, err := l.Spreader().ToRuntimeObjectSpreadReconcileStatusInterface(instance, log)
+		if err != nil {
+			return err
+		}
+	}
+	if l.ConditionsManager() != nil {
+		_, err := l.ConditionsManager().ToRuntimeObjectConditionsInterface(instance, log)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

controller/lifecycle/multicluster/lifecycle.go

@@ -0,0 +1,130 @@
+package multicluster
+
+import (
+	"context"
+	"fmt"
+
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder"
+	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
+	mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile"
+
+	"github.com/platform-mesh/golang-commons/controller/filter"
+	"github.com/platform-mesh/golang-commons/controller/lifecycle"
+	"github.com/platform-mesh/golang-commons/controller/lifecycle/conditions"
+	"github.com/platform-mesh/golang-commons/controller/lifecycle/runtimeobject"
+	"github.com/platform-mesh/golang-commons/controller/lifecycle/spread"
+	"github.com/platform-mesh/golang-commons/controller/lifecycle/subroutine"
+	"github.com/platform-mesh/golang-commons/logger"
+)
+
+type ClusterGetter interface {
+	GetCluster(ctx context.Context, clusterName string) (cluster.Cluster, error)
+}
+
+type LifecycleManager struct {
+	log                *logger.Logger
+	mgr                ClusterGetter
+	config             lifecycle.Config
+	subroutines        []subroutine.Subroutine
+	spreader           *spread.Spreader
+	conditionsManager  *conditions.ConditionManager
+	prepareContextFunc lifecycle.PrepareContextFunc
+}
+
+func NewLifecycleManager(log *logger.Logger, operatorName string, controllerName string, mgr ClusterGetter, subroutines []subroutine.Subroutine) *LifecycleManager {
+	log = log.MustChildLoggerWithAttributes("operator", operatorName, "controller", controllerName)
+	return &LifecycleManager{
+		log:         log,
+		mgr:         mgr,
+		subroutines: subroutines,
+		config: lifecycle.Config{
+			OperatorName:   operatorName,
+			ControllerName: controllerName,
+		},
+	}
+}
+
+func (l *LifecycleManager) Config() lifecycle.Config {
+	return l.config
+}
+func (l *LifecycleManager) Log() *logger.Logger {
+	return l.log
+}
+func (l *LifecycleManager) Subroutines() []subroutine.Subroutine {
+	return l.subroutines
+}
+func (l *LifecycleManager) PrepareContextFunc() lifecycle.PrepareContextFunc {
+	return l.prepareContextFunc
+}
+func (l *LifecycleManager) ConditionsManager() *conditions.ConditionManager {
+	return l.conditionsManager
+}
+func (l *LifecycleManager) Spreader() *spread.Spreader {
+	return l.spreader
+}
+func (l *LifecycleManager) Reconcile(ctx context.Context, req mcreconcile.Request, instance runtimeobject.RuntimeObject) (ctrl.Result, error) {
+	cl, err := l.mgr.GetCluster(ctx, req.ClusterName)
+	if err != nil {
+		return reconcile.Result{}, fmt.Errorf("failed to get cluster: %w", err)
+	}
+	client := cl.GetClient()
+	return lifecycle.Reconcile(ctx, req.NamespacedName, instance, client, l)
+}
+func (l *LifecycleManager) SetupWithManagerBuilder(mgr mcmanager.Manager, maxReconciles int, reconcilerName string, instance runtimeobject.RuntimeObject, debugLabelValue string, log *logger.Logger, eventPredicates ...predicate.Predicate) (*mcbuilder.Builder, error) {
+	if err := lifecycle.ValidateInterfaces(instance, log, l); err != nil {
+		return nil, err
+	}
+
+	if (l.ConditionsManager() != nil || l.Spreader() != nil) && l.Config().ReadOnly {
+		return nil, fmt.Errorf("cannot use conditions or spread reconciles in read-only mode")
+	}
+
+	eventPredicates = append([]predicate.Predicate{filter.DebugResourcesBehaviourPredicate(debugLabelValue)}, eventPredicates...)
+	opts := controller.TypedOptions[mcreconcile.Request]{
+		MaxConcurrentReconciles: maxReconciles,
+	}
+	return mcbuilder.ControllerManagedBy(mgr).
+		Named(reconcilerName).
+		For(instance).
+		WithOptions(opts).
+		WithEventFilter(predicate.And(eventPredicates...)), nil
+}
+func (l *LifecycleManager) SetupWithManager(mgr mcmanager.Manager, maxReconciles int, reconcilerName string, instance runtimeobject.RuntimeObject, debugLabelValue string, r mcreconcile.Reconciler, log *logger.Logger, eventPredicates ...predicate.Predicate) error {
+	b, err := l.SetupWithManagerBuilder(mgr, maxReconciles, reconcilerName, instance, debugLabelValue, log, eventPredicates...)
+	if err != nil {
+		return err
+	}
+
+	return b.Complete(r)
+}
+
+// WithPrepareContextFunc allows to set a function that prepares the context before each reconciliation
+// This can be used to add additional information to the context that is needed by the subroutines
+// You need to return a new context and an OperatorError in case of an error
+func (l *LifecycleManager) WithPrepareContextFunc(prepareFunction lifecycle.PrepareContextFunc) *LifecycleManager {
+	l.prepareContextFunc = prepareFunction
+	return l
+}
+
+// WithReadOnly allows to set the controller to read-only mode
+// In read-only mode, the controller will not update the status of the instance
+func (l *LifecycleManager) WithReadOnly() *LifecycleManager {
+	l.config.ReadOnly = true
+	return l
+}
+
+// WithSpreadingReconciles sets the LifecycleManager to spread out the reconciles
+func (l *LifecycleManager) WithSpreadingReconciles() *LifecycleManager {
+	l.spreader = spread.NewSpreader()
+	return l
+}
+
+func (l *LifecycleManager) WithConditionManagement() *LifecycleManager {
+	l.conditionsManager = conditions.NewConditionManager()
+	return l
+}