fix: Fix double injection of trust-bundle CA volumes in reconcile loop

pkg/reconciler/containersource/containersource.go

@@ -28,7 +28,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	appsv1listers "k8s.io/client-go/listers/apps/v1"
-	corev1listers "k8s.io/client-go/listers/core/v1"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	pkgreconciler "knative.dev/pkg/reconciler"
@@ -37,7 +36,6 @@ import (
 	clientset "knative.dev/eventing/pkg/client/clientset/versioned"
 	"knative.dev/eventing/pkg/client/injection/reconciler/sources/v1/containersource"
 	listers "knative.dev/eventing/pkg/client/listers/sources/v1"
-	"knative.dev/eventing/pkg/eventingtls"
 	"knative.dev/eventing/pkg/reconciler/containersource/resources"
 )
 
@@ -62,10 +60,9 @@ type Reconciler struct {
 	eventingClientSet clientset.Interface
 
 	// listers index properties about resources
-	containerSourceLister      listers.ContainerSourceLister
-	sinkBindingLister          listers.SinkBindingLister
-	deploymentLister           appsv1listers.DeploymentLister
-	trustBundleConfigMapLister corev1listers.ConfigMapLister
+	containerSourceLister listers.ContainerSourceLister
+	sinkBindingLister     listers.SinkBindingLister
+	deploymentLister      appsv1listers.DeploymentLister
 }
 
 // Check that our Reconciler implements Interface
