[release-1.16] Shift traffic when the minimum number of replicas are ready during progressive rollout (#314)

* Shift traffic when the minimum number of replicas are ready during progressive rollout

Fixes: #302

* Verify if the target replicas is nil

* Added the test cases

* Clean up the logic for the routeLister

* Check the final revision name

---------

Co-authored-by: Vincent Hou <[email protected]>

Author: knative-prow-robot

pkg/reconciler/service/resources/rolloutorchestrator.go

@@ -197,9 +197,9 @@ func GetInitialTargetRevision(service *servingv1.Service, config *servingv1.Conf
 }
 
 // consolidateTraffic consolidates traffic with the same revision name into one.
-func consolidateTraffic(traffic []servingv1.TrafficTarget) []servingv1.TrafficTarget {
+func consolidateTraffic(trafficTargets []servingv1.TrafficTarget) []servingv1.TrafficTarget {
 	trafficMap := map[string]*servingv1.TrafficTarget{}
-	for _, traffic := range traffic {
+	for _, traffic := range trafficTargets {
 		if trafficInMap, found := trafficMap[traffic.RevisionName]; !found {
 			trafficMap[traffic.RevisionName] = traffic.DeepCopy()
 		} else {

pkg/reconciler/service/service.go

@@ -162,7 +162,8 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, service *servingv1.Servi
 	}
 
 	// After the RolloutOrchestrator is created or updated, call the base reconciliation loop of the service.
-	err = c.baseReconciler.ReconcileKind(ctx, TransformService(service, rolloutOrchestrator, c.rolloutConfig))
+	err = c.baseReconciler.ReconcileKind(ctx, TransformService(service, rolloutOrchestrator, c.rolloutConfig, c.spaLister.StagePodAutoscalers(service.Namespace),
+		c.routeLister.Routes(service.Namespace)))
 	if err != nil {
 		return err
 	}
@@ -909,19 +910,76 @@ func calculateStageTargetRevisions(replicasMap map[string]int32, startRevisions
 	return stageRevisionTarget
 }
 
-func TransformService(service *servingv1.Service, ro *v1.RolloutOrchestrator, rc *RolloutConfig) *servingv1.Service {
+func TransformService(service *servingv1.Service, ro *v1.RolloutOrchestrator, rc *RolloutConfig,
+	spaLister listers.StagePodAutoscalerNamespaceLister, routeLister servinglisters.RouteNamespaceLister) *servingv1.Service {
 	// If Knative Service defines more than one traffic, this feature tentatively does not cover this case.
 	if len(service.Spec.Traffic) > 1 {
 		return service
 	}
 	service.Spec.RouteSpec = servingv1.RouteSpec{
-		Traffic: convertIntoTrafficTarget(service.GetName(), ro, rc),
+		Traffic: convertIntoTrafficTarget(service.GetName(), ro, rc, spaLister, routeLister),
 	}
 	return service
 }
 
-func convertIntoTrafficTarget(name string, ro *v1.RolloutOrchestrator, rc *RolloutConfig) []servingv1.TrafficTarget {
+func convertIntoTrafficTarget(name string, ro *v1.RolloutOrchestrator, rc *RolloutConfig,
+	spaLister listers.StagePodAutoscalerNamespaceLister, routeLister servinglisters.RouteNamespaceLister) []servingv1.TrafficTarget {
 	revisionTarget := ro.Spec.StageTargetRevisions
+	finalTargetRevs := ro.Spec.TargetRevisions
+	targetRevName := finalTargetRevs[0].RevisionName
+	if !ro.IsNotConvertToOneUpgrade() && rc.ProgressiveRolloutEnabled {
+		// The revisionTarget is set directly to ro.Spec.StageTargetRevisions, because this is not a
+		// one-to-one revision upgrade or the rollout feature is disabled. We do not cover this use case
+		// in the implementation. The traffic target is set directly configured in ro.Spec.StageTargetRevisions.
+		// However, if this is one-to-one or many-to-one rollout, we need to make sure the minimum number of
+		// replicas are met before shifting the traffic percentage over.
+		// Find out the name of the revision scaling up. Get the name of the spa for the revision scaling up.
+		// The name is the same as the revision name.
+		spaTargetRevName := targetRevName
+		targetNumberReplicas, minScale := finalTargetRevs[0].MinScale, finalTargetRevs[0].MinScale
+
+		// Find the target number of replicas for the current stage.
+		for _, revision := range ro.Spec.StageTargetRevisions {
+			if revision.RevisionName == spaTargetRevName && revision.TargetReplicas != nil {
+				targetNumberReplicas = revision.TargetReplicas
+			}
+		}
+		spa, err := spaLister.Get(spaTargetRevName)
+		// Check the number of replicas has reached the target number of replicas for the revision scaling up
+		if err == nil && targetNumberReplicas != nil && spa.Status.ActualScale != nil && minScale != nil &&
+			*spa.Status.ActualScale < *minScale && *spa.Status.ActualScale < *targetNumberReplicas {
+			// If we have issues getting the spa, or the number of the replicas has reached the target number of
+			// the revision to scale up, we set the revisionTarget to ro.Spec.StageTargetRevisions.
+
+			// However, if there is no issue getting yhe spa, and the actual number of replicas is less than
+			// the target number of replicas, we need to use ro.Status.StageRevisionStatus or route.Status.Traffic
+			// as the traffic information for the route.
+			if len(ro.Status.StageRevisionStatus) > 0 {
+				// If the ro has the StageRevisionStatus in the status, use it.
+				revisionTarget = ro.Status.StageRevisionStatus
+				for index := range revisionTarget {
+					if revisionTarget[index].RevisionName == spaTargetRevName {
+						continue
+					}
+					revisionTarget[index].LatestRevision = ptr.Bool(false)
+				}
+			} else {
+				// If the ro does not have the StageRevisionStatus in the status, use the existing one in route.
+				route, errRoute := routeLister.Get(ro.Name)
+				if errRoute == nil && len(route.Status.Traffic) > 0 {
+					traffics := route.Status.Traffic
+					for index := range traffics {
+						if traffics[index].RevisionName == spaTargetRevName {
+							continue
+						}
+						traffics[index].LatestRevision = ptr.Bool(false)
+					}
+					return traffics
+				}
+			}
+		}
+	}
+
 	trafficTarget := make([]servingv1.TrafficTarget, 0, len(revisionTarget))
 	for _, revision := range revisionTarget {
 		target := servingv1.TrafficTarget{}
@@ -948,6 +1006,8 @@ func convertIntoTrafficTarget(name string, ro *v1.RolloutOrchestrator, rc *Rollo
 		}
 
 		if revision.LatestRevision != nil && *revision.LatestRevision {
+			// No matter the RevisionName is available for the latest revision or not, we cannot
+			// set the RevisionName here. Only ConfigurationName is allowed.
 			if strings.TrimSpace(revision.ConfigurationName) != "" {
 				target.ConfigurationName = revision.ConfigurationName
 			} else {