feat: add workload-ref/generation to rollout (argoproj#1198)

- Add a "rollout.argoproj.io/workload-generation" annotation to the rollout metadata, which equals to the generation of reference workload
- workload-generation is updated when the referenced workload is updated.
- status.workloadObservedGeneration records the observed generation of the rollout

Signed-off-by: Hui Kang <[email protected]>

Author: huikang

controller/controller.go

@@ -146,7 +146,7 @@ func NewManager(
 	serviceWorkqueue := workqueue.NewNamedRateLimitingQueue(queue.DefaultArgoRolloutsRateLimiter(), "Services")
 	ingressWorkqueue := workqueue.NewNamedRateLimitingQueue(queue.DefaultArgoRolloutsRateLimiter(), "Ingresses")
 
-	refResolver := rollout.NewInformerBasedWorkloadRefResolver(namespace, dynamicclientset, discoveryClient, rolloutWorkqueue, rolloutsInformer.Informer())
+	refResolver := rollout.NewInformerBasedWorkloadRefResolver(namespace, dynamicclientset, discoveryClient, argoprojclientset, rolloutsInformer.Informer())
 	apiFactory := api.NewFactory(record.NewAPIFactorySettings(), defaults.Namespace(), secretInformer.Informer(), configMapInformer.Informer())
 	recorder := record.NewEventRecorder(kubeclientset, metrics.MetricRolloutEventsTotal, apiFactory)
 	notificationsController := controller.NewController(dynamicclientset.Resource(v1alpha1.RolloutGVR), rolloutsInformer.Informer(), apiFactory,

docs/migrating.md

@@ -108,6 +108,8 @@ After creation Rollout will spinup required number of Pods side-by-side with the
 Rollout won't try to manage existing Deployment Pods. That means you can safely update add Rollout
 to the production environment without any interruption but you are going to run twice more Pods during migration.
 
+Argo-rollouts controller patches the spec of rollout object with an annotation of `rollout.argoproj.io/workload-generation`, which equals the generation of referenced deployment. Users can detect if the rollout matches desired generation of deployment by checking the `workloadObservedGeneration` in the rollout status.
+
 **Traffic Management During Migration**
 
 The Rollout offers traffic management functionality that manages routing rules and flows the traffic to different

manifests/crds/rollout-crd.yaml

@@ -2820,6 +2820,8 @@ spec:
               updatedReplicas:
                 format: int32
                 type: integer
+              workloadObservedGeneration:
+                type: string
             type: object
         required:
         - spec

manifests/install.yaml

@@ -12509,6 +12509,8 @@ spec:
               updatedReplicas:
                 format: int32
                 type: integer
+              workloadObservedGeneration:
+                type: string
             type: object
         required:
         - spec

manifests/namespace-install.yaml

@@ -12509,6 +12509,8 @@ spec:
               updatedReplicas:
                 format: int32
                 type: integer
+              workloadObservedGeneration:
+                type: string
             type: object
         required:
         - spec

pkg/apiclient/rollout/rollout.swagger.json

@@ -1223,6 +1223,10 @@
           "type": "string",
           "title": "The generation observed by the rollout controller from metadata.generation\n+optional"
         },
+        "workloadObservedGeneration": {
+          "type": "string",
+          "title": "The generation of referenced workload observed by the rollout controller\n+optional"
+        },
         "conditions": {
           "type": "array",
           "items": {

pkg/apis/rollouts/v1alpha1/types.go

@@ -664,6 +664,9 @@ type RolloutStatus struct {
 	// The generation observed by the rollout controller from metadata.generation
 	// +optional
 	ObservedGeneration string `json:"observedGeneration,omitempty" protobuf:"bytes,13,opt,name=observedGeneration"`
+	// The generation of referenced workload observed by the rollout controller
+	// +optional
+	WorkloadObservedGeneration string `json:"workloadObservedGeneration,omitempty" protobuf:"bytes,24,opt,name=workloadObservedGeneration"`
 	// Conditions a list of conditions a rollout can have.
 	// +optional
 	Conditions []RolloutCondition `json:"conditions,omitempty" protobuf:"bytes,14,rep,name=conditions"`

rollout/sync.go

@@ -691,6 +691,16 @@ func (c *rolloutContext) persistRolloutStatus(newStatus *v1alpha1.RolloutStatus)
 
 	prevStatus := c.rollout.Status
 	c.pauseContext.CalculatePauseStatus(newStatus)
+	if c.rollout.Spec.TemplateResolvedFromRef {
+		workloadRefObservation, _ := annotations.GetWorkloadGenerationAnnotation(c.rollout)
+		currentWorkloadObservedGeneration, _ := strconv.ParseInt(newStatus.WorkloadObservedGeneration, 10, 32)
+		if workloadRefObservation != int32(currentWorkloadObservedGeneration) {
+			newStatus.WorkloadObservedGeneration = strconv.Itoa(int(workloadRefObservation))
+		}
+	} else {
+		newStatus.WorkloadObservedGeneration = ""
+	}
+
 	newStatus.ObservedGeneration = strconv.Itoa(int(c.rollout.Generation))
 	newStatus.Phase, newStatus.Message = rolloututil.CalculateRolloutPhase(c.rollout.Spec, *newStatus)
 

rollout/sync_test.go

@@ -215,6 +215,54 @@ func TestReconcileRevisionHistoryLimit(t *testing.T) {
 	}
 }
 
+func TestPersistWorkloadRefGeneration(t *testing.T) {
+	replica := int32(1)
+	r := &v1alpha1.Rollout{
+		Spec: v1alpha1.RolloutSpec{
+			Replicas: &replica,
+		},
+	}
+	fake := fake.Clientset{}
+	roCtx := &rolloutContext{
+		rollout: r,
+		log:     logutil.WithRollout(r),
+		reconcilerBase: reconcilerBase{
+			argoprojclientset: &fake,
+		},
+		pauseContext: &pauseContext{
+			rollout: r,
+		},
+	}
+
+	tests := []struct {
+		annotatedRefGeneration string
+		currentObserved        string
+	}{
+		{"1", ""},
+		{"2", "1"},
+		{"", "1"},
+	}
+
+	for _, tc := range tests {
+		newStatus := &v1alpha1.RolloutStatus{
+			UpdatedReplicas:   int32(1),
+			AvailableReplicas: int32(1),
+		}
+
+		if tc.annotatedRefGeneration != "" {
+			annotations.SetRolloutWorkloadRefGeneration(r, tc.annotatedRefGeneration)
+			r.Spec.TemplateResolvedFromRef = true
+
+			newStatus.WorkloadObservedGeneration = tc.currentObserved
+		} else {
+			r.Spec.TemplateResolvedFromRef = false
+			annotations.RemoveRolloutWorkloadRefGeneration(r)
+		}
+		roCtx.persistRolloutStatus(newStatus)
+		assert.Equal(t, tc.annotatedRefGeneration, newStatus.WorkloadObservedGeneration)
+	}
+}
+
 // TestCanaryPromoteFull verifies skip pause, analysis, steps when promote full is set for a canary rollout
 func TestCanaryPromoteFull(t *testing.T) {
 	f := newFixture(t)

rollout/temlateref.go

@@ -4,12 +4,16 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strconv"
 	"sync"
 	"time"
 
 	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
+	clientset "github.com/argoproj/argo-rollouts/pkg/client/clientset/versioned"
+	"github.com/argoproj/argo-rollouts/utils/annotations"
 	unstructuredutil "github.com/argoproj/argo-rollouts/utils/unstructured"
 
+	log "github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -21,7 +25,6 @@ import (
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/util/workqueue"
 )
 
 const (
@@ -57,16 +60,16 @@ type informerBasedTemplateResolver struct {
 	discoClient            discovery.DiscoveryInterface
 	ctx                    context.Context
 	cancelContext          context.CancelFunc
-	rolloutWorkQueue       workqueue.Interface
 	rolloutsInformer       cache.SharedIndexInformer
+	argoprojclientset      clientset.Interface
 }
 
 // NewInformerBasedWorkloadRefResolver create new instance of workload ref resolver.
 func NewInformerBasedWorkloadRefResolver(
 	namespace string,
 	dynamicClient dynamic.Interface,
 	discoClient discovery.DiscoveryInterface,
-	rolloutWorkQueue workqueue.Interface,
+	agrgoProjClientset clientset.Interface,
 	rolloutsInformer cache.SharedIndexInformer,
 ) *informerBasedTemplateResolver {
 	ctx, cancelContext := context.WithCancel(context.TODO())
@@ -88,9 +91,9 @@ func NewInformerBasedWorkloadRefResolver(
 		cancelContext:          cancelContext,
 		informerResyncDuration: time.Minute * 5,
 		informerSyncTimeout:    time.Minute,
+		argoprojclientset:      agrgoProjClientset,
 		dynamicClient:          dynamicClient,
 		discoClient:            discoClient,
-		rolloutWorkQueue:       rolloutWorkQueue,
 		rolloutsInformer:       rolloutsInformer,
 	}
 }
@@ -122,6 +125,7 @@ func remarshalMap(objMap map[string]interface{}, res interface{}) error {
 // Resolve verifies if given rollout has template reference and resolves pod template
 func (r *informerBasedTemplateResolver) Resolve(rollout *v1alpha1.Rollout) error {
 	if rollout.Spec.WorkloadRef == nil {
+		annotations.RemoveRolloutWorkloadRefGeneration(rollout)
 		return nil
 	}
 
@@ -170,6 +174,14 @@ func (r *informerBasedTemplateResolver) Resolve(rollout *v1alpha1.Rollout) error
 		}
 	}
 
+	// initialize rollout workload-generation annotation
+	workloadMeta, err := meta.Accessor(obj)
+	if err != nil {
+		return err
+	}
+	generation := strconv.FormatInt(workloadMeta.GetGeneration(), 10)
+	annotations.SetRolloutWorkloadRefGeneration(rollout, generation)
+
 	return nil
 }
 
@@ -198,36 +210,60 @@ func (r *informerBasedTemplateResolver) newInformerForGVK(gvk schema.GroupVersio
 		nil)
 	informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
-			r.requeueReferencedRollouts(obj, gvk)
+			r.updateRolloutsReferenceAnnotation(obj, gvk)
 		},
 		UpdateFunc: func(oldObj, newObj interface{}) {
-			r.requeueReferencedRollouts(newObj, gvk)
+			r.updateRolloutsReferenceAnnotation(newObj, gvk)
 		},
 		DeleteFunc: func(obj interface{}) {
-			r.requeueReferencedRollouts(obj, gvk)
+			r.updateRolloutsReferenceAnnotation(obj, gvk)
 		},
 	})
 	return informer, nil
 
 }
 
