add DAG for syncwaves

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

Author: SebastienFelix

gitops-engine/pkg/sync/common/types.go

@@ -12,6 +12,10 @@ const (
 	AnnotationSyncOptions = "argocd.argoproj.io/sync-options"
 	// AnnotationSyncWave indicates which wave of the sync the resource or hook should be in
 	AnnotationSyncWave = "argocd.argoproj.io/sync-wave"
+	// AnnotationSyncWaveGroup indicates which wave of the sync the resource or hook should be in
+	AnnotationSyncWaveGroup = "argocd.argoproj.io/sync-wave-group"
+	// AnnotationSyncWaveGroupDependencies indicates which wave of the sync the resource or hook should be in
+	AnnotationSyncWaveGroupDependencies = "argocd.argoproj.io/sync-wave-group-dependencies"
 	// AnnotationKeyHook contains the hook type of a resource
 	AnnotationKeyHook = "argocd.argoproj.io/hook"
 	// AnnotationKeyHookDeletePolicy is the policy of deleting a hook

gitops-engine/pkg/sync/sync_context.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -560,18 +561,25 @@ func (sc *syncContext) Sync() {
 		return
 	}
 
-	// remove any tasks not in this wave
+	// remove any tasks which have unsynced dependencies
 	phase := tasks.phase()
 	wave := tasks.wave()
-	finalWave := phase == tasks.lastPhase() && wave == tasks.lastWave()
+	syncIdentities := tasks.syncIdentities()
+	allSyncIdentities := make([]syncIdentity, 0)
+	for _, t := range tasks {
+		if !slices.Contains(allSyncIdentities, t.identity()) {
+			allSyncIdentities = append(allSyncIdentities, t.identity())
+		}
+	}
+	finalWave := phase == tasks.lastPhase() && len(syncIdentities) == len(allSyncIdentities)
 
 	// if it is the last phase/wave and the only remaining tasks are non-hooks, the we are successful
 	// EVEN if those objects subsequently degraded
 	// This handles the common case where neither hooks or waves are used and a sync equates to simply an (asynchronous) kubectl apply of manifests, which succeeds immediately.
-	remainingTasks := tasks.Filter(func(t *syncTask) bool { return t.phase != phase || wave != t.wave() || t.isHook() })
+	remainingTasks := tasks.Filter(func(t *syncTask) bool { return !slices.Contains(syncIdentities, t.identity()) || t.isHook() })
 
 	sc.log.WithValues("phase", phase, "wave", wave, "tasks", tasks, "syncFailTasks", syncFailTasks).V(1).Info("Filtering tasks in correct phase and wave")
-	tasks = tasks.Filter(func(t *syncTask) bool { return t.phase == phase && t.wave() == wave })
+	tasks = tasks.Filter(func(t *syncTask) bool { return slices.Contains(syncIdentities, t.identity()) })
 
 	sc.setOperationPhase(common.OperationRunning, "one or more tasks are running")
 
@@ -909,52 +917,62 @@ func (sc *syncContext) getSyncTasks() (_ syncTasks, successful bool) {
 	}
 
 	// for prune tasks, modify the waves for proper cleanup i.e reverse of sync wave (creation order)
-	pruneTasks := make(map[int][]*syncTask)
+
+	tasksByWaveGroup := make(map[int][]*syncTask)
 	for _, task := range tasks {
-		if task.isPrune() {
-			pruneTasks[task.wave()] = append(pruneTasks[task.wave()], task)
-		}
+		tasksByWaveGroup[task.waveGroup()] = append(tasksByWaveGroup[task.waveGroup()], task)
 	}
+	for waveGroup := range tasksByWaveGroup {
+		pruneTasks := make(map[int][]*syncTask)
+		for _, task := range tasksByWaveGroup[waveGroup] {
+			if task.isPrune() {
+				pruneTasks[task.wave()] = append(pruneTasks[task.wave()], task)
+			}
+		}
 
-	var uniquePruneWaves []int
-	for k := range pruneTasks {
-		uniquePruneWaves = append(uniquePruneWaves, k)
-	}
-	sort.Ints(uniquePruneWaves)
+		var uniquePruneWaves []int
+		for k := range pruneTasks {
+			uniquePruneWaves = append(uniquePruneWaves, k)
+		}
+		sort.Ints(uniquePruneWaves)
 
-	// reorder waves for pruning tasks using symmetric swap on prune waves
-	n := len(uniquePruneWaves)
-	for i := 0; i < n/2; i++ {
-		// waves to swap
-		startWave := uniquePruneWaves[i]
-		endWave := uniquePruneWaves[n-1-i]
+		// reorder waves for pruning tasks using symmetric swap on prune waves
 
-		for _, task := range pruneTasks[startWave] {
-			task.waveOverride = &endWave
-		}
+		n := len(uniquePruneWaves)
+		for j := 0; j < n/2; j++ {
+			// waves to swap
+			startWave := uniquePruneWaves[j]
+			endWave := uniquePruneWaves[n-1-j]
 
-		for _, task := range pruneTasks[endWave] {
-			task.waveOverride = &startWave
+			for _, task := range pruneTasks[startWave] {
+				task.waveOverride = &endWave
+			}
+
+			for _, task := range pruneTasks[endWave] {
+				task.waveOverride = &startWave
+			}
 		}
-	}
 
-	// for pruneLast tasks, modify the wave to sync phase last wave of tasks + 1
-	// to ensure proper cleanup, syncPhaseLastWave should also consider prune tasks to determine last wave
-	syncPhaseLastWave := 0
-	for _, task := range tasks {
-		if task.phase == common.SyncPhaseSync {
-			if task.wave() > syncPhaseLastWave {
-				syncPhaseLastWave = task.wave()
+		// for pruneLast tasks, modify the wave to sync phase last wave of tasks + 1
+		// to ensure proper cleanup, syncPhaseLastWave should also consider prune tasks to determine last wave
+
+		syncPhaseLastWave := 0
+		for _, task := range tasksByWaveGroup[waveGroup] {
+			if task.phase == common.SyncPhaseSync {
+				if task.wave() > syncPhaseLastWave {
+					syncPhaseLastWave = task.wave()
+				}
 			}
 		}
-	}
-	syncPhaseLastWave = syncPhaseLastWave + 1
+		syncPhaseLastWave = syncPhaseLastWave + 1
 
-	for _, task := range tasks {
-		if task.isPrune() &&
-			(sc.pruneLast || resourceutil.HasAnnotationOption(task.liveObj, common.AnnotationSyncOptions, common.SyncOptionPruneLast)) {
-			task.waveOverride = &syncPhaseLastWave
+		for _, task := range tasksByWaveGroup[waveGroup] {
+			if task.isPrune() &&
+				(sc.pruneLast || resourceutil.HasAnnotationOption(task.liveObj, common.AnnotationSyncOptions, common.SyncOptionPruneLast)) {
+				task.waveOverride = &syncPhaseLastWave
+			}
 		}
+
 	}
 
 	tasks.Sort()

gitops-engine/pkg/sync/sync_identity.go

@@ -0,0 +1,14 @@
+package sync
+
+import (
+	"github.com/argoproj/gitops-engine/pkg/sync/common"
+)
+
+// syncTask holds the live and target object. At least one should be non-nil. A targetObj of nil
+// indicates the live object needs to be pruned. A liveObj of nil indicates the object has yet to
+// be deployed
+type syncIdentity struct {
+	phase     common.SyncPhase
+	wave      int
+	waveGroup int
+}

gitops-engine/pkg/sync/sync_task.go

@@ -2,6 +2,8 @@ package sync
 
 import (
 	"fmt"
+	"strconv"
+	"strings"
 
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -16,14 +18,16 @@ import (
 // indicates the live object needs to be pruned. A liveObj of nil indicates the object has yet to
 // be deployed
 type syncTask struct {
-	phase          common.SyncPhase
-	liveObj        *unstructured.Unstructured
-	targetObj      *unstructured.Unstructured
-	skipDryRun     bool
-	syncStatus     common.ResultCode
-	operationState common.OperationPhase
-	message        string
-	waveOverride   *int
+	phase                         common.SyncPhase
+	liveObj                       *unstructured.Unstructured
+	targetObj                     *unstructured.Unstructured
+	skipDryRun                    bool
+	syncStatus                    common.ResultCode
+	operationState                common.OperationPhase
+	message                       string
+	waveOverride                  *int
+	waveGroupOverride             *int
+	waveGroupDependenciesOverride *string
 }
 
 func ternary(val bool, a, b string) string {
@@ -63,6 +67,35 @@ func (t *syncTask) wave() int {
 	return syncwaves.Wave(t.obj())
 }
 
+func (t *syncTask) waveGroup() int {
+	if t.waveGroupOverride != nil {
+		return *t.waveGroupOverride
+	}
+	return syncwaves.WaveGroup(t.obj())
+}
+
+func (t *syncTask) identity() syncIdentity {
+	return syncIdentity{t.phase, t.wave(), t.waveGroup()}
+}
+
+func (t *syncTask) waveGroupDependencies() []int {
+	if t.waveGroupDependenciesOverride != nil {
+		waveGroup := t.waveGroup()
+		stringWaveGroupDependencies := strings.Split(*t.waveGroupDependenciesOverride, ",")
+		waveGroupDependencies := make([]int, 0)
+		for _, t := range stringWaveGroupDependencies {
+			waveGroupDependency, err := strconv.Atoi(t)
+			if err == nil {
+				if waveGroupDependency < waveGroup {
+					waveGroupDependencies = append(waveGroupDependencies, waveGroupDependency)
+				}
+			}
+		}
+		return waveGroupDependencies
+	}
+	return syncwaves.WaveGroupDependencies(t.obj())
+}
+
 func (t *syncTask) isHook() bool {
 	return hook.IsHook(t.obj())
 }

gitops-engine/pkg/sync/sync_tasks.go

@@ -2,6 +2,7 @@ package sync
 
 import (
 	"fmt"
+	"slices"
 	"sort"
 	"strings"
 
@@ -79,9 +80,10 @@ func (s syncTasks) Swap(i, j int) {
 
 // order is
 // 1. phase
-// 2. wave
-// 3. kind
-// 4. name
+// 2. waveGroup
+// 3. wave
+// 4. kind
+// 5. name
 func (s syncTasks) Less(i, j int) bool {
 	tA := s[i]
 	tB := s[j]
@@ -91,6 +93,48 @@ func (s syncTasks) Less(i, j int) bool {
 		return d < 0
 	}
 
+	d = tA.waveGroup() - tB.waveGroup()
+	if d != 0 {
+		return d < 0
+	}
+
+	d = tA.wave() - tB.wave()
+	if d != 0 {
+		return d < 0
+	}
+
+	a := tA.obj()
+	b := tB.obj()
+
+	// we take advantage of the fact that if the kind is not in the kindOrder map,
+	// then it will return the default int value of zero, which is the highest value
+	d = kindOrder[a.GetKind()] - kindOrder[b.GetKind()]
+	if d != 0 {
+		return d < 0
+	}
+
+	return a.GetName() < b.GetName()
+}
+
+func (s syncTasks) LessSyncTask(tA, tB syncTask) bool {
+
+	d := syncPhaseOrder[tA.phase] - syncPhaseOrder[tB.phase]
+	if d != 0 {
+		return d < 0
+	}
+
+	if slices.Contains(tB.waveGroupDependencies(), tA.waveGroup()) {
+		return true
+	}
+
+	if slices.Contains(tA.waveGroupDependencies(), tB.waveGroup()) {
+		return false
+	}
+
+	if tA.waveGroup() != tB.waveGroup() {
+		return false
+	}
+
 	d = tA.wave() - tB.wave()
 	if d != 0 {
 		return d < 0
@@ -250,20 +294,66 @@ func (s syncTasks) phase() common.SyncPhase {
 	return ""
 }
 
+func (s syncTasks) waveGroups() []int {
+	if len(s) > 0 {
+		firstWaveGroups := make([]int, 0)
+		for i, firstSyncTask := range s {
+			belongsToFirstWaveGroups := true
+			for _, secondSyncTask := range s[:i] {
+				if s.LessSyncTask(*secondSyncTask, *firstSyncTask) {
+					belongsToFirstWaveGroups = false
+				}
+			}
+			if belongsToFirstWaveGroups && !slices.Contains(firstWaveGroups, s[i].waveGroup()) {
+				firstWaveGroups = append(firstWaveGroups, s[i].waveGroup())
+			}
+		}
+		return firstWaveGroups
+	}
+	return make([]int, 0)
+}
+
 func (s syncTasks) wave() int {
 	if len(s) > 0 {
 		return s[0].wave()
 	}
 	return 0
 }
 
+func (s syncTasks) syncIdentities() []syncIdentity {
+	if len(s) > 0 {
+		firstSyncIdentities := make([]syncIdentity, 0)
+		for i, firstSyncTask := range s {
+			belongsToFirstWaveGroups := true
+			for _, secondSyncTask := range s[:i] {
+				if s.LessSyncTask(*secondSyncTask, *firstSyncTask) {
+					belongsToFirstWaveGroups = false
+				}
+			}
+			wi := &syncIdentity{s[i].phase, s[i].wave(), s[i].waveGroup()}
+			if belongsToFirstWaveGroups && !slices.Contains(firstSyncIdentities, *wi) {
+				firstSyncIdentities = append(firstSyncIdentities, *wi)
+			}
+		}
+		return firstSyncIdentities
+	}
+	return make([]syncIdentity, 0)
+}
+
 func (s syncTasks) lastPhase() common.SyncPhase {
 	if len(s) > 0 {
 		return s[len(s)-1].phase
 	}
 	return ""
 }
 
+func (s syncTasks) lastGroup() int {
+	if len(s) > 0 {
+		return s[len(s)-1].waveGroup()
+	}
+	return 0
+}
+
 func (s syncTasks) lastWave() int {
 	if len(s) > 0 {
 		return s[len(s)-1].wave()