Repository: apache/camel-k
Pull request #2662 (2021-09-28 13:42:16)

feat: Consistent Integration scale status
This PR provides a more robust Integration scaling status, by directly reconciling Integration Pods, rather than relying on owned resources, like Deployment, KnativeService and CronJob, that do not provide a minimal common denominator. 

It fixes several issues, for which the reported number of replicas was incorrect, when the wrong ReplicaSet was selected for example.

It also restricts the caching of Pods by Integration label selector, in order to limit memory requirement for the operator, when scaling the number of Integrations.

Last but not least, It prepares better error handling, for which proper replicas reporting is a pre-requisite.

**Release Note**
```release-note
feat: Consistent Integration scale status
```


astefanutti: I think this is ready.

So besides fixing the scale status issues, it also removes the watches for Pods, that could lead to receiving all the updates for all the Pods on the entire cluster, except but for the Integration Pods, which should improve scalability by multiple orders of magnitude. This is strictly enforced by the label selector sets on the operator manager cache, so care must be taken to use a non-cached client when getting non Integration Pods, otherwise `nil` or an empty list is returned.

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/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)