Add basic scheduler structure

controllers/progressiverollout_controller.go

@@ -67,26 +67,32 @@ func (r *ProgressiveRolloutReconciler) Reconcile(ctx context.Context, req ctrl.R
 	for _, stage := range pr.Spec.Stages {
 		log = r.Log.WithValues("stage", stage.Name)
 
-		clusters, err := r.getClusters(stage.Targets.Clusters.Selector)
+		clusters, err := r.getClustersFromSelector(stage.Targets.Clusters.Selector)
 		if err != nil {
 			log.Error(err, "unable to fetch clusters")
 			return ctrl.Result{}, err
 		}
 		r.Log.V(1).Info("clusters selected", "clusters", clusters.Items)
 
-		apps, err := r.getApps(clusters, pr)
+		apps, err := r.getAppsFromClusters(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))
 
-		syncApps := scheduler.Scheduler(clusters, apps, stage)
+		scheduledApps := scheduler.Scheduler(apps, stage)
 
-		for _, s := range syncApps {
+		for _, s := range scheduledApps {
 			r.Log.Info("syncing app", "app", s)
 		}
-		r.Log.Info("stage completed")
+
+		if scheduler.IsStageComplete(apps, stage) {
+			r.Log.Info("stage completed")
+		} else {
+			r.Log.Info("stage in progress")
+			return ctrl.Result{Requeue: true}, nil
+		}
 	}
 
 	log.Info("all stages completed")
@@ -214,8 +220,8 @@ func (r *ProgressiveRolloutReconciler) requestsForSecretChange(o client.Object)
 	return requests
 }
 
-// getClusters returns a list of ArgoCD clusters matching the provided label selector
-func (r *ProgressiveRolloutReconciler) getClusters(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()
 
@@ -237,8 +243,8 @@ func (r *ProgressiveRolloutReconciler) getClusters(selector metav1.LabelSelector
 	return secrets, nil
 }
 
-// getApps returns a list of Applications targeting the specified clusters and owned by the specified ProgressiveRollout
-func (r *ProgressiveRolloutReconciler) getApps(clusters corev1.SecretList, pr deploymentskyscannernetv1alpha1.ProgressiveRollout) ([]argov1alpha1.Application, error) {
+// getAppsFromClusters returns a list of Applications targeting the specified clusters and owned by the specified ProgressiveRollout
+func (r *ProgressiveRolloutReconciler) getAppsFromClusters(clusters corev1.SecretList, pr deploymentskyscannernetv1alpha1.ProgressiveRollout) ([]argov1alpha1.Application, error) {
 	apps := []argov1alpha1.Application{{}}
 	appList := argov1alpha1.ApplicationList{}
 	ctx := context.Background()

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

internal/scheduler/scheduler.go

@@ -2,11 +2,38 @@ package scheduler
 
 import (
 	deploymentskyscannernetv1alpha1 "github.com/Skyscanner/argocd-progressive-rollout/api/v1alpha1"
+	"github.com/Skyscanner/argocd-progressive-rollout/internal/utils"
 	argov1alpha1 "github.com/argoproj/argo-cd/pkg/apis/application/v1alpha1"
-	corev1 "k8s.io/api/core/v1"
+	"github.com/argoproj/gitops-engine/pkg/health"
+	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
-func Scheduler(clusters corev1.SecretList, apps []argov1alpha1.Application, stage deploymentskyscannernetv1alpha1.ProgressiveRolloutStage) []string {
+// Scheduler returns a list of apps to sync
+func Scheduler(apps []argov1alpha1.Application, stage deploymentskyscannernetv1alpha1.ProgressiveRolloutStage) []string {
+
 	var scheduledApps []string
+
+	oufOfSyncApps := utils.GetAppsBySyncStatusCode(apps, argov1alpha1.SyncStatusCodeOutOfSync)
+	progressingApps := utils.GetAppsByHealthStatusCode(apps, health.HealthStatusProgressing)
+
+	maxTargets, err := intstr.GetScaledValueFromIntOrPercent(&stage.MaxTargets, len(oufOfSyncApps), false)
+	if err != nil {
+		return scheduledApps
+	}
+	maxParallel, err := intstr.GetScaledValueFromIntOrPercent(&stage.MaxParallel, maxTargets, false)
+	if err != nil {
+		return scheduledApps
+	}
+
+	if len(oufOfSyncApps) > 0 {
+		for i := 0; i < maxParallel-len(progressingApps); i++ {
+			scheduledApps = append(scheduledApps, oufOfSyncApps[i].Name)
+		}
+	}
 	return scheduledApps
 }
+
+func IsStageComplete(apps []argov1alpha1.Application, stage deploymentskyscannernetv1alpha1.ProgressiveRolloutStage) bool {
+
+	return true
+}

internal/utils/utils.go

@@ -2,6 +2,7 @@ package utils
 
 import (
 	argov1alpha1 "github.com/argoproj/argo-cd/pkg/apis/application/v1alpha1"
+	"github.com/argoproj/gitops-engine/pkg/health"
 	corev1 "k8s.io/api/core/v1"
 	"sort"
 )
@@ -21,3 +22,29 @@ func SortSecretsByName(secrets *corev1.SecretList) {
 func SortAppsByName(apps *[]argov1alpha1.Application) {
 	sort.SliceStable(*apps, func(i, j int) bool { return (*apps)[i].Name < (*apps)[j].Name })
 }
+
+// GetAppsBySyncStatusCode returns the Applications matching the specified sync status code
+func GetAppsBySyncStatusCode(apps []argov1alpha1.Application, code argov1alpha1.SyncStatusCode) []argov1alpha1.Application {
+	var result []argov1alpha1.Application
+
+	for _, app := range apps {
+		if app.Status.Sync.Status == code {
+			result = append(result, app)
+		}
+	}
+
+	return result
+}
+
+// GetAppsByHealthStatusCode returns the Applications matching the specified sync status code
+func GetAppsByHealthStatusCode(apps []argov1alpha1.Application, code health.HealthStatusCode) []argov1alpha1.Application {
+	var result []argov1alpha1.Application
+
+	for _, app := range apps {
+		if app.Status.Health.Status == code {
+			result = append(result, app)
+		}
+	}
+
+	return result
+}