Add scheduling logic

README.md

@@ -21,7 +21,7 @@ apiVersion: deployment.skyscanner.net/v1alpha1
 kind: ProgressiveRollout
 metadata:
   name: myprogressiverollout
-  namespace: argoc
+  namespace: argocd
 spec:
   # a reference to the target ApplicationSet
   sourceRef:
@@ -33,17 +33,17 @@ spec:
       # human friendly name
     - name: two clusters as canary in EMEA
       # how many targets to update in parallel
-      # can be an integer or %. Default to 1
+      # can be an integer or %.
       maxParallel: 2
       # how many targets to update from the selector result
-      # can be an integer or %. Default to 100%.
+      # can be an integer or %.
       maxTargets: 2
       # which targets to update
       targets:
         clusters:
           selector:
             matchLabels:
-            area: emea
+              area: emea
     - name: rollout to remaining clusters
       maxParallel: 25%
       maxTargets: 100%
@@ -62,10 +62,10 @@ See [CONTRIBUTING.md](./CONTRIBUTING.md)
 
 ## Development
 
+### Local development with Kubebuilder
+
 1. Install `pre-commit`: see <https://pre-commit.com/#install>
-1. Install `kind`: see <https://kind.sigs.k8s.io/docs/user/quick-start/#installation>
-1. Install `ArgoCD`: see <https://argoproj.github.io/argo-cd/getting_started/>
-1. Install `ApplicationSet` controller: see <https://github.com/argoproj-labs/applicationset>
+1. Install `kubebuilder`: see <https://book.kubebuilder.io/quick-start.html#installation>
 1. Install `ArgoCD Application` API pkg: see `hack/install-argocd-application.sh`
 
 ### Update ArgoCD Application API package

api/v1alpha1/progressiverollout_types.go

@@ -90,8 +90,8 @@ func (in *ProgressiveRollout) NewStatusCondition(t string, s metav1.ConditionSta
 	}
 }
 
