fix: recongnize correct traffic lane

go.mod

@@ -23,7 +23,7 @@ require (
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/kubernetes v1.22.2
 	k8s.io/utils v0.0.0-20241210054802-24370beab758
-	kusionstack.io/kube-api v0.7.3
+	kusionstack.io/kube-api v0.7.4-0.20250727122744-2399b387a919
 	kusionstack.io/kube-utils v0.2.1-0.20250613035327-11e9cdaec9d6
 	kusionstack.io/resourceconsist v0.0.2
 	sigs.k8s.io/controller-runtime v0.21.0

go.sum

@@ -1037,6 +1037,8 @@ kusionstack.io/kube-api v0.7.2 h1:qVrOnvGaqi4iZn6BZP0dQd/buLKLyDvY4PjkUCLVoZY=
 kusionstack.io/kube-api v0.7.2/go.mod h1:e1jtrQH2LK5fD2nTyfIXG6nYrYbU8VXShRxTRwVPaLk=
 kusionstack.io/kube-api v0.7.3 h1:Fxj/E+aixUuCB0VekEvEx6ss24Jke7MvcbRwAXwVYBo=
 kusionstack.io/kube-api v0.7.3/go.mod h1:e1jtrQH2LK5fD2nTyfIXG6nYrYbU8VXShRxTRwVPaLk=
+kusionstack.io/kube-api v0.7.4-0.20250727122744-2399b387a919 h1:EMcsFMNMZO3oW7pqGOZd1nJC6YyQSt2RrcGdpC1J1gU=
+kusionstack.io/kube-api v0.7.4-0.20250727122744-2399b387a919/go.mod h1:e1jtrQH2LK5fD2nTyfIXG6nYrYbU8VXShRxTRwVPaLk=
 kusionstack.io/kube-utils v0.2.1-0.20250613035327-11e9cdaec9d6 h1:HYE6Wa8EzSlA6UmaTLtNKUgkB2mmasp6Ul69d3/SpK0=
 kusionstack.io/kube-utils v0.2.1-0.20250613035327-11e9cdaec9d6/go.mod h1:5Uy3GCJ1JEGqZw/Sp/uVnHBJN1t9wjY6USPSZ9s4idk=
 kusionstack.io/resourceconsist v0.0.2 h1:gf+c/LOMsiKoVR+GLzOomw8qcUbZbPckQLczZllNdVM=

pkg/controllers/backendrouting/backendrouting_controller.go

@@ -260,7 +260,7 @@ func (b *BackendRoutingReconciler) syncInClusterBackends(ctx context.Context, sy
 	// ensure canary and stable backends
 	canaryConfig := obj.Spec.ForkedBackends.Canary.DeepCopy()
 	canaryConfig.ExtraLabelSelector = lo.Assign(canaryConfig.ExtraLabelSelector, map[string]string{
-		rolloutapi.LabelTrafficLane: rolloutapi.LabelValueTrafficLaneCanary,
+		rolloutapi.TrafficLaneLabelKey: rolloutapi.CanaryTrafficLane,
 	})
 	err := b.ensureBackendResource(ctx, syncCtx, *canaryConfig)
 	if err != nil {
@@ -271,7 +271,7 @@ func (b *BackendRoutingReconciler) syncInClusterBackends(ctx context.Context, sy
 
 	stableConfig := obj.Spec.ForkedBackends.Stable.DeepCopy()
 	stableConfig.ExtraLabelSelector = lo.Assign(stableConfig.ExtraLabelSelector, map[string]string{
-		rolloutapi.LabelTrafficLane: rolloutapi.LabelValueTrafficLaneStable,
+		rolloutapi.TrafficLaneLabelKey: rolloutapi.StableTrafficLane,
 	})
 	err = b.ensureBackendResource(ctx, syncCtx, *stableConfig)
 	if err != nil {
@@ -350,7 +350,7 @@ func (b *BackendRoutingReconciler) ensureBackendResource(ctx context.Context, sy
 		newBackendObj.GetLabels(),
 		config.ExtraLabelSelector,
 		map[string]string{
-			rolloutapi.LabelCanaryResource: "true",
+			rolloutapi.CanaryResourceLabelKey: "true",
 		},
 	))
 	// TODO: add canary-release finalizer to protect resource from deletion

pkg/controllers/backendrouting/backendrouting_controller_test.go

@@ -415,10 +415,10 @@ func (s *BackendRoutingControllerTestSuite) Test_ForBackends() {
 			return false
 		}
 
-		s.Require().Equal(rolloutapi.LabelValueTrafficLaneStable, stableSVC.Spec.Selector[rolloutapi.LabelTrafficLane])
-		s.Require().Contains(stableSVC.Labels, rolloutapi.LabelCanaryResource)
-		s.Require().Equal(rolloutapi.LabelValueTrafficLaneCanary, canarySVC.Spec.Selector[rolloutapi.LabelTrafficLane])
-		s.Require().Contains(canarySVC.Labels, rolloutapi.LabelCanaryResource)
+		s.Require().Equal(rolloutapi.StableTrafficLane, stableSVC.Spec.Selector[rolloutapi.TrafficLaneLabelKey])
+		s.Require().Contains(stableSVC.Labels, rolloutapi.CanaryResourceLabelKey)
+		s.Require().Equal(rolloutapi.CanaryTrafficLane, canarySVC.Spec.Selector[rolloutapi.TrafficLaneLabelKey])
+		s.Require().Contains(canarySVC.Labels, rolloutapi.CanaryResourceLabelKey)
 
 		return true
 	}, 60*time.Second, 5*time.Second, "stable and canary should be created")
@@ -577,7 +577,7 @@ func (s *BackendRoutingControllerTestSuite) Test_Route() {
 		if err != nil {
 			return false
 		}
-		s.Require().Contains(ingress.Labels, rolloutapi.LabelCanaryResource)
+		s.Require().Contains(ingress.Labels, rolloutapi.CanaryResourceLabelKey)
 		return true
 	}, 60*time.Second, 5*time.Second, "canary ingress should be ready")
 

pkg/controllers/podcanarylabel/podcanarylabel.go

@@ -18,6 +18,7 @@ package podcanarylabel
 
 import (
 	"context"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	rolloutapi "kusionstack.io/kube-api/rollout"
@@ -111,7 +112,7 @@ func (r *PodCanaryReconciler) Reconcile(ctx context.Context, req reconcile.Reque
 		// this workload is not controlled by rollout, we need to make sure pod revision label is not added
 		updated, err := utils.UpdateOnConflict(ctx, r.Client, r.Client, pod, func() error {
 			utils.MutateLabels(pod, func(labels map[string]string) {
-				delete(labels, rolloutapi.LabelTrafficLane)
+				delete(labels, rolloutapi.TrafficLaneLabelKey)
 			})
 			return nil
 		})
@@ -127,37 +128,27 @@ func (r *PodCanaryReconciler) Reconcile(ctx context.Context, req reconcile.Reque
 		return reconcile.Result{}, nil
 	}
 
-	trafficLane := recognizeTrafficLane(pc, r.Client, workloadObj.Object, pod)
+	trafficLane := workload.RecognizeTrafficLane(ctx, workloadObj.Accessor, pc, r.Client, workloadObj.Object, pod)
 
 	// patch pod label
 	updated, err := utils.UpdateOnConflict(ctx, r.Client, r.Client, pod, func() error {
 		utils.MutateLabels(pod, func(labels map[string]string) {
-			labels[rolloutapi.LabelTrafficLane] = trafficLane
+			labels[rolloutapi.TrafficLaneLabelKey] = trafficLane
 		})
 		return nil
 	})
+	if err != nil {
+		return reconcile.Result{}, err
+	}
 
 	if updated {
 		logger.V(2).Info("updated pod traffic lane label value", "traffic-lane", trafficLane)
 	}
 
-	return reconcile.Result{}, err
-}
-
-func recognizeTrafficLane(pc workload.ReplicaObjectControl, reader client.Reader, workloadObj client.Object, pod *corev1.Pod) string {
-	if workload.IsCanary(workloadObj) {
-		// canary workload, always set pod revision to canary
-		return rolloutapi.LabelValueTrafficLaneCanary
-	}
-
-	if !workload.IsProgressing(workloadObj) {
-		// workload is not progressing, set pod revision to base
-		return rolloutapi.LabelValueTrafficLaneStable
+	if trafficLane == rolloutapi.UnknownTrafficLane {
+		// unknown traffic lane, requeue after 5 seconds
+		return reconcile.Result{RequeueAfter: 5 * time.Second}, nil
 	}
 
-	// workload is progressing, set updated pod revision to canary
-	if updated, _ := pc.IsUpdateObject(context.TODO(), reader, workloadObj, pod); updated {
-		return rolloutapi.LabelValueTrafficLaneCanary
-	}
-	return rolloutapi.LabelValueTrafficLaneStable
+	return reconcile.Result{}, err
 }

pkg/controllers/rolloutrun/control/control.go

@@ -272,7 +272,7 @@ func (c *CanaryReleaseControl) canaryObject(stable *workload.Info) (client.Objec
 func (c *CanaryReleaseControl) applyCanaryDefaults(canaryObj client.Object) {
 	controllerutil.AddFinalizer(canaryObj, rolloutapi.FinalizerCanaryResourceProtection)
 	utils.MutateLabels(canaryObj, func(labels map[string]string) {
-		labels[rolloutapi.LabelCanaryResource] = "true"
+		labels[rolloutapi.CanaryResourceLabelKey] = "true"
 	})
 }
 

pkg/controllers/rolloutrun/executor/canary.go

@@ -254,8 +254,8 @@ func appendBuiltinTemplateMetadataPatch(patch *rolloutv1alpha1.MetadataPatch) *r
 		patch.Labels = map[string]string{}
 	}
 
-	patch.Labels[rolloutapi.LabelCanaryResource] = "true"
-	patch.Labels[rolloutapi.LabelTrafficLane] = rolloutapi.LabelValueTrafficLaneCanary
+	patch.Labels[rolloutapi.CanaryResourceLabelKey] = "true"
+	patch.Labels[rolloutapi.TrafficLaneLabelKey] = rolloutapi.CanaryTrafficLane
 	return patch
 }
 

pkg/trafficrouting/route/ingress/control.go

@@ -150,7 +150,7 @@ func (c *ingressControl) AddCanary(ctx context.Context) error {
 		if canaryIgs.Labels == nil {
 			canaryIgs.Labels = make(map[string]string)
 		}
-		canaryIgs.Labels[rolloutapi.LabelCanaryResource] = "true"
+		canaryIgs.Labels[rolloutapi.CanaryResourceLabelKey] = "true"
 		return nil
 	})
 

pkg/workload/collaset/pod_control.go

@@ -27,7 +27,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	"kusionstack.io/rollout/pkg/utils"
 	"kusionstack.io/rollout/pkg/workload"
 )
 
