add analysisRun to the rollout notification

Signed-off-by: asingh51 <[email protected]>

utils/annotations/annotations.go

@@ -2,6 +2,7 @@ package annotations
 
 import (
 	"fmt"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"strconv"
 	"strings"
 
@@ -52,17 +53,30 @@ func GetWorkloadGenerationAnnotation(ro *v1alpha1.Rollout) (int32, bool) {
 }
 
 // GetRevisionAnnotation returns revision of rollout
-func GetRevisionAnnotation(ro *v1alpha1.Rollout) (int32, bool) {
-	if ro == nil {
+func GetRevisionAnnotation(anyObj metav1.Object) (int32, bool) {
+
+	if anyObj == nil {
+		return 0, false
+	}
+	var obj interface{}
+
+	switch anyObj.(type) {
+	case *v1alpha1.Rollout:
+		obj = anyObj.(*v1alpha1.Rollout)
+	case *v1alpha1.AnalysisRun:
+		obj = anyObj.(*v1alpha1.AnalysisRun)
+	default:
 		return 0, false
 	}
-	annotationValue, ok := ro.Annotations[RevisionAnnotation]
+
+	annotationValue, ok := obj.(metav1.Object).GetAnnotations()[RevisionAnnotation]
 	if !ok {
-		return int32(0), false
+		return 0, false
 	}
+
 	intValue, err := strconv.ParseInt(annotationValue, 10, 32)
 	if err != nil {
-		log.Warnf("Cannot convert the value %q with annotation key %q for the replica set %q", annotationValue, RevisionAnnotation, ro.Name)
+		log.Warnf("Cannot convert the value %q with annotation key %q for the object %q", annotationValue, RevisionAnnotation, anyObj.GetName())
 		return int32(0), false
 	}
 	return int32(intValue), true

utils/annotations/annotations_test.go

@@ -483,4 +483,27 @@ func TestGetRevisionAnnotation(t *testing.T) {
 	})
 	assert.False(t, found)
 	assert.Equal(t, int32(0), rev)
+
+	revR, found := GetRevisionAnnotation(&v1alpha1.Rollout{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "foo",
+			Namespace: metav1.NamespaceDefault,
+			Annotations: map[string]string{
+				RevisionAnnotation: "1",
+			},
+		},
+	})
+	assert.True(t, found)
+
+	revAR, found := GetRevisionAnnotation(&v1alpha1.AnalysisRun{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "foo",
+			Namespace: metav1.NamespaceDefault,
+			Annotations: map[string]string{
+				RevisionAnnotation: "1",
+			},
+		},
+	})
+	assert.True(t, found)
+	assert.Equal(t, revR, revAR)
 }

utils/record/record.go

@@ -6,7 +6,11 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
+	rolloutscheme "github.com/argoproj/argo-rollouts/pkg/client/clientset/versioned/scheme"
 	argoinformers "github.com/argoproj/argo-rollouts/pkg/client/informers/externalversions/rollouts/v1alpha1"
+	"github.com/argoproj/argo-rollouts/utils/annotations"
+	logutil "github.com/argoproj/argo-rollouts/utils/log"
 	timeutil "github.com/argoproj/argo-rollouts/utils/time"
 	"github.com/argoproj/notifications-engine/pkg/api"
 	"github.com/argoproj/notifications-engine/pkg/services"
