Repository: apache/camel-k
Pull request #2668 (2021-10-01 10:35:41)

chore: Parallel Kamelets reconciliation
As the number of Kamelets grows, so does the reconciliation requests that queue during the operator start. That PR enables Kamelets to be reconciled in parallel.

**Release Note**
```release-note
chore: Parallel Kamelets reconciliation
```



Patch:
diff --git a/pkg/cmd/operator/operator.go b/pkg/cmd/operator/operator.go
index d10bd66ac..2013142d2 100644
--- a/pkg/cmd/operator/operator.go
+++ b/pkg/cmd/operator/operator.go
@@ -21,7 +21,6 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"k8s.io/klog/v2"
 	"math/rand"
 	"os"
 	"runtime"
@@ -30,16 +29,23 @@ import (
 
 	coordination "k8s.io/api/coordination/v1"
 	corev1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
 
-	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
 
 	"github.com/apache/camel-k/pkg/apis"
+	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/client"
 	"github.com/apache/camel-k/pkg/controller"
 	"github.com/apache/camel-k/pkg/event"
@@ -51,7 +57,6 @@ import (
 
 var log = logf.Log.WithName("cmd")
 
-// GitCommit --
 var GitCommit string
 
 func printVersion() {
@@ -114,6 +119,10 @@ func Run(healthPort, monitoringPort int32, leaderElection bool) {
 		}
 	}
 
+	// Set the operator container image if it runs in-container
+	platform.OperatorImage, err = getOperatorImage(context.TODO(), c)
+	exitOnError(err, "cannot get operator container image")
+
 	if ok, err := kubernetes.CheckPermission(context.TODO(), c, coordination.GroupName, "leases", operatorNamespace, "", "create"); err != nil || !ok {
 		leaderElection = false
 		exitOnError(err, "cannot check permissions for creating Leases")
@@ -124,7 +133,10 @@ func Run(healthPort, monitoringPort int32, leaderElection bool) {
 		log.Info("Leader election is disabled!")
 	}
 
-	mgr, err := ctrl.NewManager(c.GetConfig(), ctrl.Options{
+	podLabelSelector, err := labels.NewRequirement(v1.IntegrationLabel, selection.Exists, []string{})
+	exitOnError(err, "cannot create Pod labels selector")
+
+	mgr, err := manager.New(c.GetConfig(), manager.Options{
 		Namespace:                     watchNamespace,
 		EventBroadcaster:              broadcaster,
 		LeaderElection:                leaderElection,
@@ -134,9 +146,26 @@ func Run(healthPort, monitoringPort int32, leaderElection bool) {
 		LeaderElectionReleaseOnCancel: true,
 		HealthProbeBindAddress:        ":" + strconv.Itoa(int(healthPort)),
 		MetricsBindAddress:            ":" + strconv.Itoa(int(monitoringPort)),
+		NewCache: cache.BuilderWithOptions(
+			cache.Options{
+				SelectorsByObject: cache.SelectorsByObject{
+					&corev1.Pod{}: {
+						Label: labels.NewSelector().Add(*podLabelSelector),
+					},
+				},
+			},
+		),
 	})
 	exitOnError(err, "")
 
+	exitOnError(
+		mgr.GetFieldIndexer().IndexField(context.Background(), &corev1.Pod{}, "status.phase",
+			func(obj ctrl.Object) []string {
+				return []string{string(obj.(*corev1.Pod).Status.Phase)}
+			}),
+		"unable to set up field indexer for status.phase: %v",
+	)
+
 	log.Info("Configuring manager")
 	exitOnError(mgr.AddHealthzCheck("health-probe", healthz.Ping), "Unable add liveness check")
 	exitOnError(apis.AddToScheme(mgr.GetScheme()), "")
@@ -160,6 +189,26 @@ func getWatchNamespace() (string, error) {
 	return ns, nil
 }
 
+// getOperatorImage returns the image currently used by the running operator if present (when running out of cluster, it may be absent).
+func getOperatorImage(ctx context.Context, c ctrl.Reader) (string, error) {
+	ns := platform.GetOperatorNamespace()
+	name := platform.GetOperatorPodName()
+	if ns == "" || name == "" {
+		return "", nil
+	}
+
+	pod := corev1.Pod{}
+	if err := c.Get(ctx, ctrl.ObjectKey{Namespace: ns, Name: name}, &pod); err != nil && k8serrors.IsNotFound(err) {
+		return "", nil
+	} else if err != nil {
+		return "", err
+	}
+	if len(pod.Spec.Containers) == 0 {
+		return "", fmt.Errorf("no containers found in operator pod")
+	}
+	return pod.Spec.Containers[0].Image, nil
+}
+
 func exitOnError(err error, msg string) {
 	if err != nil {
 		log.Error(err, msg)
diff --git a/pkg/controller/build/build_controller.go b/pkg/controller/build/build_controller.go
index fb9e671bb..99ca1d0bd 100644
--- a/pkg/controller/build/build_controller.go
+++ b/pkg/controller/build/build_controller.go
@@ -21,21 +21,17 @@ import (
 	"context"
 	"time"
 
-	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-
 	"k8s.io/client-go/tools/record"
 
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	ctrl "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/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/client"
@@ -54,7 +50,6 @@ func Add(mgr manager.Manager) error {
 	return add(mgr, newReconciler(mgr, c))
 }
 
-// newReconciler returns a new reconcile.Reconciler
 func newReconciler(mgr manager.Manager, c client.Client) reconcile.Reconciler {
 	return monitoring.NewInstrumentedReconciler(
 		&reconcileBuild{
@@ -71,54 +66,23 @@ func newReconciler(mgr manager.Manager, c client.Client) reconcile.Reconciler {
 	)
 }
 
-// add adds a new Controller to mgr with r as the reconcile.Reconciler
 func add(mgr manager.Manager, r reconcile.Reconciler) error {
-	// Create a new controller
-	c, err := controller.New("build-controller", mgr, controller.Options{Reconciler: r})
-	if err != nil {
-		return err
-	}
-
-	// Watch for changes to primary resource Build
-	err = c.Watch(&source.Kind{Type: &v1.Build{}},
-		&handler.EnqueueRequestForObject{},
-		predicate.Funcs{
-			UpdateFunc: func(e event.UpdateEvent) bool {
-				oldBuild := e.ObjectOld.(*v1.Build)
-				newBuild := e.ObjectNew.(*v1.Build)
-				// Ignore updates to the build status in which case metadata.Generation does not change,
-				// or except when the build phase changes as it's used to transition from one phase
-				// to another
-				return oldBuild.Generation != newBuild.Generation ||
-					oldBuild.Status.Phase != newBuild.Status.Phase
-			},
-		},
-	)
-	if err != nil {
-		return err
-	}
-
-	// Watch for changes to secondary resource Pods and requeue the owner Build
-	err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
-		&handler.EnqueueRequestForOwner{
-			IsController: true,
-			OwnerType:    &v1.Build{},
-		},
-		predicate.Funcs{
-			UpdateFunc: func(e event.UpdateEvent) bool {
-				oldPod := e.ObjectOld.(*corev1.Pod)
-				newPod := e.ObjectNew.(*corev1.Pod)
-				// Ignore updates to the build pods except when the pod phase changes
-				// as it's used to transition the builds from one phase to another
-				return oldPod.Status.Phase != newPod.Status.Phase
-			},
-		},
-	)
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return builder.ControllerManagedBy(mgr).
+		Named("build-controller").
+		// Watch for changes to primary resource Build
+		For(&v1.Build{}, builder.WithPredicates(
+			predicate.Funcs{
+				UpdateFunc: func(e event.UpdateEvent) bool {
+					oldBuild := e.ObjectOld.(*v1.Build)
+					newBuild := e.ObjectNew.(*v1.Build)
+					// Ignore updates to the build status in which case metadata.Generation does not change,
+					// or except when the build phase changes as it's used to transition from one phase
+					// to another
+					return oldBuild.Generation != newBuild.Generation ||
+						oldBuild.Status.Phase != newBuild.Status.Phase
+				},
+			})).
+		Complete(r)
 }
 
 var _ reconcile.Reconciler = &reconcileBuild{}
@@ -197,9 +161,9 @@ func (r *reconcileBuild) Reconcile(ctx context.Context, request reconcile.Reques
 	switch pl.Status.Build.BuildStrategy {
 	case v1.IntegrationPlatformBuildStrategyPod:
 		actions = []Action{
-			newInitializePodAction(),
+			newInitializePodAction(r.reader),
 			newScheduleAction(r.reader),
-			newMonitorPodAction(),
+			newMonitorPodAction(r.reader),
 			newErrorRecoveryAction(),
 			newErrorAction(),
 		}
@@ -253,13 +217,12 @@ func (r *reconcileBuild) Reconcile(ctx context.Context, request reconcile.Reques
 
 	if target.Status.Phase == v1.BuildPhaseScheduling || target.Status.Phase == v1.BuildPhaseFailed {
 		// Requeue scheduling (resp. failed) build so that it re-enters the build (resp. recovery) working queue
-		return reconcile.Result{
-			RequeueAfter: 5 * time.Second,
-		}, nil
+		return reconcile.Result{RequeueAfter: 5 * time.Second}, nil
 	}
 
-	if pl.Status.Build.BuildStrategy == v1.IntegrationPlatformBuildStrategyPod && (target.Status.Phase == v1.BuildPhasePending || target.Status.Phase == v1.BuildPhaseRunning) {
-		// Requeue running Build to signal timeout to the Build pod
+	if pl.Status.Build.BuildStrategy == v1.IntegrationPlatformBuildStrategyPod &&
+		(target.Status.Phase == v1.BuildPhasePending || target.Status.Phase == v1.BuildPhaseRunning) {
+		// Requeue running Build to poll Pod and signal timeout
 		return reconcile.Result{RequeueAfter: 1 * time.Second}, nil
 	}
 
diff --git a/pkg/controller/build/build_pod.go b/pkg/controller/build/build_pod.go
index e81cadf6d..3cea93801 100644
--- a/pkg/controller/build/build_pod.go
+++ b/pkg/controller/build/build_pod.go
@@ -34,7 +34,6 @@ import (
 
 	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/builder"
-	"github.com/apache/camel-k/pkg/client"
 	"github.com/apache/camel-k/pkg/platform"
 	"github.com/apache/camel-k/pkg/util/defaults"
 	"github.com/apache/camel-k/pkg/util/kubernetes"
@@ -114,7 +113,7 @@ var (
 	}
 )
 
-func newBuildPod(ctx context.Context, c client.Client, build *v1.Build) (*corev1.Pod, error) {
+func newBuildPod(ctx context.Context, c ctrl.Reader, build *v1.Build) (*corev1.Pod, error) {
 	pod := &corev1.Pod{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: corev1.SchemeGroupVersion.String(),
@@ -138,7 +137,7 @@ func newBuildPod(ctx context.Context, c client.Client, build *v1.Build) (*corev1
 
 	for _, task := range build.Spec.Tasks {
 		if task.Builder != nil {
-			err := addBuildTaskToPod(ctx, c, build, task.Builder.Name, pod)
+			err := addBuildTaskToPod(build, task.Builder.Name, pod)
 			if err != nil {
 				return nil, err
 			}
@@ -153,12 +152,12 @@ func newBuildPod(ctx context.Context, c client.Client, build *v1.Build) (*corev1
 				return nil, err
 			}
 		} else if task.S2i != nil {
-			err := addBuildTaskToPod(ctx, c, build, task.S2i.Name, pod)
+			err := addBuildTaskToPod(build, task.S2i.Name, pod)
 			if err != nil {
 				return nil, err
 			}
 		} else if task.Spectrum != nil {
-			err := addBuildTaskToPod(ctx, c, build, task.Spectrum.Name, pod)
+			err := addBuildTaskToPod(build, task.Spectrum.Name, pod)
 			if err != nil {
 				return nil, err
 			}
@@ -209,7 +208,7 @@ func buildPodName(build *v1.Build) string {
 	return "camel-k-" + build.Name + "-builder"
 }
 
-func addBuildTaskToPod(ctx context.Context, c ctrl.Reader, build *v1.Build, taskName string, pod *corev1.Pod) error {
+func addBuildTaskToPod(build *v1.Build, taskName string, pod *corev1.Pod) error {
 	if !hasBuilderVolume(pod) {
 		// Add the EmptyDir volume used to share the build state across tasks
 		pod.Spec.Volumes = append(pod.Spec.Volumes, corev1.Volume{
@@ -220,11 +219,7 @@ func addBuildTaskToPod(ctx context.Context, c ctrl.Reader, build *v1.Build, task
 		})
 	}
 
-	// TODO: Move the retrieval of the operator image into the controller
-	operatorImage, err := platform.GetCurrentOperatorImage(ctx, c)
-	if err != nil {
-		return err
-	}
+	operatorImage := platform.OperatorImage
 	if operatorImage == "" {
 		operatorImage = defaults.ImageName + ":" + defaults.Version
 	}
@@ -250,7 +245,7 @@ func addBuildTaskToPod(ctx context.Context, c ctrl.Reader, build *v1.Build, task
 	return nil
 }
 
-func addBuildahTaskToPod(ctx context.Context, c client.Client, build *v1.Build, task *v1.BuildahTask, pod *corev1.Pod) error {
+func addBuildahTaskToPod(ctx context.Context, c ctrl.Reader, build *v1.Build, task *v1.BuildahTask, pod *corev1.Pod) error {
 	bud := []string{
 		"buildah",
 		"bud",
diff --git a/pkg/controller/build/initialize_pod.go b/pkg/controller/build/initialize_pod.go
index bb0fb1ba7..4fd8cb80d 100644
--- a/pkg/controller/build/initialize_pod.go
+++ b/pkg/controller/build/initialize_pod.go
@@ -20,17 +20,22 @@ package build
 import (
 	"context"
 
+	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
+
 	"github.com/pkg/errors"
 
 	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 )
 
-func newInitializePodAction() Action {
-	return &initializePodAction{}
+func newInitializePodAction(reader ctrl.Reader) Action {
+	return &initializePodAction{
+		reader: reader,
+	}
 }
 
 type initializePodAction struct {
 	baseAction
+	reader ctrl.Reader
 }
 
 // Name returns a common name of the action
@@ -49,7 +54,7 @@ func (action *initializePodAction) Handle(ctx context.Context, build *v1.Build)
 		return nil, errors.Wrap(err, "cannot delete build pod")
 	}
 
-	pod, err := getBuilderPod(ctx, action.client, build)
+	pod, err := getBuilderPod(ctx, action.reader, build)
 	if err != nil || pod != nil {
 		// We return and wait for the pod to be deleted before de-queue the build pod.
 		return nil, err
diff --git a/pkg/controller/build/monitor_pod.go b/pkg/controller/build/monitor_pod.go
index 3c3bf1307..5ec3b619e 100644
--- a/pkg/controller/build/monitor_pod.go
+++ b/pkg/controller/build/monitor_pod.go
@@ -39,12 +39,15 @@ import (
 
 const timeoutAnnotation = "camel.apache.org/timeout"
 
-func newMonitorPodAction() Action {
-	return &monitorPodAction{}
+func newMonitorPodAction(reader ctrl.Reader) Action {
+	return &monitorPodAction{
+		reader: reader,
+	}
 }
 
 type monitorPodAction struct {
 	baseAction
+	reader ctrl.Reader
 }
 
 // Name returns a common name of the action
@@ -59,7 +62,7 @@ func (action *monitorPodAction) CanHandle(build *v1.Build) bool {
 
 // Handle handles the builds
 func (action *monitorPodAction) Handle(ctx context.Context, build *v1.Build) (*v1.Build, error) {
-	pod, err := getBuilderPod(ctx, action.client, build)
+	pod, err := getBuilderPod(ctx, action.reader, build)
 	if err != nil {
 		return nil, err
 	}
@@ -68,7 +71,7 @@ func (action *monitorPodAction) Handle(ctx context.Context, build *v1.Build) (*v
 		switch build.Status.Phase {
 
 		case v1.BuildPhasePending:
-			if pod, err = newBuildPod(ctx, action.client, build); err != nil {
+			if pod, err = newBuildPod(ctx, action.reader, build); err != nil {
 				return nil, err
 			}
 			// Set the Build as the Pod owner and controller
diff --git a/pkg/controller/integration/integration_controller.go b/pkg/controller/integration/integration_controller.go
index 187b6db71..962a62c60 100644
--- a/pkg/controller/integration/integration_controller.go
+++ b/pkg/controller/integration/integration_controller.go
@@ -20,16 +20,15 @@ package integration
 import (
 	"context"
 
-	appsv1 "k8s.io/api/apps/v1"
-	"k8s.io/api/batch/v1beta1"
+	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	ctrl "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/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -51,7 +50,7 @@ func Add(mgr manager.Manager) error {
 	if err != nil {
 		return err
 	}
-	return add(mgr, newReconciler(mgr, c), c)
+	return add(mgr, newReconciler(mgr, c))
 }
 
 func newReconciler(mgr manager.Manager, c client.Client) reconcile.Reconciler {
@@ -69,109 +68,59 @@ func newReconciler(mgr manager.Manager, c client.Client) reconcile.Reconciler {
 	)
 }
 
-func add(mgr manager.Manager, r reconcile.Reconciler, cl client.Client) error {
-	// Create a new controller
-	c, err := controller.New("integration-controller", mgr, controller.Options{Reconciler: r})
-	if err != nil {
-		return err
-	}
-
-	// Watch for changes to primary resource Integration
-	err = c.Watch(&source.Kind{Type: &v1.Integration{}},
-		&handler.EnqueueRequestForObject{},
-		predicate.Funcs{
-			UpdateFunc: func(e event.UpdateEvent) bool {
-				oldIntegration := e.ObjectOld.(*v1.Integration)
-				newIntegration := e.ObjectNew.(*v1.Integration)
-				// Ignore updates to the integration status in which case metadata.Generation does not change,
-				// or except when the integration phase changes as it's used to transition from one phase
-				// to another.
-				return oldIntegration.Generation != newIntegration.Generation ||
-					oldIntegration.Status.Phase != newIntegration.Status.Phase
-			},
-			DeleteFunc: func(e event.DeleteEvent) bool {
-				// Evaluates to false if the object has been confirmed deleted
-				return !e.DeleteStateUnknown
-			},
-		},
-	)
-	if err != nil {
-		return err
-	}
-
-	// Watch for IntegrationKit phase transitioning to ready or error, and
-	// enqueue requests for any integration that matches the kit, in building
-	// or running phase.
-	err = c.Watch(&source.Kind{Type: &v1.IntegrationKit{}},
-		handler.EnqueueRequestsFromMapFunc(func(a ctrl.Object) []reconcile.Request {
-			kit := a.(*v1.IntegrationKit)
-			var requests []reconcile.Request
-
-			if kit.Status.Phase != v1.IntegrationKitPhaseReady && kit.Status.Phase != v1.IntegrationKitPhaseError {
-				return requests
-			}
-
-			list := &v1.IntegrationList{}
-			// Do global search in case of global operator (it may be using a global platform)
-			var opts []ctrl.ListOption
-			if !platform.IsCurrentOperatorGlobal() {
-				opts = append(opts, ctrl.InNamespace(kit.Namespace))
-			}
-			if err := mgr.GetClient().List(context.Background(), list, opts...); err != nil {
-				log.Error(err, "Failed to retrieve integration list")
-				return requests
-			}
-
-			for _, integration := range list.Items {
-				if match, err := integrationMatches(&integration, kit); err != nil {
-					log.Errorf(err, "Error matching integration %q with kit %q", integration.Name, kit.Name)
-					continue
-				} else if !match {
-					continue
-				}
-				if integration.Status.Phase == v1.IntegrationPhaseBuildingKit ||
-					integration.Status.Phase == v1.IntegrationPhaseRunning {
-					log.Infof("Kit %s ready, notify integration: %s", kit.Name, integration.Name)
-					requests = append(requests, reconcile.Request{
-						NamespacedName: types.NamespacedName{
-							Namespace: integration.Namespace,
-							Name:      integration.Name,
-						},
-					})
+func add(mgr manager.Manager, r reconcile.Reconciler) error {
+	return builder.ControllerManagedBy(mgr).
+		Named("integration-controller").
+		// Watch for changes to primary resource Integration
+		For(&v1.Integration{}, builder.WithPredicates(
+			predicate.Funcs{
+				UpdateFunc: func(e event.UpdateEvent) bool {
+					oldIntegration := e.ObjectOld.(*v1.Integration)
+					newIntegration := e.ObjectNew.(*v1.Integration)
+					// Ignore updates to the integration status in which case metadata.Generation does not change,
+					// or except when the integration phase changes as it's used to transition from one phase
+					// to another.
+					return oldIntegration.Generation != newIntegration.Generation ||
+						oldIntegration.Status.Phase != newIntegration.Status.Phase
+				},
+				DeleteFunc: func(e event.DeleteEvent) bool {
+					// Evaluates to false if the object has been confirmed deleted
+					return !e.DeleteStateUnknown
+				},
+			})).
+		// Watch for IntegrationKit phase transitioning to ready or error, and
+		// enqueue requests for any integration that matches the kit, in building
+		// or running phase.
+		Watches(&source.Kind{Type: &v1.IntegrationKit{}},
+			handler.EnqueueRequestsFromMapFunc(func(a ctrl.Object) []reconcile.Request {
+				kit := a.(*v1.IntegrationKit)
+				var requests []reconcile.Request
+
+				if kit.Status.Phase != v1.IntegrationKitPhaseReady && kit.Status.Phase != v1.IntegrationKitPhaseError {
+					return requests
 				}
-			}
-
-			return requests
-		}),
-	)
-	if err != nil {
-		return err
-	}
-
-	// Watch for IntegrationPlatform phase transitioning to ready and enqueue
-	// requests for any integrations that are in phase waiting for platform
-	err = c.Watch(&source.Kind{Type: &v1.IntegrationPlatform{}},
-		handler.EnqueueRequestsFromMapFunc(func(a ctrl.Object) []reconcile.Request {
-			p := a.(*v1.IntegrationPlatform)
-			var requests []reconcile.Request
 
-			if p.Status.Phase == v1.IntegrationPlatformPhaseReady {
 				list := &v1.IntegrationList{}
-
 				// Do global search in case of global operator (it may be using a global platform)
 				var opts []ctrl.ListOption
 				if !platform.IsCurrentOperatorGlobal() {
-					opts = append(opts, ctrl.InNamespace(p.Namespace))
+					opts = append(opts, ctrl.InNamespace(kit.Namespace))
 				}
-
-				if err := mgr.GetClient().List(context.TODO(), list, opts...); err != nil {
-					log.Error(err, "Failed to list integrations")
+				if err := mgr.GetClient().List(context.Background(), list, opts...); err != nil {
+					log.Error(err, "Failed to retrieve integration list")
 					return requests
 				}
 
 				for _, integration := range list.Items {
-					if integration.Status.Phase == v1.IntegrationPhaseWaitingForPlatform {
-						log.Infof("Platform %s ready, wake-up integration: %s", p.Name, integration.Name)
+					if match, err := integrationMatches(&integration, kit); err != nil {
+						log.Errorf(err, "Error matching integration %q with kit %q", integration.Name, kit.Name)
+						continue
+					} else if !match {
+						continue
+					}
+					if integration.Status.Phase == v1.IntegrationPhaseBuildingKit ||
+						integration.Status.Phase == v1.IntegrationPhaseRunning {
+						log.Infof("Kit %s ready, notify integration: %s", kit.Name, integration.Name)
 						requests = append(requests, reconcile.Request{
 							NamespacedName: types.NamespacedName{
 								Namespace: integration.Namespace,
@@ -180,63 +129,59 @@ func add(mgr manager.Manager, r reconcile.Reconciler, cl client.Client) error {
 						})
 					}
 				}
-			}
 
-			return requests
-		}),
-	)
-	if err != nil {
-		return err
-	}
-
-	// Watch for ReplicaSet to reconcile replicas to the integration status. We cannot use
-	// the EnqueueRequestForOwner handler as the owner depends on the deployment strategy,
-	// either regular deployment or Knative service. In any case, the integration is not the
-	// direct owner of the ReplicaSet.
-	err = c.Watch(&source.Kind{Type: &appsv1.ReplicaSet{}},
-		handler.EnqueueRequestsFromMapFunc(func(a ctrl.Object) []reconcile.Request {
-			rs := a.(*appsv1.ReplicaSet)
-			var requests []reconcile.Request
-
-			labels := rs.GetLabels()
-			integrationName, ok := labels[v1.IntegrationLabel]
-			if ok {
-				requests = append(requests, reconcile.Request{
-					NamespacedName: types.NamespacedName{
-						Namespace: rs.Namespace,
-						Name:      integrationName,
-					},
-				})
-			}
+				return requests
+			})).
+		// Watch for IntegrationPlatform phase transitioning to ready and enqueue
+		// requests for any integrations that are in phase waiting for platform
+		Watches(&source.Kind{Type: &v1.IntegrationPlatform{}},
+			handler.EnqueueRequestsFromMapFunc(func(a ctrl.Object) []reconcile.Request {
+				p := a.(*v1.IntegrationPlatform)
+				var requests []reconcile.Request
+
+				if p.Status.Phase == v1.IntegrationPlatformPhaseReady {
+					list := &v1.IntegrationList{}
+
+					// Do global search in case of global operator (it may be using a global platform)
+					var opts []ctrl.ListOption
+					if !platform.IsCurrentOperatorGlobal() {
+						opts = append(opts, ctrl.InNamespace(p.Namespace))
+					}
 
-			return requests
-		}),
-		predicate.Funcs{
-			UpdateFunc: func(e event.UpdateEvent) bool {
-				oldReplicaSet := e.ObjectOld.(*appsv1.ReplicaSet)
-				newReplicaSet := e.ObjectNew.(*appsv1.ReplicaSet)
-				// Ignore updates to the ReplicaSet other than the replicas ones,
-				// that are used to reconcile the integration replicas.
-				return oldReplicaSet.Status.Replicas != newReplicaSet.Status.Replicas ||
-					oldReplicaSet.Status.ReadyReplicas != newReplicaSet.Status.ReadyReplicas ||
-					oldReplicaSet.Status.AvailableReplicas != newReplicaSet.Status.AvailableReplicas
-			},
-		},
-	)
-	if err != nil {
-		return err
-	}
+					if err := mgr.GetClient().List(context.Background(), list, opts...); err != nil {
+						log.Error(err, "Failed to list integrations")
+						return requests
+					}
 
-	// Watch for CronJob to update the ready condition
-	err = c.Watch(&source.Kind{Type: &v1beta1.CronJob{}}, &handler.EnqueueRequestForOwner{
-		OwnerType:    &v1.Integration{},
-		IsController: false,
-	})
-	if err != nil {
-		return err
-	}
+					for _, integration := range list.Items {
+						if integration.Status.Phase == v1.IntegrationPhaseWaitingForPlatform {
+							log.Infof("Platform %s ready, wake-up integration: %s", p.Name, integration.Name)
+							requests = append(requests, reconcile.Request{
+								NamespacedName: types.NamespacedName{
+									Namespace: integration.Namespace,
+									Name:      integration.Name,
+								},
+							})
+						}
+					}
+				}
 
-	return nil
+				return requests
+			})).
+		// Watch for the Integration Pods
+		Watches(&source.Kind{Type: &corev1.Pod{}},
+			handler.EnqueueRequestsFromMapFunc(func(a ctrl.Object) []reconcile.Request {
+				pod := a.(*corev1.Pod)
+				return []reconcile.Request{
+					{
+						NamespacedName: types.NamespacedName{
+							Namespace: pod.GetNamespace(),
+							Name:      pod.Labels[v1.IntegrationLabel],
+						},
+					},
+				}
+			})).
+		Complete(r)
 }
 
 var _ reconcile.Reconciler = &reconcileIntegration{}
diff --git a/pkg/controller/integration/monitor.go b/pkg/controller/integration/monitor.go
index a475abec6..33911aebc 100644
--- a/pkg/controller/integration/monitor.go
+++ b/pkg/controller/integration/monitor.go
@@ -36,7 +36,6 @@ import (
 	"github.com/apache/camel-k/pkg/util/kubernetes"
 )
 
-// NewMonitorAction creates a new monitoring action for an integration
 func NewMonitorAction() Action {
 	return &monitorAction{}
 }
@@ -115,25 +114,32 @@ func (action *monitorAction) Handle(ctx context.Context, integration *v1.Integra
 	// to list the pods owned by the integration.
 	integration.Status.Selector = v1.IntegrationLabel + "=" + integration.Name
 
-	// Check replicas
-	replicaSets := &appsv1.ReplicaSetList{}
-	err = action.client.List(ctx, replicaSets,
+	// Update the replicas count
+	pendingPods := &corev1.PodList{}
+	err = action.client.List(ctx, pendingPods,
 		ctrl.InNamespace(integration.Namespace),
-		ctrl.MatchingLabels{
-			v1.IntegrationLabel: integration.Name,
-		})
+		ctrl.MatchingLabels{v1.IntegrationLabel: integration.Name},
+		ctrl.MatchingFields{"status.phase": string(corev1.PodPending)})
 	if err != nil {
 		return nil, err
 	}
-
-	// And update the scale status accordingly
-	if len(replicaSets.Items) > 0 {
-		replicaSet := findLatestReplicaSet(replicaSets)
-		replicas := replicaSet.Status.Replicas
-		if integration.Status.Replicas == nil || replicas != *integration.Status.Replicas {
-			integration.Status.Replicas = &replicas
+	runningPods := &corev1.PodList{}
+	err = action.client.List(ctx, runningPods,
+		ctrl.InNamespace(integration.Namespace),
+		ctrl.MatchingLabels{v1.IntegrationLabel: integration.Name},
+		ctrl.MatchingFields{"status.phase": string(corev1.PodRunning)})
+	if err != nil {
+		return nil, err
+	}
+	nonTerminatingPods := 0
+	for _, pod := range runningPods.Items {
+		if pod.DeletionTimestamp != nil {
+			continue
 		}
+		nonTerminatingPods++
 	}
+	podCount := int32(len(pendingPods.Items) + nonTerminatingPods)
+	integration.Status.Replicas = &podCount
 
 	// Reconcile Integration phase
 	if integration.Status.Phase == v1.IntegrationPhaseDeploying {
@@ -218,16 +224,6 @@ func (action *monitorAction) Handle(ctx context.Context, integration *v1.Integra
 	return integration, nil
 }
 
-func findLatestReplicaSet(list *appsv1.ReplicaSetList) *appsv1.ReplicaSet {
-	latest := list.Items[0]
-	for i, rs := range list.Items[1:] {
-		if latest.CreationTimestamp.Before(&rs.CreationTimestamp) {
-			latest = list.Items[i+1]
-		}
-	}
-	return &latest
-}
-
 func findHighestPriorityReadyKit(kits []v1.IntegrationKit) (*v1.IntegrationKit, error) {
 	if len(kits) == 0 {
 		return nil, nil
diff --git a/pkg/controller/integrationplatform/integrationplatform_controller.go b/pkg/controller/integrationplatform/integrationplatform_controller.go
index 5fa590ebd..264ae4d51 100644
--- a/pkg/controller/integrationplatform/integrationplatform_controller.go
+++ b/pkg/controller/integrationplatform/integrationplatform_controller.go
@@ -57,6 +57,7 @@ func newReconciler(mgr manager.Manager, c client.Client) reconcile.Reconciler {
 	return monitoring.NewInstrumentedReconciler(
 		&reconcileIntegrationPlatform{
 			client:   c,
+			reader:   mgr.GetAPIReader(),
 			scheme:   mgr.GetScheme(),
 			recorder: mgr.GetEventRecorderFor("camel-k-integration-platform-controller"),
 		},
@@ -108,7 +109,9 @@ var _ reconcile.Reconciler = &reconcileIntegrationPlatform{}
 type reconcileIntegrationPlatform struct {
 	// This client, initialized using mgr.Client() above, is a split client
 	// that reads objects from the cache and writes to the API server
-	client   client.Client
+	client client.Client
+	// Non-caching client
+	reader   ctrl.Reader
 	scheme   *runtime.Scheme
 	recorder record.EventRecorder
 }
@@ -148,7 +151,7 @@ func (r *reconcileIntegrationPlatform) Reconcile(ctx context.Context, request re
 
 	actions := []Action{
 		NewInitializeAction(),
-		NewWarmAction(),
+		NewWarmAction(r.reader),
 		NewCreateAction(),
 		NewMonitorAction(),
 	}
diff --git a/pkg/controller/integrationplatform/warm.go b/pkg/controller/integrationplatform/warm.go
index 7fd48aaf2..21f6608c9 100644
--- a/pkg/controller/integrationplatform/warm.go
+++ b/pkg/controller/integrationplatform/warm.go
@@ -25,16 +25,20 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 
+	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
+
 	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 )
 
-// NewWarmAction returns a action that creates resources needed by the platform
-func NewWarmAction() Action {
-	return &warmAction{}
+func NewWarmAction(reader ctrl.Reader) Action {
+	return &warmAction{
+		reader: reader,
+	}
 }
 
 type warmAction struct {
 	baseAction
+	reader ctrl.Reader
 }
 
 func (action *warmAction) Name() string {
@@ -58,7 +62,7 @@ func (action *warmAction) Handle(ctx context.Context, platform *v1.IntegrationPl
 		},
 	}
 
-	err := action.client.Get(ctx, types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, &pod)
+	err := action.reader.Get(ctx, types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, &pod)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/controller/kamelet/kamelet_controller.go b/pkg/controller/kamelet/kamelet_controller.go
index 84729dfb4..fc12f1926 100644
--- a/pkg/controller/kamelet/kamelet_controller.go
+++ b/pkg/controller/kamelet/kamelet_controller.go
@@ -19,6 +19,7 @@ package kamelet
 
 import (
 	"context"
+	goruntime "runtime"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -26,14 +27,13 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/tools/record"
 
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	ctrl "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/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/apache/camel-k/pkg/apis/camel/v1alpha1"
 	"github.com/apache/camel-k/pkg/client"
@@ -52,7 +52,6 @@ func Add(mgr manager.Manager) error {
 	return add(mgr, newReconciler(mgr, c))
 }
 
-// newReconciler returns a new reconcile.Reconciler
 func newReconciler(mgr manager.Manager, c client.Client) reconcile.Reconciler {
 	return monitoring.NewInstrumentedReconciler(
 		&reconcileKamelet{
@@ -68,38 +67,30 @@ func newReconciler(mgr manager.Manager, c client.Client) reconcile.Reconciler {
 	)
 }
 
-// add adds a new Controller to mgr with r as the reconcile.Reconciler
 func add(mgr manager.Manager, r reconcile.Reconciler) error {
-	// Create a new controller
-	c, err := controller.New("kamelet-controller", mgr, controller.Options{Reconciler: r})
-	if err != nil {
-		return err
-	}
-
-	// Watch for changes to primary resource Kamelet
-	err = c.Watch(&source.Kind{Type: &v1alpha1.Kamelet{}},
-		&handler.EnqueueRequestForObject{},
-		predicate.Funcs{
-			UpdateFunc: func(e event.UpdateEvent) bool {
-				oldKamelet := e.ObjectOld.(*v1alpha1.Kamelet)
-				newKamelet := e.ObjectNew.(*v1alpha1.Kamelet)
-				// Ignore updates to the kamelet status in which case metadata.Generation
-				// does not change, or except when the kamelet phase changes as it's used
-				// to transition from one phase to another
-				return oldKamelet.Generation != newKamelet.Generation ||
-					oldKamelet.Status.Phase != newKamelet.Status.Phase
-			},
-			DeleteFunc: func(e event.DeleteEvent) bool {
-				// Evaluates to false if the object has been confirmed deleted
-				return !e.DeleteStateUnknown
-			},
-		},
-	)
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return builder.ControllerManagedBy(mgr).
+		Named("kamelet-controller").
+		// Watch for changes to primary resource Kamelet
+		For(&v1alpha1.Kamelet{}, builder.WithPredicates(
+			predicate.Funcs{
+				UpdateFunc: func(e event.UpdateEvent) bool {
+					oldKamelet := e.ObjectOld.(*v1alpha1.Kamelet)
+					newKamelet := e.ObjectNew.(*v1alpha1.Kamelet)
+					// Ignore updates to the Kamelet status in which case metadata.Generation
+					// does not change, or except when the Kamelet phase changes as it's used
+					// to transition from one phase to another
+					return oldKamelet.Generation != newKamelet.Generation ||
+						oldKamelet.Status.Phase != newKamelet.Status.Phase
+				},
+				DeleteFunc: func(e event.DeleteEvent) bool {
+					// Evaluates to false if the object has been confirmed deleted
+					return !e.DeleteStateUnknown
+				},
+			})).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: goruntime.GOMAXPROCS(0),
+		}).
+		Complete(r)
 }
 
 var _ reconcile.Reconciler = &reconcileKamelet{}
diff --git a/pkg/platform/operator.go b/pkg/platform/operator.go
index 74921585a..8842ab666 100644
--- a/pkg/platform/operator.go
+++ b/pkg/platform/operator.go
@@ -19,12 +19,10 @@ package platform
 
 import (
 	"context"
-	"errors"
 	"os"
 	"strings"
 
 	coordination "k8s.io/api/coordination/v1"
-	v1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 
 	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
@@ -36,25 +34,7 @@ const operatorPodNameEnvVariable = "POD_NAME"
 
 const OperatorLockName = "camel-k-lock"
 
-// GetCurrentOperatorImage returns the image currently used by the running operator if present (when running out of cluster, it may be absent).
-func GetCurrentOperatorImage(ctx context.Context, c ctrl.Reader) (string, error) {
-	ns := GetOperatorNamespace()
-	name := GetOperatorPodName()
-	if ns == "" || name == "" {
-		return "", nil
-	}
-
-	pod := v1.Pod{}
-	if err := c.Get(ctx, ctrl.ObjectKey{Namespace: ns, Name: name}, &pod); err != nil && k8serrors.IsNotFound(err) {
-		return "", nil
-	} else if err != nil {
-		return "", err
-	}
-	if len(pod.Spec.Containers) == 0 {
-		return "", errors.New("no containers found in operator pod")
-	}
-	return pod.Spec.Containers[0].Image, nil
-}
+var OperatorImage string
 
 // IsCurrentOperatorGlobal returns true if the operator is configured to watch all namespaces
 func IsCurrentOperatorGlobal() bool {


Test patch:
diff --git a/e2e/common/operator_metrics_test.go b/e2e/common/operator_metrics_test.go
index e9a0c1097..ad3aeb40b 100644
--- a/e2e/common/operator_metrics_test.go
+++ b/e2e/common/operator_metrics_test.go
@@ -218,7 +218,23 @@ func TestMetrics(t *testing.T) {
 				platformRequeuedCount = *platformRequeued.Histogram.SampleCount
 			}
 
-			Expect(platformReconciliations).To(BeNumerically("==", platformReconciledCount+platformRequeuedCount))
+			platformErrored := getMetric(metrics["camel_k_reconciliation_duration_seconds"],
+				MatchFieldsP(IgnoreExtras, Fields{
+					"Label": ConsistOf(
+						label("group", v1.SchemeGroupVersion.Group),
+						label("version", v1.SchemeGroupVersion.Version),
+						label("kind", "IntegrationPlatform"),
+						label("namespace", ns),
+						label("result", "Errored"),
+						label("tag", "PlatformError"),
+					),
+				}))
+			platformErroredCount := uint64(0)
+			if platformErrored != nil {
+				platformErroredCount = *platformErrored.Histogram.SampleCount
+			}
+
+			Expect(platformReconciliations).To(BeNumerically("==", platformReconciledCount+platformRequeuedCount+platformErroredCount))
 
 			// Count the number of Integration reconciliations
 			integrationReconciliations, err := counter.Count(MatchFields(IgnoreExtras, Fields{
diff --git a/pkg/controller/integrationplatform/warm_test.go b/pkg/controller/integrationplatform/warm_test.go
index d580a2dbc..3c5a6c4d3 100644
--- a/pkg/controller/integrationplatform/warm_test.go
+++ b/pkg/controller/integrationplatform/warm_test.go
@@ -58,7 +58,7 @@ func TestWarm_Succeeded(t *testing.T) {
 
 	assert.Nil(t, platform.ConfigureDefaults(context.TODO(), c, &ip, false))
 
-	h := NewWarmAction()
+	h := NewWarmAction(c)
 	h.InjectLogger(log.Log)
 	h.InjectClient(c)
 
@@ -93,7 +93,7 @@ func TestWarm_Failing(t *testing.T) {
 
 	assert.Nil(t, platform.ConfigureDefaults(context.TODO(), c, &ip, false))
 
-	h := NewWarmAction()
+	h := NewWarmAction(c)
 	h.InjectLogger(log.Log)
 	h.InjectClient(c)
 
@@ -128,7 +128,7 @@ func TestWarm_WarmingUp(t *testing.T) {
 
 	assert.Nil(t, platform.ConfigureDefaults(context.TODO(), c, &ip, false))
 
-	h := NewWarmAction()
+	h := NewWarmAction(c)
 	h.InjectLogger(log.Log)
 	h.InjectClient(c)