-// requeueReferencedRollouts re-queues all rollouts referenced by given object
-func (r *informerBasedTemplateResolver) requeueReferencedRollouts(obj interface{}, gvk schema.GroupVersionKind) {
-	roMeta, err := meta.Accessor(obj)
+// updateRolloutsReferenceAnnotation update the annotation of all rollouts referenced by given object
+func (r *informerBasedTemplateResolver) updateRolloutsReferenceAnnotation(obj interface{}, gvk schema.GroupVersionKind) {
+	workloadMeta, err := meta.Accessor(obj)
 	if err != nil {
 		return
 	}
+
 	rollouts, err := r.rolloutsInformer.GetIndexer().ByIndex(templateRefIndexName, refKey(v1alpha1.ObjectRef{
 		Kind:       gvk.Kind,
 		APIVersion: gvk.GroupVersion().String(),
-		Name:       roMeta.GetName(),
-	}, roMeta.GetNamespace()))
+		Name:       workloadMeta.GetName(),
+	}, workloadMeta.GetNamespace()))
 	if err != nil {
 		return
 	}
+
+	var updateAnnotation func(ro *v1alpha1.Rollout)
+
+	generation := strconv.FormatInt(workloadMeta.GetGeneration(), 10)
+	updateAnnotation = func(ro *v1alpha1.Rollout) {
+		updated := annotations.SetRolloutWorkloadRefGeneration(ro, generation)
+		if updated {
+			// update the annotation causes the rollout to be requeued and the template will be resolved to the referred
+			// workload during next reconciliation
+			ro.Spec.Template.Spec.Containers = []corev1.Container{}
+			_, err := r.argoprojclientset.ArgoprojV1alpha1().Rollouts(ro.Namespace).Update(context.TODO(), ro, v1.UpdateOptions{})
+			if err != nil {
+				log.Errorf("Cannot update the workload-ref/annotation for %s/%s", ro.GetName(), ro.GetNamespace())
+			}
+		}
+	}
 	for _, ro := range rollouts {
-		if key, err := cache.MetaNamespaceKeyFunc(ro); err == nil {
-			r.rolloutWorkQueue.Add(key)
+		rollout, ok := ro.(*v1alpha1.Rollout)
+		if ok {
+			updateAnnotation(rollout)
+		} else {
+			un, ok := ro.(*unstructured.Unstructured)
+			if ok {
+				rollout := unstructuredutil.ObjectToRollout(un)
+				updateAnnotation(rollout)
+			}
 		}
 	}
 }