Skip to content

🌱 Refactor ClusterExtension reconciler to use composable step-based pipeline #2332

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

🌱 Refactor ClusterExtension reconciler to use composable step-based pipeline #2332

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fa112fe
Refactor `ClusterExtension` reconciler to use composable step-based p…
pedjak Nov 5, 2025
ea86263
Removed all fields but `Client` from `ClusterExtensionReconciler`
pedjak Nov 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
165 changes: 103 additions & 62 deletions cmd/operator-controller/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,6 @@ import (
crcache "sigs.k8s.io/controller-runtime/pkg/cache"
"sigs.k8s.io/controller-runtime/pkg/certwatcher"
"sigs.k8s.io/controller-runtime/pkg/client"
crcontroller "sigs.k8s.io/controller-runtime/pkg/controller"
crfinalizer "sigs.k8s.io/controller-runtime/pkg/finalizer"
"sigs.k8s.io/controller-runtime/pkg/healthz"
"sigs.k8s.io/controller-runtime/pkg/log"
Expand All @@ -68,6 +67,7 @@ import (
"github.com/operator-framework/operator-controller/internal/operator-controller/catalogmetadata/cache"
catalogclient "github.com/operator-framework/operator-controller/internal/operator-controller/catalogmetadata/client"
"github.com/operator-framework/operator-controller/internal/operator-controller/contentmanager"
cmcache "github.com/operator-framework/operator-controller/internal/operator-controller/contentmanager/cache"
"github.com/operator-framework/operator-controller/internal/operator-controller/controllers"
"github.com/operator-framework/operator-controller/internal/operator-controller/features"
"github.com/operator-framework/operator-controller/internal/operator-controller/finalizers"
Expand Down Expand Up @@ -108,6 +108,31 @@ type config struct {
globalPullSecret string
}

type clusterExtensionReconcilerConfigurator interface {
Configure(cer *controllers.ClusterExtensionReconciler) error
}

type boxcutterCERConfigurator struct {
Copy link
Member

@joelanford joelanford Nov 21, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Naming nit: my first thought here was that CER stands for ClusterExtensionRevision, but it is actually ClusterExtension reconciler? Can we disambiguate? Same for helmCERConfigurator.

mgr manager.Manager
preflights []applier.Preflight
regv1ManifestProvider applier.ManifestProvider
resolver resolve.Resolver
imageCache imageutil.Cache
imagePuller imageutil.Puller
finalizers crfinalizer.Finalizers
}

type helmCERConfigurator struct {
mgr manager.Manager
preflights []applier.Preflight
regv1ManifestProvider applier.ManifestProvider
resolver resolve.Resolver
imageCache imageutil.Cache
imagePuller imageutil.Puller
finalizers crfinalizer.Finalizers
watcher cmcache.Watcher
}

const (
authFilePrefix = "operator-controller-global-pull-secrets"
fieldOwnerPrefix = "olm.operatorframework.io"
Expand Down Expand Up @@ -440,11 +465,7 @@ func run() error {
}

ceReconciler := &controllers.ClusterExtensionReconciler{
Client: cl,
Resolver: resolver,
ImageCache: imageCache,
ImagePuller: imagePuller,
Finalizers: clusterExtensionFinalizers,
Client: cl,
}
ceController, err := ceReconciler.SetupWithManager(mgr, ctrlBuilderOpts...)
if err != nil {
Expand All @@ -459,13 +480,30 @@ func run() error {
IsWebhookSupportEnabled: certProvider != nil,
IsSingleOwnNamespaceEnabled: features.OperatorControllerFeatureGate.Enabled(features.SingleOwnNamespaceInstallSupport),
}

var cerCfg clusterExtensionReconcilerConfigurator
if features.OperatorControllerFeatureGate.Enabled(features.BoxcutterRuntime) {
err = setupBoxcutter(mgr, ceReconciler, preflights, clusterExtensionFinalizers, regv1ManifestProvider)
cerCfg = &boxcutterCERConfigurator{
mgr: mgr,
preflights: preflights,
regv1ManifestProvider: regv1ManifestProvider,
resolver: resolver,
imageCache: imageCache,
imagePuller: imagePuller,
finalizers: clusterExtensionFinalizers,
}
} else {
err = setupHelm(mgr, ceReconciler, preflights, ceController, clusterExtensionFinalizers, regv1ManifestProvider)
cerCfg = &helmCERConfigurator{
mgr: mgr,
preflights: preflights,
regv1ManifestProvider: regv1ManifestProvider,
resolver: resolver,
imageCache: imageCache,
imagePuller: imagePuller,
finalizers: clusterExtensionFinalizers,
watcher: ceController,
}
}
if err != nil {
if err := cerCfg.Configure(ceReconciler); err != nil {
setupLog.Error(err, "unable to setup lifecycler")
return err
}
Expand Down Expand Up @@ -524,19 +562,13 @@ func getCertificateProvider() render.CertificateProvider {
return nil
}

func setupBoxcutter(
mgr manager.Manager,
ceReconciler *controllers.ClusterExtensionReconciler,
preflights []applier.Preflight,
clusterExtensionFinalizers crfinalizer.Registerer,
regv1ManifestProvider applier.ManifestProvider,
) error {
coreClient, err := corev1client.NewForConfig(mgr.GetConfig())
func (c *boxcutterCERConfigurator) Configure(ceReconciler *controllers.ClusterExtensionReconciler) error {
coreClient, err := corev1client.NewForConfig(c.mgr.GetConfig())
if err != nil {
return fmt.Errorf("unable to create core client: %w", err)
}
cfgGetter, err := helmclient.NewActionConfigGetter(mgr.GetConfig(), mgr.GetRESTMapper(),
helmclient.StorageDriverMapper(action.ChunkedStorageDriverMapper(coreClient, mgr.GetAPIReader(), cfg.systemNamespace)),
cfgGetter, err := helmclient.NewActionConfigGetter(c.mgr.GetConfig(), c.mgr.GetRESTMapper(),
helmclient.StorageDriverMapper(action.ChunkedStorageDriverMapper(coreClient, c.mgr.GetAPIReader(), cfg.systemNamespace)),
helmclient.ClientNamespaceMapper(func(obj client.Object) (string, error) {
ext := obj.(*ocv1.ClusterExtension)
return ext.Spec.Namespace, nil
Expand All @@ -557,7 +589,7 @@ func setupBoxcutter(
// This finalizer was added by the Helm applier for ClusterExtensions created
// before BoxcutterRuntime was enabled. Boxcutter doesn't use contentmanager,
// so we just need to acknowledge the finalizer to allow deletion to proceed.
err = clusterExtensionFinalizers.Register(controllers.ClusterExtensionCleanupContentManagerCacheFinalizer, finalizers.FinalizerFunc(func(ctx context.Context, obj client.Object) (crfinalizer.Result, error) {
err = c.finalizers.Register(controllers.ClusterExtensionCleanupContentManagerCacheFinalizer, finalizers.FinalizerFunc(func(ctx context.Context, obj client.Object) (crfinalizer.Result, error) {
// No-op: Boxcutter doesn't use contentmanager, so no cleanup is needed
return crfinalizer.Result{}, nil
}))
Expand All @@ -568,27 +600,35 @@ func setupBoxcutter(

// TODO: add support for preflight checks
// TODO: better scheme handling - which types do we want to support?
_ = apiextensionsv1.AddToScheme(mgr.GetScheme())
_ = apiextensionsv1.AddToScheme(c.mgr.GetScheme())
rg := &applier.SimpleRevisionGenerator{
Scheme: mgr.GetScheme(),
ManifestProvider: regv1ManifestProvider,
Scheme: c.mgr.GetScheme(),
ManifestProvider: c.regv1ManifestProvider,
}
ceReconciler.Applier = &applier.Boxcutter{
Client: mgr.GetClient(),
Scheme: mgr.GetScheme(),
appl := &applier.Boxcutter{
Client: c.mgr.GetClient(),
Scheme: c.mgr.GetScheme(),
RevisionGenerator: rg,
Preflights: preflights,
Preflights: c.preflights,
FieldOwner: fmt.Sprintf("%s/clusterextension-controller", fieldOwnerPrefix),
}
ceReconciler.RevisionStatesGetter = &controllers.BoxcutterRevisionStatesGetter{Reader: mgr.GetClient()}
ceReconciler.StorageMigrator = &applier.BoxcutterStorageMigrator{
Client: mgr.GetClient(),
Scheme: mgr.GetScheme(),
revisionStatesGetter := &controllers.BoxcutterRevisionStatesGetter{Reader: c.mgr.GetClient()}
storageMigrator := &applier.BoxcutterStorageMigrator{
Client: c.mgr.GetClient(),
Scheme: c.mgr.GetScheme(),
ActionClientGetter: acg,
RevisionGenerator: rg,
}
ceReconciler.ReconcileSteps = []controllers.ReconcileStepFunc{
controllers.HandleFinalizers(c.finalizers),
controllers.MigrateStorage(storageMigrator),
controllers.RetrieveRevisionStates(revisionStatesGetter),
controllers.RetrieveRevisionMetadata(c.resolver),
controllers.UnpackBundle(c.imagePuller, c.imageCache),
controllers.ApplyBundle(appl),
}

baseDiscoveryClient, err := discovery.NewDiscoveryClientForConfig(mgr.GetConfig())
baseDiscoveryClient, err := discovery.NewDiscoveryClientForConfig(c.mgr.GetConfig())
if err != nil {
return fmt.Errorf("unable to create discovery client: %w", err)
}
Expand All @@ -598,48 +638,41 @@ func setupBoxcutter(

trackingCache, err := managedcache.NewTrackingCache(
ctrl.Log.WithName("trackingCache"),
mgr.GetConfig(),
c.mgr.GetConfig(),
crcache.Options{
Scheme: mgr.GetScheme(), Mapper: mgr.GetRESTMapper(),
Scheme: c.mgr.GetScheme(), Mapper: c.mgr.GetRESTMapper(),
},
)
if err != nil {
return fmt.Errorf("unable to create boxcutter tracking cache: %v", err)
}
if err := mgr.Add(trackingCache); err != nil {
if err := c.mgr.Add(trackingCache); err != nil {
return fmt.Errorf("unable to add tracking cache to manager: %v", err)
}

if err = (&controllers.ClusterExtensionRevisionReconciler{
Client: mgr.GetClient(),
Client: c.mgr.GetClient(),
RevisionEngine: machinery.NewRevisionEngine(
machinery.NewPhaseEngine(
machinery.NewObjectEngine(
mgr.GetScheme(), trackingCache, mgr.GetClient(),
ownerhandling.NewNative(mgr.GetScheme()),
machinery.NewComparator(ownerhandling.NewNative(mgr.GetScheme()), discoveryClient, mgr.GetScheme(), fieldOwnerPrefix),
c.mgr.GetScheme(), trackingCache, c.mgr.GetClient(),
ownerhandling.NewNative(c.mgr.GetScheme()),
machinery.NewComparator(ownerhandling.NewNative(c.mgr.GetScheme()), discoveryClient, c.mgr.GetScheme(), fieldOwnerPrefix),
fieldOwnerPrefix, fieldOwnerPrefix,
),
validation.NewClusterPhaseValidator(mgr.GetRESTMapper(), mgr.GetClient()),
validation.NewClusterPhaseValidator(c.mgr.GetRESTMapper(), c.mgr.GetClient()),
),
validation.NewRevisionValidator(), mgr.GetClient(),
validation.NewRevisionValidator(), c.mgr.GetClient(),
),
TrackingCache: trackingCache,
}).SetupWithManager(mgr); err != nil {
}).SetupWithManager(c.mgr); err != nil {
return fmt.Errorf("unable to setup ClusterExtensionRevision controller: %w", err)
}
return nil
}

func setupHelm(
mgr manager.Manager,
ceReconciler *controllers.ClusterExtensionReconciler,
preflights []applier.Preflight,
ceController crcontroller.Controller,
clusterExtensionFinalizers crfinalizer.Registerer,
regv1ManifestProvider applier.ManifestProvider,
) error {
coreClient, err := corev1client.NewForConfig(mgr.GetConfig())
func (c *helmCERConfigurator) Configure(ceReconciler *controllers.ClusterExtensionReconciler) error {
coreClient, err := corev1client.NewForConfig(c.mgr.GetConfig())
if err != nil {
return fmt.Errorf("unable to create core client: %w", err)
}
Expand All @@ -649,8 +682,8 @@ func setupHelm(
clientRestConfigMapper = action.SyntheticUserRestConfigMapper(clientRestConfigMapper)
}

cfgGetter, err := helmclient.NewActionConfigGetter(mgr.GetConfig(), mgr.GetRESTMapper(),
helmclient.StorageDriverMapper(action.ChunkedStorageDriverMapper(coreClient, mgr.GetAPIReader(), cfg.systemNamespace)),
cfgGetter, err := helmclient.NewActionConfigGetter(c.mgr.GetConfig(), c.mgr.GetRESTMapper(),
helmclient.StorageDriverMapper(action.ChunkedStorageDriverMapper(coreClient, c.mgr.GetAPIReader(), cfg.systemNamespace)),
helmclient.ClientNamespaceMapper(func(obj client.Object) (string, error) {
ext := obj.(*ocv1.ClusterExtension)
return ext.Spec.Namespace, nil
Expand All @@ -671,11 +704,11 @@ func setupHelm(
// determine if PreAuthorizer should be enabled based on feature gate
var preAuth authorization.PreAuthorizer
if features.OperatorControllerFeatureGate.Enabled(features.PreflightPermissions) {
preAuth = authorization.NewRBACPreAuthorizer(mgr.GetClient())
preAuth = authorization.NewRBACPreAuthorizer(c.mgr.GetClient())
}

cm := contentmanager.NewManager(clientRestConfigMapper, mgr.GetConfig(), mgr.GetRESTMapper())
err = clusterExtensionFinalizers.Register(controllers.ClusterExtensionCleanupContentManagerCacheFinalizer, finalizers.FinalizerFunc(func(ctx context.Context, obj client.Object) (crfinalizer.Result, error) {
cm := contentmanager.NewManager(clientRestConfigMapper, c.mgr.GetConfig(), c.mgr.GetRESTMapper())
err = c.finalizers.Register(controllers.ClusterExtensionCleanupContentManagerCacheFinalizer, finalizers.FinalizerFunc(func(ctx context.Context, obj client.Object) (crfinalizer.Result, error) {
ext := obj.(*ocv1.ClusterExtension)
err := cm.Delete(ext)
return crfinalizer.Result{}, err
Expand All @@ -686,18 +719,26 @@ func setupHelm(
}

// now initialize the helmApplier, assigning the potentially nil preAuth
ceReconciler.Applier = &applier.Helm{
appl := &applier.Helm{
ActionClientGetter: acg,
Preflights: preflights,
Preflights: c.preflights,
HelmChartProvider: &applier.RegistryV1HelmChartProvider{
ManifestProvider: regv1ManifestProvider,
ManifestProvider: c.regv1ManifestProvider,
},
HelmReleaseToObjectsConverter: &applier.HelmReleaseToObjectsConverter{},
PreAuthorizer: preAuth,
Watcher: ceController,
Watcher: c.watcher,
Manager: cm,
}
ceReconciler.RevisionStatesGetter = &controllers.HelmRevisionStatesGetter{ActionClientGetter: acg}
revisionStatesGetter := &controllers.HelmRevisionStatesGetter{ActionClientGetter: acg}
ceReconciler.ReconcileSteps = []controllers.ReconcileStepFunc{
controllers.HandleFinalizers(c.finalizers),
controllers.RetrieveRevisionStates(revisionStatesGetter),
controllers.RetrieveRevisionMetadata(c.resolver),
controllers.UnpackBundle(c.imagePuller, c.imageCache),
controllers.ApplyBundle(appl),
}

return nil
}

Expand Down
4 changes: 2 additions & 2 deletions internal/operator-controller/applier/helm.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,13 @@ import (
apimachyaml "k8s.io/apimachinery/pkg/util/yaml"
"k8s.io/klog/v2"
"sigs.k8s.io/controller-runtime/pkg/client"
crcontroller "sigs.k8s.io/controller-runtime/pkg/controller"

helmclient "github.com/operator-framework/helm-operator-plugins/pkg/client"

ocv1 "github.com/operator-framework/operator-controller/api/v1"
"github.com/operator-framework/operator-controller/internal/operator-controller/authorization"
"github.com/operator-framework/operator-controller/internal/operator-controller/contentmanager"
"github.com/operator-framework/operator-controller/internal/operator-controller/contentmanager/cache"
"github.com/operator-framework/operator-controller/internal/operator-controller/features"
"github.com/operator-framework/operator-controller/internal/operator-controller/rukpak/util"
imageutil "github.com/operator-framework/operator-controller/internal/shared/util/image"
Expand Down Expand Up @@ -65,7 +65,7 @@ type Helm struct {
HelmReleaseToObjectsConverter HelmReleaseToObjectsConverterInterface

Manager contentmanager.Manager
Watcher crcontroller.Controller
Watcher cache.Watcher
}

// runPreAuthorizationChecks performs pre-authorization checks for a Helm release
Expand Down
Loading
Loading