@@ -89,14 +86,10 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, source *v1.ContainerSour
 }
 
 func (r *Reconciler) reconcileReceiveAdapter(ctx context.Context, source *v1.ContainerSource) (*appsv1.Deployment, error) {
-	podTemplate, err := eventingtls.AddTrustBundleVolumes(r.trustBundleConfigMapLister, source, &source.Spec.Template.Spec)
-	if err != nil {
-		return nil, fmt.Errorf("failed to add trust bundle volumes: %w", err)
-	}
-
-	updatedSource := source.DeepCopy() // Avoid update Spec of the given object
-	updatedSource.Spec.Template.Spec = *podTemplate
-	expected := resources.MakeDeployment(updatedSource)
+	// Trust bundle volumes are injected by SinkBinding (which is always bound to this
+	// exact Deployment), not here - adding them again would race with SinkBinding's own
+	// independent injection and cause spurious podTemplateChanged overwrites.
+	expected := resources.MakeDeployment(source)
 
 	ra, err := r.deploymentLister.Deployments(expected.Namespace).Get(expected.Name)
 	if apierrors.IsNotFound(err) {

pkg/reconciler/containersource/containersource_test.go

@@ -40,6 +40,7 @@ import (
 
 	sourcesv1 "knative.dev/eventing/pkg/apis/sources/v1"
 	"knative.dev/eventing/pkg/client/injection/reconciler/sources/v1/containersource"
+	"knative.dev/eventing/pkg/eventingtls"
 	"knative.dev/eventing/pkg/reconciler/containersource/resources"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/client/injection/ducks/duck/v1/addressable"
@@ -274,18 +275,54 @@ func TestAllCases(t *testing.T) {
 				Eventf(corev1.EventTypeNormal, sourceReconciled, `ContainerSource reconciled: "%s/%s"`, testNS, sourceName),
 			},
 		},
+		{
+			Name: "does not wipe SinkBinding-injected deployment fields when trust bundle configmap appears",
+			Objects: []runtime.Object{
+				NewContainerSource(sourceName, testNS,
+					WithContainerSourceUID(sourceUID),
+					WithContainerSourceSpec(makeContainerSourceSpec(sinkDest)),
+					WithContainerSourceObjectMetaGeneration(generation),
+				),
+				makeSinkBinding(NewContainerSource(sourceName, testNS,
+					WithContainerSourceSpec(makeContainerSourceSpec(sinkDest)),
+					WithContainerSourceUID(sourceUID),
+				), &conditionTrue),
+				makeDeploymentWithSinkBindingInjectedEnv(NewContainerSource(sourceName, testNS,
+					WithContainerSourceSpec(makeContainerSourceSpec(sinkDest)),
+					WithContainerSourceUID(sourceUID),
+				), &conditionTrue),
+				makeTrustBundleConfigMap(),
+			},
+			Key: testNS + "/" + sourceName,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeNormal, sourceReconciled, `ContainerSource reconciled: "%s/%s"`, testNS, sourceName),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: NewContainerSource(sourceName, testNS,
+					WithContainerSourceUID(sourceUID),
+					WithContainerSourceSpec(makeContainerSourceSpec(sinkDest)),
+					WithContainerSourceObjectMetaGeneration(generation),
+					WithInitContainerSourceConditions,
+					WithContainerSourceStatusObservedGeneration(generation),
+					WithContainerSourcePropagateSinkbindingStatus(makeSinkBindingStatus(&conditionTrue)),
+					WithContainerSourcePropagateReceiveAdapterStatus(makeDeploymentWithSinkBindingInjectedEnv(NewContainerSource(sourceName, testNS,
+						WithContainerSourceSpec(makeContainerSourceSpec(sinkDest)),
+						WithContainerSourceUID(sourceUID),
+					), &conditionTrue)),
+				),
+			}},
+		},
 	}
 
 	logger := logtesting.TestLogger(t)
 	table.Test(t, MakeFactory(func(ctx context.Context, listers *Listers, cmw configmap.Watcher) controller.Reconciler {
 		ctx = addressable.WithDuck(ctx)
 		r := &Reconciler{
-			kubeClientSet:              fakekubeclient.Get(ctx),
-			eventingClientSet:          fakeeventingclient.Get(ctx),
-			containerSourceLister:      listers.GetContainerSourceLister(),
-			deploymentLister:           listers.GetDeploymentLister(),
-			sinkBindingLister:          listers.GetSinkBindingLister(),
-			trustBundleConfigMapLister: listers.GetConfigMapLister(),
+			kubeClientSet:         fakekubeclient.Get(ctx),
+			eventingClientSet:     fakeeventingclient.Get(ctx),
+			containerSourceLister: listers.GetContainerSourceLister(),
+			deploymentLister:      listers.GetDeploymentLister(),
+			sinkBindingLister:     listers.GetSinkBindingLister(),
 		}
 		return containersource.NewReconciler(ctx, logging.FromContext(ctx), fakeeventingclient.Get(ctx), listers.GetContainerSourceLister(), controller.GetEventRecorder(ctx), r)
 	},
@@ -372,6 +409,35 @@ func makeDeployment(source *sourcesv1.ContainerSource, available *corev1.Conditi
 	}
 }
 
+// makeDeploymentWithSinkBindingInjectedEnv simulates a Deployment already mutated by
+// SinkBinding's independent reconciler, which injects K_SINK (and friends) into every
+// container once bound - regardless of what ContainerSource's own reconcile computes.
+func makeDeploymentWithSinkBindingInjectedEnv(source *sourcesv1.ContainerSource, available *corev1.ConditionStatus) *appsv1.Deployment {
+	d := makeDeployment(source, available)
+	for i := range d.Spec.Template.Spec.Containers {
+		d.Spec.Template.Spec.Containers[i].Env = append(d.Spec.Template.Spec.Containers[i].Env, corev1.EnvVar{
+			Name:  "K_SINK",
+			Value: "http://" + sinkName + "." + testNS + ".svc.cluster.local",
+		})
+	}
+	return d
+}
+
+func makeTrustBundleConfigMap() *corev1.ConfigMap {
+	return &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-trust-bundle",
+			Namespace: testNS,
+			Labels: map[string]string{
+				eventingtls.TrustBundleLabelKey: eventingtls.TrustBundleLabelValue,
+			},
+		},
+		Data: map[string]string{
+			"ca.crt": "test-cert-data",
+		},
+	}
+}
+
 func getOwnerReferences() []metav1.OwnerReference {
 	return []metav1.OwnerReference{{
 		APIVersion:         sourcesv1.SchemeGroupVersion.String(),

pkg/reconciler/containersource/controller.go

@@ -19,12 +19,7 @@ package containersource
 import (
 	"context"
 
-	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
-	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/filtered"
-	"knative.dev/pkg/kmeta"
-	"knative.dev/pkg/system"
 
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	deploymentinformer "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
@@ -38,7 +33,6 @@ import (
 	containersourceinformer "knative.dev/eventing/pkg/client/injection/informers/sources/v1/containersource"
 	sinkbindinginformer "knative.dev/eventing/pkg/client/injection/informers/sources/v1/sinkbinding"
 	v1containersource "knative.dev/eventing/pkg/client/injection/reconciler/sources/v1/containersource"
-	"knative.dev/eventing/pkg/eventingtls"
 )
 
 // NewController creates a Reconciler for ContainerSource and returns the result of NewImpl.
@@ -52,7 +46,6 @@ func NewController(
 	containersourceInformer := containersourceinformer.Get(ctx)
 	sinkbindingInformer := sinkbindinginformer.Get(ctx)
 	deploymentInformer := deploymentinformer.Get(ctx)
-	trustBundleConfigMapInformer := configmapinformer.Get(ctx, eventingtls.TrustBundleLabelSelector)
 
 	var globalResync func(obj interface{})
 	featureStore := feature.NewStore(logging.FromContext(ctx).Named("feature-config-store"),
@@ -64,12 +57,11 @@ func NewController(
 	featureStore.WatchConfigs(cmw)
 
 	r := &Reconciler{
-		kubeClientSet:              kubeClient,
-		eventingClientSet:          eventingClient,
-		containerSourceLister:      containersourceInformer.Lister(),
-		deploymentLister:           deploymentInformer.Lister(),
-		sinkBindingLister:          sinkbindingInformer.Lister(),
-		trustBundleConfigMapLister: trustBundleConfigMapInformer.Lister(),
+		kubeClientSet:         kubeClient,
+		eventingClientSet:     eventingClient,
+		containerSourceLister: containersourceInformer.Lister(),
+		deploymentLister:      deploymentInformer.Lister(),
+		sinkBindingLister:     sinkbindingInformer.Lister(),
 	}
 	impl := v1containersource.NewImpl(ctx, r, func(impl *controller.Impl) controller.Options {
 		return controller.Options{ConfigStore: featureStore}
@@ -91,27 +83,5 @@ func NewController(
 		Handler:    controller.HandleAll(impl.EnqueueControllerOf),
 	})
 
-	trustBundleConfigMapInformer.Informer().AddEventHandler(controller.HandleAll(func(i interface{}) {
-		obj, err := kmeta.DeletionHandlingAccessor(i)
-		if err != nil {
-			return
-		}
-		if obj.GetNamespace() == system.Namespace() {
-			globalResync(i)
-			return
-		}
-
-		sources, err := containersourceInformer.Lister().ContainerSources(obj.GetNamespace()).List(labels.Everything())
-		if err != nil {
-			return
-		}
-		for _, src := range sources {
-			impl.EnqueueKey(types.NamespacedName{
-				Namespace: src.Namespace,
-				Name:      src.Name,
-			})
-		}
-	}))
-
 	return impl
 }