@@ -30,10 +34,6 @@ import (
 	"sort"
 	"strings"
 	"time"
-
-	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
-	rolloutscheme "github.com/argoproj/argo-rollouts/pkg/client/clientset/versioned/scheme"
-	logutil "github.com/argoproj/argo-rollouts/utils/log"
 )
 
 func init() {
@@ -229,31 +229,45 @@ func (e *EventRecorderAdapter) K8sRecorder() record.EventRecorder {
 	return e.Recorder
 }
 
-func getAnalysisRunsMatchLabels(ro v1alpha1.Rollout, arInformer argoinformers.AnalysisRunInformer) (interface{}, error) {
+func getAnalysisRunsFilterWithLabels(ro v1alpha1.Rollout, arInformer argoinformers.AnalysisRunInformer) (interface{}, error) {
 
 	set := labels.Set(map[string]string{
 		RolloutsPodTemplateHash: ro.Status.CurrentPodHash,
 	})
 
 	ars, err := arInformer.Lister().AnalysisRuns(ro.Namespace).List(labels.SelectorFromSet(set))
-
 	if err != nil || len(ars) == 0 {
 		return nil, err
 	}
 
-	sort.Slice(ars[:], func(i, j int) bool {
-		ts1 := ars[i].ObjectMeta.CreationTimestamp.Time
-		ts2 := ars[j].ObjectMeta.CreationTimestamp.Time
+	revision, _ := annotations.GetRevisionAnnotation(&ro)
+	filteredArs := make([]*v1alpha1.AnalysisRun, 0, len(ars))
+
+	for _, ar := range ars {
+		arRevision, _ := annotations.GetRevisionAnnotation(ar)
+		if arRevision == revision {
+			filteredArs = append(filteredArs, ar)
+		}
+	}
+
+	sort.Slice(filteredArs[:], func(i, j int) bool {
+		ts1 := filteredArs[i].ObjectMeta.CreationTimestamp.Time
+		ts2 := filteredArs[j].ObjectMeta.CreationTimestamp.Time
 		return ts1.After(ts2)
 	})
 
 	var arsObj interface{}
-	arBytes, _ := json.Marshal(ars)
-	err = json.Unmarshal(arBytes, &arsObj)
+	arBytes, _ := json.Marshal(filteredArs)
 
 	if err != nil {
 		return nil, err
 	}
+
+	err = json.Unmarshal(arBytes, &arsObj)
+	if err != nil {
+		return nil, err
+	}
+
 	return arsObj, nil
 }
 
@@ -275,13 +289,16 @@ func NewAPIFactorySettings(arInformer argoinformers.AnalysisRunInformer) api.Set
 				if err != nil || ro.Namespace == "" {
 					return vars
 				}
-				arsObj, err := getAnalysisRunsMatchLabels(ro, arInformer)
 
-				if err != nil {
+				arsObj, err := getAnalysisRunsFilterWithLabels(ro, arInformer)
+
+				if err != nil || arsObj == nil {
+					log.Errorf("Failed to fetch analysisRuns:, namespace: %s, err: %v",
+						ro.Namespace, err)
 					return vars
+				} else {
+					vars = map[string]any{"rollout": obj, "analysisRuns": arsObj, "time": timeExprs}
 				}
-
-				vars = map[string]any{"rollout": obj, "analysisRuns": arsObj, "time": timeExprs}
 				return vars
 			}, nil
 		},

utils/record/record_test.go

@@ -441,6 +441,16 @@ func TestSendNotificationsNoTrigger(t *testing.T) {
 	assert.Len(t, err, 1)
 }
 
+func createAnalysisRunInformer(ars []*v1alpha1.AnalysisRun) argoinformers.AnalysisRunInformer {
+	f := argofake.NewSimpleClientset()
+	rolloutsI := argoinformersfactory.NewSharedInformerFactory(f, noResyncPeriodFunc())
+	arInformer := rolloutsI.Argoproj().V1alpha1().AnalysisRuns()
+	for _, ar := range ars {
+		_ = arInformer.Informer().GetStore().Add(ar)
+	}
+	return arInformer
+}
+
 func TestNewAPIFactorySettings(t *testing.T) {
 
 	ars := []*v1alpha1.AnalysisRun{
@@ -450,6 +460,7 @@ func TestNewAPIFactorySettings(t *testing.T) {
 				CreationTimestamp: metav1.NewTime(timeutil.Now().Add(-1 * time.Hour)),
 				Namespace:         "default",
 				Labels:            map[string]string{"rollouts-pod-template-hash": "85659df978"},
+				Annotations:       map[string]string{"rollout.argoproj.io/revision": "1"},
 			},
 		},
 		{
@@ -458,36 +469,38 @@ func TestNewAPIFactorySettings(t *testing.T) {
 				CreationTimestamp: metav1.NewTime(timeutil.Now().Add(-2 * time.Hour)),
 				Namespace:         "default",
 				Labels:            map[string]string{"rollouts-pod-template-hash": "85659df978"},
+				Annotations:       map[string]string{"rollout.argoproj.io/revision": "1"},
 			},
 		},
 	}
-
-	f := argofake.NewSimpleClientset()
-	rolloutsI := argoinformersfactory.NewSharedInformerFactory(f, noResyncPeriodFunc())
-	arInformer := rolloutsI.Argoproj().V1alpha1().AnalysisRuns()
 	ro := v1alpha1.Rollout{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      "rollout",
-			Namespace: "default",
+			Name:        "rollout",
+			Namespace:   "default",
+			Annotations: map[string]string{"rollout.argoproj.io/revision": "1"},
 		},
 		Status: v1alpha1.RolloutStatus{
 			CurrentPodHash: "85659df978",
 		},
 	}
 
 	type expectedFunc func(obj map[string]interface{}, ar interface{}) map[string]interface{}
+	type arInformerFunc func([]*v1alpha1.AnalysisRun) argoinformers.AnalysisRunInformer
+
 	testcase := []struct {
 		name       string
-		arInformer argoinformers.AnalysisRunInformer
+		arInformer arInformerFunc
 		rollout    v1alpha1.Rollout
 		ars        []*v1alpha1.AnalysisRun
 		expected   expectedFunc
 	}{
 		{
-			name:       "Send notification for rollout and analysisRun matches label",
-			arInformer: arInformer,
-			rollout:    ro,
-			ars:        ars,
+			name: "Send notification with rollout and analysisRun",
+			arInformer: func(ars []*v1alpha1.AnalysisRun) argoinformers.AnalysisRunInformer {
+				return createAnalysisRunInformer(ars)
+			},
+			rollout: ro,
+			ars:     ars,
 			expected: func(obj map[string]interface{}, ar interface{}) map[string]interface{} {
 				return map[string]interface{}{
 					"rollout":      obj,
@@ -497,30 +510,22 @@ func TestNewAPIFactorySettings(t *testing.T) {
 			},
 		},
 		{
-			name:       "Send notification only rollout when pod hash doesn't match",
-			arInformer: arInformer,
-			rollout:    ro,
-			ars: append(ars, &v1alpha1.AnalysisRun{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:              "analysis-run-2",
-					CreationTimestamp: metav1.Now(),
-					Namespace:         "default-1",
-					Labels:            map[string]string{"rollouts-pod-template-hash": "123456"},
+			name: "Send notification rollout when revision  and lables doesn't match",
+			arInformer: func(ars []*v1alpha1.AnalysisRun) argoinformers.AnalysisRunInformer {
+				return createAnalysisRunInformer(ars)
+			},
+			rollout: ro,
+			ars: []*v1alpha1.AnalysisRun{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:              "analysis-run-3",
+						CreationTimestamp: metav1.NewTime(timeutil.Now().Add(-2 * time.Hour)),
+						Namespace:         "default",
+						Labels:            map[string]string{"rollouts-pod-template-hash": "1234"},
+						Annotations:       map[string]string{"rollout.argoproj.io/revision": "2"},
+					},
 				},
-			}),
-			expected: func(obj map[string]interface{}, ar interface{}) map[string]interface{} {
-				return map[string]interface{}{
-					"rollout":      obj,
-					"analysisRuns": ar,
-					"time":         timeExprs,
-				}
 			},
-		},
-		{
-			name:       "arInformer is nil",
-			arInformer: nil,
-			rollout:    ro,
-			ars:        ars,
 			expected: func(obj map[string]interface{}, ar interface{}) map[string]interface{} {
 				return map[string]interface{}{
 					"rollout": obj,
@@ -529,27 +534,26 @@ func TestNewAPIFactorySettings(t *testing.T) {
 			},
 		},
 		{
-			name: "rollout object namespace and pod-hash not matched",
-			rollout: v1alpha1.Rollout{
-				ObjectMeta: metav1.ObjectMeta{
-					Name: "rollout",
-				},
+			name: "arInformer is nil",
+			arInformer: func(ars []*v1alpha1.AnalysisRun) argoinformers.AnalysisRunInformer {
+				return nil
 			},
-			ars: ars,
+			rollout: ro,
+			ars:     nil,
 			expected: func(obj map[string]interface{}, ar interface{}) map[string]interface{} {
-				ro.Namespace = ""
-				ro.Status.CurrentPodHash = "1234"
 				return map[string]interface{}{
 					"rollout": obj,
 					"time":    timeExprs,
 				}
 			},
 		},
 		{
-			name:       "no analysisRuns",
-			arInformer: nil,
-			rollout:    ro,
-			ars:        []*v1alpha1.AnalysisRun{},
+			name: "analysisRuns list empty in given namespace",
+			arInformer: func(ars []*v1alpha1.AnalysisRun) argoinformers.AnalysisRunInformer {
+				return createAnalysisRunInformer(ars)
+			},
+			rollout: ro,
+			ars:     nil,
 			expected: func(obj map[string]interface{}, ar interface{}) map[string]interface{} {
 				return map[string]interface{}{
 					"rollout": obj,
@@ -561,10 +565,8 @@ func TestNewAPIFactorySettings(t *testing.T) {
 
 	for _, test := range testcase {
 		t.Run(test.name, func(t *testing.T) {
-			for _, ar := range test.ars {
-				_ = arInformer.Informer().GetStore().Add(ar)
-			}
-			settings := NewAPIFactorySettings(test.arInformer)
+
+			settings := NewAPIFactorySettings(test.arInformer(test.ars))
 			getVars, err := settings.InitGetVars(nil, nil, nil)
 			require.NoError(t, err)
 			if err != nil {
@@ -573,7 +575,7 @@ func TestNewAPIFactorySettings(t *testing.T) {
 
 			obj, _ := runtime.DefaultUnstructuredConverter.ToUnstructured(&test.rollout)
 
-			arBytes, err := json.Marshal(ars)
+			arBytes, err := json.Marshal(test.ars)
 			var arsObj interface{}
 			_ = json.Unmarshal(arBytes, &arsObj)
 			vars := getVars(obj, services.Destination{})