Fix #703: fix warming cache and affinity to work in global mode

deploy/operator-role-kubernetes.yaml

@@ -20,6 +20,9 @@ rules:
   - persistentvolumeclaims
   - configmaps
   - secrets
+  - serviceaccounts
+  - roles
+  - rolebindings
   verbs:
   - create
   - delete

deploy/operator-role-openshift.yaml

@@ -20,6 +20,9 @@ rules:
   - persistentvolumeclaims
   - configmaps
   - secrets
+  - serviceaccounts
+  - roles
+  - rolebindings
   verbs:
   - create
   - delete

pkg/builder/kaniko/publisher.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/apache/camel-k/pkg/builder"
+	"github.com/apache/camel-k/pkg/platform"
 	"github.com/apache/camel-k/pkg/util/kubernetes"
 	"github.com/apache/camel-k/pkg/util/tar"
 
@@ -131,19 +132,32 @@ func publisher(ctx *builder.Context) error {
 			},
 			RestartPolicy: corev1.RestartPolicyNever,
 			Volumes:       volumes,
-			// Co-locate with builder pod for sharing the volume
-			Affinity: &corev1.Affinity{
-				PodAffinity: &corev1.PodAffinity{
-					RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-						{
-							LabelSelector: &metav1.LabelSelector{
-								MatchLabels: map[string]string{
-									"camel.apache.org/component": "operator",
-								},
-							},
-							TopologyKey: "kubernetes.io/hostname",
+		},
+	}
+
+
+	var labelKey string
+	var labelValue string
+	if ctx.Namespace == platform.GetOperatorNamespace() {
+		// Check if the operator is running in the same namespace
+		labelKey = "camel.apache.org/component"
+		labelValue = "operator"
+	} else {
+		labelKey = "camel.apache.org/build"
+		labelValue = ctx.Build.Meta.Name
+	}
+
+	// Co-locate with builder pod for sharing the volume
+	pod.Spec.Affinity = &corev1.Affinity{
+		PodAffinity: &corev1.PodAffinity{
+			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+				{
+					LabelSelector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{
+							labelKey: labelValue,
 						},
 					},
+					TopologyKey: "kubernetes.io/hostname",
 				},
 			},
 		},

pkg/controller/build/schedule_pod.go

@@ -156,6 +156,9 @@ func newBuildPod(build *v1alpha1.Build, operatorImage string) *corev1.Pod {
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: build.Namespace,
 			Name:      buildPodName(build.Spec.Meta),
+			Labels: map[string]string{
+				"camel.apache.org/build": build.Name,
+			},
 		},
 		Spec: corev1.PodSpec{
 			ServiceAccountName: "camel-k-operator",
@@ -193,22 +196,26 @@ func newBuildPod(build *v1alpha1.Build, operatorImage string) *corev1.Pod {
 				MountPath: build.Spec.BuildDir,
 			},
 		}
-		// Co-locate with the builder pod for sharing the host path volume as the current
-		// persistent volume claim uses the default storage class which is likely relying
-		// on the host path provisioner.
-		pod.Spec.Affinity = &corev1.Affinity{
-			PodAffinity: &corev1.PodAffinity{
-				RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-					{
-						LabelSelector: &metav1.LabelSelector{
-							MatchLabels: map[string]string{
-								"camel.apache.org/component": "operator",
+
+		// Use affinity only when the operator is present in the namespaced
+		if build.Namespace == platform.GetOperatorNamespace() {
+			// Co-locate with the builder pod for sharing the host path volume as the current
+			// persistent volume claim uses the default storage class which is likely relying
+			// on the host path provisioner.
+			pod.Spec.Affinity = &corev1.Affinity{
+				PodAffinity: &corev1.PodAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+						{
+							LabelSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{
+									"camel.apache.org/component": "operator",
+								},
 							},
+							TopologyKey: "kubernetes.io/hostname",
 						},
-						TopologyKey: "kubernetes.io/hostname",
 					},
 				},
-			},
+			}
 		}
 	}
 

pkg/controller/integrationplatform/initialize.go

@@ -161,13 +161,21 @@ func (action *initializeAction) Handle(ctx context.Context, ip *v1alpha1.Integra
 			return err
 		}
 
-		// Create the Kaniko warmer pod that caches the base image into the Camel K builder volume
-		action.L.Info("Create Kaniko cache warmer pod")
-		err = createKanikoCacheWarmerPod(ctx, action.client, target)
-		if err != nil {
-			return err
+		// Check if the operator is running in the same namespace before starting the cache warmer
+		if target.Namespace == platform.GetOperatorNamespace() {
+			// Create the Kaniko warmer pod that caches the base image into the Camel K builder volume
+			action.L.Info("Create Kaniko cache warmer pod")
+			err = createKanikoCacheWarmerPod(ctx, action.client, target)
+			if err != nil {
+				return err
+			}
+
+			target.Status.Phase = v1alpha1.IntegrationPlatformPhaseWarming
+		} else {
+			// Skip the warmer pod creation
+			target.Status.Phase = v1alpha1.IntegrationPlatformPhaseCreating
 		}
-		target.Status.Phase = v1alpha1.IntegrationPlatformPhaseWarming
+
 	} else {
 		target.Status.Phase = v1alpha1.IntegrationPlatformPhaseCreating
 	}

pkg/platform/operator.go

@@ -24,6 +24,7 @@ import (
 	"strings"
 
 	v1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -33,13 +34,9 @@ const operatorPodNameEnvVariable = "POD_NAME"
 
 // 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 client.Client) (string, error) {
-	var podNamespace string
-	var podName string
-	var envSet bool
-	if podNamespace, envSet = os.LookupEnv(operatorNamespaceEnvVariable); !envSet || podNamespace == "" {
-		return "", nil
-	}
-	if podName, envSet = os.LookupEnv(operatorPodNameEnvVariable); !envSet || podName == "" {
+	podNamespace := GetOperatorNamespace()
+	podName := GetOperatorPodName()
+	if podNamespace == "" || podName == "" {
 		return "", nil
 	}
 
@@ -49,7 +46,9 @@ func GetCurrentOperatorImage(ctx context.Context, c client.Client) (string, erro
 	}
 	pod := v1.Pod{}
 
-	if err := c.Get(ctx, podKey, &pod); err != nil {
+	if err := c.Get(ctx, podKey, &pod); err != nil && k8serrors.IsNotFound(err) {
+		return "", nil
+	} else if err != nil {
 		return "", err
 	}
 	if len(pod.Spec.Containers) == 0 {
@@ -65,3 +64,19 @@ func IsCurrentOperatorGlobal() bool {
 	}
 	return false
 }
+
+// GetOperatorNamespace returns the namespace where the current operator is located (if set)
+func GetOperatorNamespace() string {
+	if podNamespace, envSet := os.LookupEnv(operatorNamespaceEnvVariable); envSet {
+		return podNamespace
+	}
+	return ""
+}
+
+// GetOperatorPodName returns the pod that is running the current operator (if any)
+func GetOperatorPodName() string {
+	if podName, envSet := os.LookupEnv(operatorPodNameEnvVariable); envSet {
+		return podName
+	}
+	return ""
+}