-// IsOwnedBy returns true if the ProgressiveRollout object has a reference to one of the owners
-func (in *ProgressiveRollout) IsOwnedBy(owners []metav1.OwnerReference) bool {
+// Owns returns true if the ProgressiveRollout object has a reference to one of the owners
+func (in *ProgressiveRollout) Owns(owners []metav1.OwnerReference) bool {
 	for _, owner := range owners {
 		if owner.Kind == in.Spec.SourceRef.Kind && owner.APIVersion == *in.Spec.SourceRef.APIGroup && owner.Name == in.Spec.SourceRef.Name {
 			return true

api/v1alpha1/progressiverollout_types_test.go

@@ -8,7 +8,7 @@ import (
 	"testing"
 )
 
-func TestHasOwnerReference(t *testing.T) {
+func TestOwns(t *testing.T) {
 	testCases := []struct {
 		ownerReferences []metav1.OwnerReference
 		expected        bool
@@ -45,8 +45,8 @@ func TestHasOwnerReference(t *testing.T) {
 		}}
 
 	for _, testCase := range testCases {
-		got := pr.IsOwnedBy(testCase.ownerReferences)
-		g := NewGomegaWithT(t)
+		got := pr.Owns(testCase.ownerReferences)
+		g := NewWithT(t)
 		g.Expect(got).To(Equal(testCase.expected))
 	}
 }

controllers/progressiverollout_controller.go

@@ -19,6 +19,9 @@ package controllers
 import (
 	"context"
 	"fmt"
+
+	deploymentskyscannernetv1alpha1 "github.com/Skyscanner/argocd-progressive-rollout/api/v1alpha1"
+	"github.com/Skyscanner/argocd-progressive-rollout/internal/scheduler"
 	"github.com/Skyscanner/argocd-progressive-rollout/internal/utils"
 	argov1alpha1 "github.com/argoproj/argo-cd/pkg/apis/application/v1alpha1"
 	"github.com/go-logr/logr"
@@ -34,8 +37,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
-
-	deploymentskyscannernetv1alpha1 "github.com/Skyscanner/argocd-progressive-rollout/api/v1alpha1"
 )
 
 // ProgressiveRolloutReconciler reconciles a ProgressiveRollout object
@@ -51,6 +52,7 @@ type ProgressiveRolloutReconciler struct {
 // +kubebuilder:rbac:groups="argoproj.io",resources=applications,verbs=get;list;watch
 // +kubebuilder:rbac:groups="argoproj.io",resources=applications/status,verbs=get;list;watch
 
+// Reconcile performs the reconciling for a single named ProgressiveRollout object
 func (r *ProgressiveRolloutReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	log := r.Log.WithValues("progressiverollout", req.NamespacedName)
 
@@ -60,24 +62,63 @@ func (r *ProgressiveRolloutReconciler) Reconcile(ctx context.Context, req ctrl.R
 		log.Error(err, "unable to fetch ProgressiveRollout")
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
-	// Always log the ApplicationSet owner
+
 	log = r.Log.WithValues("applicationset", pr.Spec.SourceRef.Name)
 
 	for _, stage := range pr.Spec.Stages {
 		log = r.Log.WithValues("stage", stage.Name)
 
-		targets, err := r.GetTargetClusters(stage.Targets.Clusters.Selector)
+		// Get the clusters to update
+		clusters, err := r.getClustersFromSelector(stage.Targets.Clusters.Selector)
 		if err != nil {
-			log.Error(err, "unable to fetch targets")
+			log.Error(err, "unable to fetch clusters")
+			return ctrl.Result{}, err
+		}
+		r.Log.V(1).Info("clusters selected", "clusters", fmt.Sprintf("%v", clusters.Items))
+
+		// Get the Applications owned by the ProgressiveRollout targeting the clusters
+		apps, err := r.getOwnedAppsFromClusters(clusters, pr)
+		if err != nil {
+			log.Error(err, "unable to fetch apps")
+			return ctrl.Result{}, err
+		}
+		r.Log.V(1).Info("apps selected", "apps", fmt.Sprintf("%v", apps))
+
+		// Remove the annotation from the OutOfSync Applications before passing them to the Scheduler
+		// This action allows the Scheduler to keep track at which stage an Application has been synced.
+		outOfSyncApps := utils.FilterAppsBySyncStatusCode(apps, argov1alpha1.SyncStatusCodeOutOfSync)
+		if err = r.removeAnnotationFromApps(&outOfSyncApps, utils.ProgressiveRolloutSyncedAtStageKey); err != nil {
 			return ctrl.Result{}, err
 		}
-		r.Log.V(1).Info("targets selected", "targets", targets.Items)
-		r.Log.Info("stage completed")
+
+		// Get the Applications to update
+		scheduledApps := scheduler.Scheduler(apps, stage)
+
+		for _, s := range scheduledApps {
+			// TODO: add sync method here
+			r.Log.Info("syncing app", "app", s)
+		}
+
+		if scheduler.IsStageFailed(apps, stage) {
+			// TODO: updated status
+			r.Log.Info("stage failed")
+			return ctrl.Result{}, nil
+		}
+
+		if scheduler.IsStageComplete(apps, stage) {
+			// TODO: update status
+			r.Log.Info("stage completed")
+		} else {
+			// TODO: update status
+			r.Log.Info("stage in progress")
+			// Stage in progress, we reconcile again until the stage is completed or failed
+			return ctrl.Result{Requeue: true}, nil
+		}
 	}
 
 	log.Info("all stages completed")
 
-	// Rollout completed
+	// Progressive rollout completed
 	completed := pr.NewStatusCondition(deploymentskyscannernetv1alpha1.CompletedCondition, metav1.ConditionTrue, deploymentskyscannernetv1alpha1.StagesCompleteReason, "All stages completed")
 	apimeta.SetStatusCondition(pr.GetStatusConditions(), completed)
 	if err := r.Client.Status().Update(ctx, &pr); err != nil {
@@ -88,6 +129,7 @@ func (r *ProgressiveRolloutReconciler) Reconcile(ctx context.Context, req ctrl.R
 	return ctrl.Result{}, nil
 }
 
+// SetupWithManager adds the reconciler to the manager, so that it gets started when the manager is started.
 func (r *ProgressiveRolloutReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&deploymentskyscannernetv1alpha1.ProgressiveRollout{}).
@@ -100,6 +142,7 @@ func (r *ProgressiveRolloutReconciler) SetupWithManager(mgr ctrl.Manager) error
 		Complete(r)
 }
 
+// requestsForApplicationChange returns a reconcile request for a Progressive Rollout object when an Application change
 func (r *ProgressiveRolloutReconciler) requestsForApplicationChange(o client.Object) []reconcile.Request {
 
 	/*
@@ -124,7 +167,7 @@ func (r *ProgressiveRolloutReconciler) requestsForApplicationChange(o client.Obj
 	}
 
 	for _, pr := range list.Items {
-		if pr.IsOwnedBy(app.GetOwnerReferences()) {
+		if pr.Owns(app.GetOwnerReferences()) {
 			requests = append(requests, reconcile.Request{NamespacedName: types.NamespacedName{
 				Namespace: pr.Namespace,
 				Name:      pr.Name,
@@ -135,6 +178,7 @@ func (r *ProgressiveRolloutReconciler) requestsForApplicationChange(o client.Obj
 	return requests
 }
 
+// requestsForSecretChange returns a reconcile request for a Progressive Rollout object when a Secret change
 func (r *ProgressiveRolloutReconciler) requestsForSecretChange(o client.Object) []reconcile.Request {
 
 	/*
@@ -174,7 +218,7 @@ func (r *ProgressiveRolloutReconciler) requestsForSecretChange(o client.Object)
 
 	for _, pr := range prList.Items {
 		for _, app := range appList.Items {
-			if app.Spec.Destination.Server == string(s.Data["server"]) && pr.IsOwnedBy(app.GetOwnerReferences()) {
+			if app.Spec.Destination.Server == string(s.Data["server"]) && pr.Owns(app.GetOwnerReferences()) {
 				/*
 					Consider the following scenario:
 					- 2 Applications
@@ -198,8 +242,8 @@ func (r *ProgressiveRolloutReconciler) requestsForSecretChange(o client.Object)
 	return requests
 }
 
-// GetTargetClusters returns a list of ArgoCD clusters matching the provided label selector
-func (r *ProgressiveRolloutReconciler) GetTargetClusters(selector metav1.LabelSelector) (corev1.SecretList, error) {
+// getClustersFromSelector returns a list of ArgoCD clusters matching the provided label selector
+func (r *ProgressiveRolloutReconciler) getClustersFromSelector(selector metav1.LabelSelector) (corev1.SecretList, error) {
 	secrets := corev1.SecretList{}
 	ctx := context.Background()
 
@@ -220,3 +264,43 @@ func (r *ProgressiveRolloutReconciler) GetTargetClusters(selector metav1.LabelSe
 
 	return secrets, nil
 }
+
+// getOwnedAppsFromClusters returns a list of Applications targeting the specified clusters and owned by the specified ProgressiveRollout
+func (r *ProgressiveRolloutReconciler) getOwnedAppsFromClusters(clusters corev1.SecretList, pr deploymentskyscannernetv1alpha1.ProgressiveRollout) ([]argov1alpha1.Application, error) {
+	apps := []argov1alpha1.Application{{}}
+	appList := argov1alpha1.ApplicationList{}
+	ctx := context.Background()
+
+	if err := r.List(ctx, &appList); err != nil {
+		r.Log.Error(err, "failed to list Application")
+		return apps, err
+	}
+
+	for _, c := range clusters.Items {
+		for _, app := range appList.Items {
+			if pr.Owns(app.GetOwnerReferences()) && string(c.Data["server"]) == app.Spec.Destination.Server {
+				apps = append(apps, app)
+			}
+		}
+	}
+
+	utils.SortAppsByName(&apps)
+
+	return apps, nil
+}
+
+// removeAnnotationFromApps remove an annotation from the given Applications
+func (r *ProgressiveRolloutReconciler) removeAnnotationFromApps(apps *[]argov1alpha1.Application, annotation string) error {
+	ctx := context.Background()
+
+	for _, app := range *apps {
+		if _, ok := app.Annotations[annotation]; ok {
+			delete(app.Annotations, annotation)
+			if err := r.Client.Update(ctx, &app); err != nil {
+				r.Log.Error(err, "failed to update Application", "app", app.Name)
+				return err
+			}
+		}
+	}
+	return nil
+}

controllers/progressiverollout_controller_test.go

@@ -219,14 +219,44 @@ var _ = Describe("ProgressiveRollout Controller", func() {
 
 	Describe("Reconciliation loop", func() {
 		It("should reconcile", func() {
-			By("creating a progressive rollout object")
+			By("creating an ArgoCD cluster")
+			cluster := &corev1.Secret{
+				ObjectMeta: metav1.ObjectMeta{Name: "single-stage-cluster", Namespace: namespace, Labels: map[string]string{utils.ArgoCDSecretTypeLabel: utils.ArgoCDSecretTypeCluster}},
+				Data: map[string][]byte{
+					"server": []byte("https://single-stage-pr.kubernetes.io"),
+				},
+			}
+			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
+
+			By("creating an application targeting the cluster")
+			singleStageApp := &argov1alpha1.Application{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "single-stage-app",
+					Namespace: namespace,
+					OwnerReferences: []metav1.OwnerReference{{
+						APIVersion: utils.AppSetAPIGroup,
+						Kind:       utils.AppSetKind,
+						Name:       "single-stage-appset",
+						UID:        uuid.NewUUID(),
+					}},
+				},
+				Spec: argov1alpha1.ApplicationSpec{Destination: argov1alpha1.ApplicationDestination{
+					Server:    "https://single-stage-pr.kubernetes.io",
+					Namespace: namespace,
+					Name:      "remote-cluster",
+				}},
+				Status: argov1alpha1.ApplicationStatus{Sync: argov1alpha1.SyncStatus{Status: argov1alpha1.SyncStatusCodeOutOfSync}},
+			}
+			Expect(k8sClient.Create(ctx, singleStageApp)).To(Succeed())
+
+			By("creating a progressive rollout")
 			singleStagePR = &deploymentskyscannernetv1alpha1.ProgressiveRollout{
 				ObjectMeta: metav1.ObjectMeta{Name: "single-stage-pr", Namespace: namespace},
 				Spec: deploymentskyscannernetv1alpha1.ProgressiveRolloutSpec{
 					SourceRef: corev1.TypedLocalObjectReference{
 						APIGroup: &appSetAPIRef,
-						Kind:     "",
-						Name:     "",
+						Kind:     utils.AppSetKind,
+						Name:     "single-stage-appset",
 					},
 					Stages: []deploymentskyscannernetv1alpha1.ProgressiveRolloutStage{{
 						Name:        "stage 1",

go.mod

@@ -4,6 +4,7 @@ go 1.15
 
 require (
 	github.com/argoproj/argo-cd v0.8.1-0.20210218202601-6de3cf44a4cb
+	github.com/argoproj/gitops-engine v0.2.1-0.20210129183711-c5b7114c501f
 	github.com/go-logr/logr v0.3.0
 	github.com/onsi/ginkgo v1.14.2
 	github.com/onsi/gomega v1.10.3