fix: Wait for full deployment rollout before marking IntegrationSink Ready (#8858)

Previously, IntegrationSink would mark DeploymentReady=True when any
replica was available, even if old pods with stale AUTH_POLICIES were
still running during a rollout. This caused race conditions where:

1. Tests would send requests hitting old pods with outdated policies,
   resulting in 403 errors and CI flakes
2. Production clients could see Ready=True but hit pods without the
   latest EventPolicy changes

Now PropagateDeploymentStatus checks that:
- ObservedGeneration == Generation (controller processed latest spec)
- UpdatedReplicas == Replicas (all pods updated to new template)
- AvailableReplicas == Replicas (all updated pods are ready)

This ensures IntegrationSink only becomes Ready when all pods have the
current AUTH_POLICIES configuration, fixing both test flakes and the
production correctness issue.

Author: creydr

pkg/apis/sinks/v1alpha1/integration_sink_lifecycle.go

@@ -106,22 +106,51 @@ func (s *IntegrationSinkStatus) MarkEventPoliciesTrueWithReason(reason, messageF
 	IntegrationSinkCondSet.Manage(s).MarkTrueWithReason(IntegrationSinkConditionEventPoliciesReady, reason, messageFormat, messageA...)
 }
 
-func (s *IntegrationSinkStatus) PropagateDeploymentStatus(d *appsv1.DeploymentStatus) {
+func (s *IntegrationSinkStatus) PropagateDeploymentStatus(d *appsv1.Deployment) {
+	// A deployment is fully rolled out when:
+	// 1. ObservedGeneration == Generation: controller has observed the latest spec
+	// 2. UpdatedReplicas == Replicas: all pods updated to the new pod template
+	// 3. AvailableReplicas == Replicas: all updated pods are ready
+	// This ensures EventPolicy changes have propagated to all pods before marking Ready.
+	desiredReplicas := int32(1)
+	if d.Spec.Replicas != nil {
+		desiredReplicas = *d.Spec.Replicas
+	}
+
+	deploymentFullyRolledOut := d.Status.ObservedGeneration == d.Generation &&
+		d.Status.UpdatedReplicas == desiredReplicas &&
+		d.Status.AvailableReplicas == desiredReplicas
+
+	if deploymentFullyRolledOut {
+		IntegrationSinkCondSet.Manage(s).MarkTrue(IntegrationSinkConditionDeploymentReady)
+		return
+	}
+
+	// Check deployment conditions for error states
 	deploymentAvailableFound := false
-	for _, cond := range d.Conditions {
+	for _, cond := range d.Status.Conditions {
 		if cond.Type == appsv1.DeploymentAvailable {
 			deploymentAvailableFound = true
-			if cond.Status == corev1.ConditionTrue {
-				IntegrationSinkCondSet.Manage(s).MarkTrue(IntegrationSinkConditionDeploymentReady)
-			} else if cond.Status == corev1.ConditionFalse {
+			if cond.Status == corev1.ConditionFalse {
 				IntegrationSinkCondSet.Manage(s).MarkFalse(IntegrationSinkConditionDeploymentReady, cond.Reason, cond.Message)
-			} else if cond.Status == corev1.ConditionUnknown {
-				IntegrationSinkCondSet.Manage(s).MarkUnknown(IntegrationSinkConditionDeploymentReady, cond.Reason, cond.Message)
+				return
 			}
 		}
+		// Also check Progressing condition for failures (e.g., ImagePullBackOff, insufficient quota)
+		if cond.Type == appsv1.DeploymentProgressing && cond.Status == corev1.ConditionFalse {
+			IntegrationSinkCondSet.Manage(s).MarkFalse(IntegrationSinkConditionDeploymentReady, cond.Reason, cond.Message)
+			return
+		}
 	}
-	if !deploymentAvailableFound {
-		IntegrationSinkCondSet.Manage(s).MarkUnknown(IntegrationSinkConditionDeploymentReady, "DeploymentUnavailable", "The Deployment '%s' is unavailable.", d)
+
+	// Deployment is progressing but not fully rolled out yet
+	if deploymentAvailableFound {
+		IntegrationSinkCondSet.Manage(s).MarkUnknown(IntegrationSinkConditionDeploymentReady,
+			"DeploymentRollingOut",
+			"Deployment rollout in progress: %d/%d replicas updated and available",
+			d.Status.AvailableReplicas, desiredReplicas)
+	} else {
+		IntegrationSinkCondSet.Manage(s).MarkUnknown(IntegrationSinkConditionDeploymentReady, "DeploymentUnavailable", "The Deployment is unavailable")
 	}
 }
 

pkg/reconciler/integration/sink/integrationsink.go

@@ -177,7 +177,7 @@ func (r *Reconciler) reconcileDeployment(ctx context.Context, sink *sinks.Integr
 		logging.FromContext(ctx).Debugw("Reusing existing Deployment", zap.Any("Deployment", deployment))
 	}
 
-	sink.Status.PropagateDeploymentStatus(&deployment.Status)
+	sink.Status.PropagateDeploymentStatus(deployment)
 	return deployment, nil
 }
 

pkg/reconciler/integration/sink/integrationsink_test.go

@@ -193,24 +193,35 @@ func makeDeployment(sink *sinksv1alpha1.IntegrationSink, ready *corev1.Condition
 		}
 		if *ready == corev1.ConditionTrue {
 			status.ReadyReplicas = 1
+			status.UpdatedReplicas = 1
+			status.AvailableReplicas = 1
+			status.Replicas = 1
+			status.ObservedGeneration = 1
 		}
 	}
 
+	objectMeta := metav1.ObjectMeta{
+		Name:      deploymentName,
+		Namespace: sink.Namespace,
+		OwnerReferences: []metav1.OwnerReference{
+			*kmeta.NewControllerRef(sink),
+		},
+		Labels: integration.Labels(sink.Name),
+	}
+	// Simulate what Kubernetes API server would set for existing deployments
+	if ready != nil {
+		objectMeta.Generation = 1
+	}
+
 	d := &appsv1.Deployment{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: "apps/v1",
 			Kind:       "Deployment",
 		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      deploymentName,
-			Namespace: sink.Namespace,
-			OwnerReferences: []metav1.OwnerReference{
-				*kmeta.NewControllerRef(sink),
-			},
-			Labels: integration.Labels(sink.Name),
-		},
-		Status: status,
+		ObjectMeta: objectMeta,
+		Status:     status,
 		Spec: appsv1.DeploymentSpec{
+			Replicas: ptr.To(int32(1)),
 			Selector: &metav1.LabelSelector{
 				MatchLabels: integration.Labels(sink.Name),
 			},
@@ -376,12 +387,23 @@ func makeIntegrationSinkSpec() sinksv1alpha1.IntegrationSinkSpec {
 	}
 }
 
-func makeDeploymentStatus(ready *corev1.ConditionStatus) *appsv1.DeploymentStatus {
-	return &appsv1.DeploymentStatus{
-		Conditions: []appsv1.DeploymentCondition{{
-			Type:   appsv1.DeploymentAvailable,
-			Status: *ready,
-		}},
-		Replicas: 1,
+func makeDeploymentStatus(ready *corev1.ConditionStatus) *appsv1.Deployment {
+	return &appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Generation: 1,
+		},
+		Spec: appsv1.DeploymentSpec{
+			Replicas: ptr.To(int32(1)),
+		},
+		Status: appsv1.DeploymentStatus{
+			Conditions: []appsv1.DeploymentCondition{{
+				Type:   appsv1.DeploymentAvailable,
+				Status: *ready,
+			}},
+			Replicas:           1,
+			UpdatedReplicas:    1,
+			AvailableReplicas:  1,
+			ObservedGeneration: 1,
+		},
 	}
 }

pkg/reconciler/integration/sink/resources/container_image.go

@@ -62,6 +62,7 @@ func MakeDeploymentSpec(sink *v1alpha1.IntegrationSink, authProxyImage string, f
 			Labels: integration.Labels(sink.Name),
 		},
 		Spec: appsv1.DeploymentSpec{
+			Replicas: ptr.To(int32(1)),
 			Selector: &metav1.LabelSelector{
 				MatchLabels: integration.Labels(sink.Name),
 			},

pkg/reconciler/integration/sink/resources/container_image_test.go

@@ -88,6 +88,7 @@ func TestNewSQSContainerSink(t *testing.T) {
 			Labels: integration.Labels(testName),
 		},
 		Spec: appsv1.DeploymentSpec{
+			Replicas: ptr.To(int32(1)),
 			Selector: &metav1.LabelSelector{
 				MatchLabels: integration.Labels(testName),
 			},

pkg/reconciler/testing/v1alpha1/integrationsink.go

@@ -57,9 +57,9 @@ func WithInitIntegrationSinkConditions(s *v1alpha1.IntegrationSink) {
 	s.Status.InitializeConditions()
 }
 
-func WithIntegrationSinkPropagateDeploymenteStatus(status *appsv1.DeploymentStatus) IntegrationSinkOption {
+func WithIntegrationSinkPropagateDeploymenteStatus(deployment *appsv1.Deployment) IntegrationSinkOption {
 	return func(s *v1alpha1.IntegrationSink) {
-		s.Status.PropagateDeploymentStatus(status)
+		s.Status.PropagateDeploymentStatus(deployment)
 	}
 }