feat: ability to auto-create Services for each template in an Experiment (argoproj#1158)

* feat: Modify Experiment Controller to auto-create service for each Experiment Template

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

Author: khhirani

controller/controller.go

@@ -199,6 +199,7 @@ func NewManager(
 		AnalysisRunInformer:             analysisRunInformer,
 		AnalysisTemplateInformer:        analysisTemplateInformer,
 		ClusterAnalysisTemplateInformer: clusterAnalysisTemplateInformer,
+		ServiceInformer:                 servicesInformer,
 		ResyncPeriod:                    resyncPeriod,
 		RolloutWorkQueue:                rolloutWorkqueue,
 		ExperimentWorkQueue:             experimentWorkqueue,

experiments/analysisrun_test.go

@@ -460,6 +460,8 @@ func TestAssessAnalysisRunStatusesAfterTemplateSuccess(t *testing.T) {
 			ar2.Status.Phase = test.second
 			e.Status.AnalysisRuns[1].Phase = test.second
 			f := newFixture(t, e, rs, ar1, ar2)
+			f.expectPatchReplicaSetAction(rs) // Add scaleDownDelay annotation to RS
+			f.expectGetReplicaSetAction(rs)   // Happens during scale down logic
 			if test.expected != v1alpha1.AnalysisPhaseRunning {
 				patchIdx := f.expectPatchExperimentAction(e)
 				f.run(getKey(e, t))
@@ -478,6 +480,7 @@ func TestAssessAnalysisRunStatusesAfterTemplateSuccess(t *testing.T) {
 func TestFailExperimentWhenAnalysisFails(t *testing.T) {
 	templates := generateTemplates("bar")
 	e := newExperiment("foo", templates, "")
+	e.Spec.ScaleDownDelaySeconds = pointer.Int32Ptr(0)
 	e.Spec.Analyses = []v1alpha1.ExperimentAnalysisTemplateRef{
 		{
 			Name:         "success-rate",
@@ -490,6 +493,7 @@ func TestFailExperimentWhenAnalysisFails(t *testing.T) {
 	}
 	e.Status.Phase = v1alpha1.AnalysisPhaseRunning
 	e.Spec.Duration = "5m"
+	e.Spec.ScaleDownDelaySeconds = pointer.Int32Ptr(0)
 	e.Status.AvailableAt = secondsAgo(60)
 	rs := templateToRS(e, templates[0], 1)
 	ar1 := analysisTemplateToRun("success-rate", e, &v1alpha1.AnalysisTemplateSpec{})
@@ -551,7 +555,8 @@ func TestFailExperimentWhenAnalysisFails(t *testing.T) {
 			f := newFixture(t, e, rs, ar1, ar2)
 
 			if test.expected == v1alpha1.AnalysisPhaseFailed {
-				f.expectUpdateReplicaSetAction(rs) // scale down to 0
+				// No scale down delay actions since scaleDownDelay seconds is 0
+				f.expectUpdateReplicaSetAction(rs)
 			}
 			patchIdx := f.expectPatchExperimentAction(e)
 			f.run(getKey(e, t))
@@ -593,6 +598,7 @@ func TestCompleteExperimentOnSuccessfulRequiredAnalysisRun(t *testing.T) {
 
 	f := newFixture(t, e, rs, ar)
 	defer f.Close()
+	f.expectGetReplicaSetAction(rs)
 	f.expectUpdateReplicaSetAction(rs)
 	patchIndex := f.expectPatchExperimentAction(e)
 	f.run(getKey(e, t))
@@ -643,6 +649,7 @@ func TestDoNotCompleteExperimentWithRemainingRequiredAnalysisRun(t *testing.T) {
 
 	f := newFixture(t, e, rs, ar, ar2)
 	defer f.Close()
+	f.expectGetReplicaSetAction(rs)
 	f.expectUpdateReplicaSetAction(rs)
 	patchIndex := f.expectPatchExperimentAction(e)
 	f.run(getKey(e, t))
@@ -653,6 +660,7 @@ func TestDoNotCompleteExperimentWithRemainingRequiredAnalysisRun(t *testing.T) {
 func TestCompleteExperimentWithNoRequiredAnalysis(t *testing.T) {
 	templates := generateTemplates("bar")
 	e := newExperiment("foo", templates, "1m")
+	e.Spec.ScaleDownDelaySeconds = pointer.Int32Ptr(0)
 	e.Spec.Analyses = []v1alpha1.ExperimentAnalysisTemplateRef{
 		{
 			Name:         "success-rate",
@@ -692,6 +700,7 @@ func TestCompleteExperimentWithNoRequiredAnalysis(t *testing.T) {
 func TestTerminateAnalysisRuns(t *testing.T) {
 	templates := generateTemplates("bar")
 	e := newExperiment("foo", templates, "")
+	e.Spec.ScaleDownDelaySeconds = pointer.Int32Ptr(0)
 	e.Spec.Analyses = []v1alpha1.ExperimentAnalysisTemplateRef{
 		{
 			Name:         "success-rate",

experiments/controller.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"time"
 
+	informersv1 "k8s.io/client-go/informers/core/v1"
+	listersv1 "k8s.io/client-go/listers/core/v1"
+
 	log "github.com/sirupsen/logrus"
 	appsv1 "k8s.io/api/apps/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
@@ -48,6 +51,7 @@ type Controller struct {
 	analysisTemplateLister        listers.AnalysisTemplateLister
 	clusterAnalysisTemplateLister listers.ClusterAnalysisTemplateLister
 	analysisRunLister             listers.AnalysisRunLister
+	serviceLister                 listersv1.ServiceLister
 
 	replicaSetSynced              cache.InformerSynced
 	experimentSynced              cache.InformerSynced
@@ -83,6 +87,7 @@ type ControllerConfig struct {
 	AnalysisRunInformer             informers.AnalysisRunInformer
 	AnalysisTemplateInformer        informers.AnalysisTemplateInformer
 	ClusterAnalysisTemplateInformer informers.ClusterAnalysisTemplateInformer
+	ServiceInformer                 informersv1.ServiceInformer
 	ResyncPeriod                    time.Duration
 	RolloutWorkQueue                workqueue.RateLimitingInterface
 	ExperimentWorkQueue             workqueue.RateLimitingInterface
@@ -107,6 +112,7 @@ func NewController(cfg ControllerConfig) *Controller {
 		analysisTemplateLister:        cfg.AnalysisTemplateInformer.Lister(),
 		clusterAnalysisTemplateLister: cfg.ClusterAnalysisTemplateInformer.Lister(),
 		analysisRunLister:             cfg.AnalysisRunInformer.Lister(),
+		serviceLister:                 cfg.ServiceInformer.Lister(),
 		metricsServer:                 cfg.MetricsServer,
 		rolloutWorkqueue:              cfg.RolloutWorkQueue,
 		experimentWorkqueue:           cfg.ExperimentWorkQueue,
@@ -275,16 +281,24 @@ func (ec *Controller) syncHandler(key string) error {
 		return err
 	}
 
+	templateServices, err := ec.getServicesForExperiment(experiment)
+	if err != nil {
+		return err
+	}
+
 	exCtx := newExperimentContext(
 		experiment,
 		templateRSs,
+		templateServices,
 		ec.kubeclientset,
 		ec.argoProjClientset,
 		ec.replicaSetLister,
 		ec.analysisTemplateLister,
 		ec.clusterAnalysisTemplateLister,
 		ec.analysisRunLister,
+		ec.serviceLister,
 		ec.recorder,
+		ec.resyncPeriod,
 		ec.enqueueExperimentAfter,
 	)
 

experiments/controller_test.go

@@ -8,11 +8,13 @@ import (
 	"testing"
 	"time"
 
+	"k8s.io/apimachinery/pkg/util/intstr"
+
+	"github.com/stretchr/testify/assert"
 	"github.com/undefinedlabs/go-mpatch"
 
 	"github.com/argoproj/argo-rollouts/utils/queue"
 
-	"github.com/stretchr/testify/assert"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
@@ -70,6 +72,7 @@ type fixture struct {
 	analysisRunLister             []*v1alpha1.AnalysisRun
 	analysisTemplateLister        []*v1alpha1.AnalysisTemplate
 	clusterAnalysisTemplateLister []*v1alpha1.ClusterAnalysisTemplate
+	serviceLister                 []*corev1.Service
 	// Actions expected to happen on the client.
 	kubeactions []core.Action
 	actions     []core.Action
@@ -102,6 +105,9 @@ func newFixture(t *testing.T, objects ...runtime.Object) *fixture {
 		case *appsv1.ReplicaSet:
 			f.kubeobjects = append(f.kubeobjects, obj)
 			f.replicaSetLister = append(f.replicaSetLister, obj.(*appsv1.ReplicaSet))
+		case *corev1.Service:
+			f.kubeobjects = append(f.kubeobjects, obj)
+			f.serviceLister = append(f.serviceLister, obj.(*corev1.Service))
 		}
 	}
 	f.client = fake.NewSimpleClientset(f.objects...)
@@ -234,6 +240,32 @@ func newCondition(reason string, experiment *v1alpha1.Experiment) *v1alpha1.Expe
 	return nil
 }
 
+func templateToService(ex *v1alpha1.Experiment, template v1alpha1.TemplateSpec, replicaSet appsv1.ReplicaSet) *corev1.Service {
+	if template.Service != nil {
+		service := &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      replicaSet.Name,
+				Namespace: replicaSet.Namespace,
+				Annotations: map[string]string{
+					v1alpha1.ExperimentNameAnnotationKey:         ex.Name,
+					v1alpha1.ExperimentTemplateNameAnnotationKey: template.Name,
+				},
+				OwnerReferences: []metav1.OwnerReference{*metav1.NewControllerRef(ex, experimentKind)},
+			},
+			Spec: corev1.ServiceSpec{
+				Selector: replicaSet.Spec.Selector.MatchLabels,
+				Ports: []corev1.ServicePort{{
+					Protocol:   "TCP",
+					Port:       int32(80),
+					TargetPort: intstr.FromInt(8080),
+				}},
+			},
+		}
+		return service
+	}
+	return nil
+}
+
 func templateToRS(ex *v1alpha1.Experiment, template v1alpha1.TemplateSpec, availableReplicas int32) *appsv1.ReplicaSet {
 	rsLabels := map[string]string{}
 	for k, v := range template.Selector.MatchLabels {
@@ -335,6 +367,7 @@ func (f *fixture) newController(resync resyncFunc) (*Controller, informers.Share
 		AnalysisRunInformer:             i.Argoproj().V1alpha1().AnalysisRuns(),
 		AnalysisTemplateInformer:        i.Argoproj().V1alpha1().AnalysisTemplates(),
 		ClusterAnalysisTemplateInformer: i.Argoproj().V1alpha1().ClusterAnalysisTemplates(),
+		ServiceInformer:                 k8sI.Core().V1().Services(),
 		ResyncPeriod:                    resync(),
 		RolloutWorkQueue:                rolloutWorkqueue,
 		ExperimentWorkQueue:             experimentWorkqueue,
@@ -371,6 +404,10 @@ func (f *fixture) newController(resync resyncFunc) (*Controller, informers.Share
 		k8sI.Apps().V1().ReplicaSets().Informer().GetIndexer().Add(r)
 	}
 
+	for _, r := range f.serviceLister {
+		k8sI.Core().V1().Services().Informer().GetIndexer().Add(r)
+	}
+
 	for _, r := range f.analysisRunLister {
 		i.Argoproj().V1alpha1().AnalysisRuns().Informer().GetIndexer().Add(r)
 	}
@@ -484,7 +521,9 @@ func filterInformerActions(actions []core.Action) []core.Action {
 			action.Matches("list", "clusteranalysistemplates") ||
 			action.Matches("watch", "clusteranalysistemplates") ||
 			action.Matches("list", "analysisruns") ||
-			action.Matches("watch", "analysisruns") {
+			action.Matches("watch", "analysisruns") ||
+			action.Matches("watch", "services") ||
+			action.Matches("list", "services") {
 			continue
 		}
 		ret = append(ret, action)
@@ -493,6 +532,18 @@ func filterInformerActions(actions []core.Action) []core.Action {
 	return ret
 }
 
+func (f *fixture) expectCreateServiceAction(service *corev1.Service) int {
+	len := len(f.kubeactions)
+	f.kubeactions = append(f.kubeactions, core.NewCreateAction(schema.GroupVersionResource{Resource: "services"}, service.Namespace, service))
+	return len
+}
+
+func (f *fixture) expectDeleteServiceAction(service *corev1.Service) int {
+	len := len(f.kubeactions)
+	f.kubeactions = append(f.kubeactions, core.NewDeleteAction(schema.GroupVersionResource{Resource: "services"}, service.Namespace, service.Name))
+	return len
+}
+
 func (f *fixture) expectCreateReplicaSetAction(r *appsv1.ReplicaSet) int {
 	len := len(f.kubeactions)
 	f.kubeactions = append(f.kubeactions, core.NewCreateAction(schema.GroupVersionResource{Resource: "replicasets"}, r.Namespace, r))
@@ -586,6 +637,27 @@ func (f *fixture) getCreatedReplicaSet(index int) *appsv1.ReplicaSet {
 	return rs
 }
 
+func (f *fixture) verifyPatchedReplicaSetAddScaleDownDelay(index int, scaleDownDelaySeconds int32) {
+	action := filterInformerActions(f.kubeclient.Actions())[index]
+	patchAction, ok := action.(core.PatchAction)
+	if !ok {
+		assert.Fail(f.t, "Expected Patch action, not %s", action.GetVerb())
+	}
+	now := metav1.Now().Add(time.Duration(scaleDownDelaySeconds) * time.Second).UTC().Format(time.RFC3339)
+	patch := fmt.Sprintf(addScaleDownAtAnnotationsPatch, v1alpha1.DefaultReplicaSetScaleDownDeadlineAnnotationKey, now)
+	assert.Equal(f.t, string(patchAction.GetPatch()), patch)
+}
+
+func (f *fixture) verifyPatchedReplicaSetRemoveScaleDownDelayAnnotation(index int) {
+	action := filterInformerActions(f.kubeclient.Actions())[index]
+	patchAction, ok := action.(core.PatchAction)
+	if !ok {
+		assert.Fail(f.t, "Expected Patch action, not %s", action.GetVerb())
+	}
+	patch := fmt.Sprintf(removeScaleDownAtAnnotationsPatch, v1alpha1.DefaultReplicaSetScaleDownDeadlineAnnotationKey)
+	assert.Equal(f.t, string(patchAction.GetPatch()), patch)
+}
+
 func (f *fixture) getUpdatedReplicaSet(index int) *appsv1.ReplicaSet {
 	action := filterInformerActions(f.kubeclient.Actions())[index]
 	updateAction, ok := action.(core.UpdateAction)