@@ -37,23 +36,24 @@ func (c *accessorImpl) ReplicaType() schema.GroupVersionKind {
 	return corev1.SchemeGroupVersion.WithKind("Pod")
 }
 
-func (c *accessorImpl) IsUpdateObject(_ context.Context, _ client.Reader, workload, obj client.Object) (bool, error) {
+func (c *accessorImpl) RecognizeRevision(_ context.Context, _ client.Reader, workload, obj client.Object) (isCurrent, isUpdated bool, err error) {
 	cls, err := checkObj(workload)
 	if err != nil {
-		return false, err
+		return false, false, err
 	}
 	pod, ok := obj.(*corev1.Pod)
 	if !ok {
-		return false, fmt.Errorf("object must be Pod")
+		return false, false, fmt.Errorf("object must be Pod")
 	}
-	revision := utils.GetMapValueByDefault(pod.Labels, appsv1.ControllerRevisionHashLabelKey, cls.Status.CurrentRevision)
+	revision := pod.Labels[appsv1.ControllerRevisionHashLabelKey]
 	if revision == cls.Status.CurrentRevision {
-		return false, nil
+		isCurrent = true
 	}
-	if revision == cls.Status.UpdatedRevision {
-		return true, nil
+	if cls.Generation == cls.Status.ObservedGeneration &&
+		revision == cls.Status.UpdatedRevision {
+		isUpdated = true
 	}
-	return false, nil
+	return isCurrent, isUpdated, nil
 }
 
 func (c *accessorImpl) GetReplicObjects(ctx context.Context, reader client.Reader, workload client.Object) ([]client.Object, error) {

pkg/workload/collaset/pod_control_test.go

@@ -0,0 +1,170 @@
+/**
+ * Copyright 2025 The KusionStack Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package collaset
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kusionstackappsv1 "kusionstack.io/kube-api/apps/v1alpha1"
+)
+
+func TestRecognizeRevision(t *testing.T) {
+	// Test cases where workload is StatefulSet and obj is Pod
+	tests := []struct {
+		name        string
+		sts         *kusionstackappsv1.CollaSet
+		pod         *corev1.Pod // Input Pod
+		wantCurrent bool        // Expected isCurrent result
+		wantUpdated bool        // Expected isUpdated result
+		wantErr     bool        // Expected error
+	}{
+		{
+			name: "pod with current revision label",
+			sts: &kusionstackappsv1.CollaSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Generation: 1,
+				},
+				Status: kusionstackappsv1.CollaSetStatus{
+					CurrentRevision:    "current-rev",
+					UpdatedRevision:    "update-rev",
+					ObservedGeneration: 1,
+				},
+			},
+			pod: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						appsv1.ControllerRevisionHashLabelKey: "current-rev",
+					},
+				},
+			},
+			wantCurrent: true,
+			wantUpdated: false,
+			wantErr:     false,
+		},
+		{
+			name: "pod with update revision label and matching generation",
+			sts: &kusionstackappsv1.CollaSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Generation: 1,
+				},
+				Status: kusionstackappsv1.CollaSetStatus{
+					CurrentRevision:    "current-rev",
+					UpdatedRevision:    "update-rev",
+					ObservedGeneration: 1,
+				},
+			},
+			pod: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						appsv1.ControllerRevisionHashLabelKey: "update-rev",
+					},
+				},
+			},
+			wantCurrent: false,
+			wantUpdated: true,
+			wantErr:     false,
+		},
+		{
+			name: "pod with no revision label",
+			sts: &kusionstackappsv1.CollaSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Generation: 1,
+				},
+				Status: kusionstackappsv1.CollaSetStatus{
+					CurrentRevision:    "current-rev",
+					UpdatedRevision:    "update-rev",
+					ObservedGeneration: 1,
+				},
+			},
+			pod: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{},
+				},
+			},
+			wantCurrent: false,
+			wantUpdated: false,
+			wantErr:     false,
+		},
+		{
+			name: "pod with update revision label but generation mismatch",
+			sts: &kusionstackappsv1.CollaSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Generation: 2,
+				},
+				Status: kusionstackappsv1.CollaSetStatus{
+					CurrentRevision:    "current-rev",
+					UpdatedRevision:    "update-rev",
+					ObservedGeneration: 1,
+				},
+			},
+			pod: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						appsv1.ControllerRevisionHashLabelKey: "update-rev",
+					},
+				},
+			},
+			wantCurrent: false,
+			wantUpdated: false,
+			wantErr:     false,
+		},
+		{
+			name: "pod with unknown revision label",
+			sts: &kusionstackappsv1.CollaSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Generation: 1,
+				},
+				Status: kusionstackappsv1.CollaSetStatus{
+					CurrentRevision:    "current-rev",
+					UpdatedRevision:    "update-rev",
+					ObservedGeneration: 1,
+				},
+			},
+			pod: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						appsv1.ControllerRevisionHashLabelKey: "unknown-rev",
+					},
+				},
+			},
+			wantCurrent: false,
+			wantUpdated: false,
+			wantErr:     false,
+		},
+	}
+
+	a := &accessorImpl{}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			isCurrent, isUpdated, err := a.RecognizeRevision(context.Background(), nil, tt.sts, tt.pod)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+			assert.Equal(t, tt.wantCurrent, isCurrent)
+			assert.Equal(t, tt.wantUpdated, isUpdated)
+		})
+	}
+}

pkg/workload/interface.go

@@ -60,8 +60,8 @@ type CanaryReleaseControl interface {
 type ReplicaObjectControl interface {
 	// RepliceType returns the type of replica object
 	ReplicaType() schema.GroupVersionKind
-	// IsUpdateObject checks if the replica object revision is updated of the workload
-	IsUpdateObject(ctx context.Context, reader client.Reader, workload, object client.Object) (bool, error)
+	// RecognizeRevision checks if the replica object revision is crrent or updated of the workload
+	RecognizeRevision(ctx context.Context, reader client.Reader, workload, object client.Object) (current, updated bool, err error)
 	// GetReplicObjects gets the pod selector of the workload
 	GetReplicObjects(ctx context.Context, reader client.Reader, workload client.Object) ([]client.Object, error)
 }