feat(syncwaves): add DAG ordering for syncwaves

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

Author: SebastienFelix

controller/sync.go

@@ -571,7 +571,7 @@ func hasSharedResourceCondition(app *v1alpha1.Application) (bool, string) {
 // Note, this is not foolproof, since a proper fix would require the CRD record
 // status.observedGeneration coupled with a health.lua that verifies
 // status.observedGeneration == metadata.generation
-func delayBetweenSyncWaves(_ common.SyncPhase, _ int, finalWave bool) error {
+func delayBetweenSyncWaves(_ []common.SyncIdentity, finalWave bool) error {
 	if !finalWave {
 		delaySec := 2
 		if delaySecStr := os.Getenv(EnvVarSyncWaveDelay); delaySecStr != "" {

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
@@ -59,10 +63,16 @@ type PermissionValidator func(un *unstructured.Unstructured, res *metav1.APIReso
 
 type SyncPhase string
 
+type SyncIdentity struct {
+	Phase     SyncPhase
+	Wave      int
+	WaveGroup int
+}
+
 // SyncWaveHook is a callback function which will be invoked after each sync wave is successfully
 // applied during a sync operation. The callback indicates which phase and wave it had just
 // executed, and whether or not that wave was the final one.
-type SyncWaveHook func(phase SyncPhase, wave int, final bool) error
+type SyncWaveHook func(t []SyncIdentity, final bool) error
 
 const (
 	SyncPhasePreSync  = "PreSync"

gitops-engine/pkg/sync/doc.go

@@ -75,6 +75,10 @@ that runs before all other resources. The `argocd.argoproj.io/sync-wave` annotat
 	  annotations:
 	    argocd.argoproj.io/sync-wave: "5"
 
+# Sync Groups
+
+The wave groups allow to define independant/dependent sync processes
+
 # Sync Options
 
 The sync options allows customizing the synchronization of selected resources. The options are specified using the
@@ -89,6 +93,7 @@ How Does It Work Together?
 Syncing process orders the resources in the following precedence:
 
 - The phase
+- The group with respect to group dependencies
 - The wave they are in (lower values first)
 - By kind (e.g. namespaces first)
 - By name

gitops-engine/pkg/sync/sync_context.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -560,26 +561,29 @@ 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()
+	independantSyncIdentities := tasks.independantSyncIdentities()
+	allSyncIdentities := tasks.syncIdentities()
 
 	// 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(independantSyncIdentities, 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 })
+	sc.log.WithValues("phase", phase, "independantSyncIdentities", independantSyncIdentities, "tasks", tasks, "syncFailTasks", syncFailTasks).V(1).Info("Filtering tasks in correct phase and wave")
+	tasks = tasks.Filter(func(t *syncTask) bool { return slices.Contains(independantSyncIdentities, t.identity()) })
 
 	sc.setOperationPhase(common.OperationRunning, "one or more tasks are running")
 
 	sc.log.WithValues("tasks", tasks).V(1).Info("Wet-run")
 	runState := sc.runTasks(tasks, false)
 
 	if sc.syncWaveHook != nil && runState != failed {
-		err := sc.syncWaveHook(phase, wave, finalWave)
+		finalWave := phase == tasks.lastPhase() && len(independantSyncIdentities) == len(allSyncIdentities)
+		err := sc.syncWaveHook(independantSyncIdentities, finalWave)
 		if err != nil {
 			sc.deleteHooks(hooksPendingDeletionFailed)
 			sc.setOperationPhase(common.OperationFailed, fmt.Sprintf("SyncWaveHook failed: %v", err))
@@ -909,52 +913,61 @@ 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
+		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[startWave] {
-			task.waveOverride = &endWave
-		}
+			for _, task := range pruneTasks[startWave] {
+				task.waveOverride = &endWave
+			}
 
-		for _, task := range pruneTasks[endWave] {
-			task.waveOverride = &startWave
+			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_context_test.go

@@ -1793,10 +1793,11 @@ func TestSyncWaveHook(t *testing.T) {
 	syncCtx.hooks = []*unstructured.Unstructured{pod3}
 
 	called := false
-	syncCtx.syncWaveHook = func(phase synccommon.SyncPhase, wave int, final bool) error {
+	syncCtx.syncWaveHook = func(syncIdentities []synccommon.SyncIdentity, final bool) error {
 		called = true
-		assert.Equal(t, synccommon.SyncPhaseSync, string(phase))
-		assert.Equal(t, -1, wave)
+		assert.Equal(t, 1, len(syncIdentities))
+		assert.Equal(t, synccommon.SyncPhaseSync, string(syncIdentities[0].Phase))
+		assert.Equal(t, -1, syncIdentities[0].Wave)
 		assert.False(t, final)
 		return nil
 	}
@@ -1806,7 +1807,7 @@ func TestSyncWaveHook(t *testing.T) {
 	// call sync again, it should not invoke the SyncWaveHook callback since we only should be
 	// doing this after an apply, and not every reconciliation
 	called = false
-	syncCtx.syncWaveHook = func(_ synccommon.SyncPhase, _ int, _ bool) error {
+	syncCtx.syncWaveHook = func(_ []synccommon.SyncIdentity, _ bool) error {
 		called = true
 		return nil
 	}
@@ -1819,10 +1820,11 @@ func TestSyncWaveHook(t *testing.T) {
 	pod1Res.HookPhase = synccommon.OperationSucceeded
 	syncCtx.syncRes[resourceResultKey(pod1Res.ResourceKey, synccommon.SyncPhaseSync)] = pod1Res
 	called = false
-	syncCtx.syncWaveHook = func(phase synccommon.SyncPhase, wave int, final bool) error {
+	syncCtx.syncWaveHook = func(syncIdentities []synccommon.SyncIdentity, final bool) error {
 		called = true
-		assert.Equal(t, synccommon.SyncPhaseSync, string(phase))
-		assert.Equal(t, 0, wave)
+		assert.Equal(t, 1, len(syncIdentities))
+		assert.Equal(t, synccommon.SyncPhaseSync, string(syncIdentities[0].Phase))
+		assert.Equal(t, 0, syncIdentities[0].Wave)
 		assert.False(t, final)
 		return nil
 	}
@@ -1835,17 +1837,135 @@ func TestSyncWaveHook(t *testing.T) {
 	pod2Res.HookPhase = synccommon.OperationSucceeded
 	syncCtx.syncRes[resourceResultKey(pod2Res.ResourceKey, synccommon.SyncPhaseSync)] = pod2Res
 	called = false
-	syncCtx.syncWaveHook = func(phase synccommon.SyncPhase, wave int, final bool) error {
+	syncCtx.syncWaveHook = func(syncIdentities []synccommon.SyncIdentity, final bool) error {
 		called = true
-		assert.Equal(t, synccommon.SyncPhasePostSync, string(phase))
-		assert.Equal(t, 0, wave)
+		assert.Equal(t, 1, len(syncIdentities))
+		assert.Equal(t, synccommon.SyncPhasePostSync, string(syncIdentities[0].Phase))
+		assert.Equal(t, 0, syncIdentities[0].Wave)
 		assert.True(t, final)
 		return nil
 	}
 	syncCtx.Sync()
 	assert.True(t, called)
 }
 
+func TestSyncWaveGroup(t *testing.T) {
+	syncCtx := newTestSyncCtx(nil, WithOperationSettings(false, false, false, false))
+	pod1 := testingutils.NewPod()
+	pod1.SetName("pod-1")
+	pod1.SetAnnotations(map[string]string{synccommon.AnnotationSyncWave: "-1", synccommon.AnnotationSyncWaveGroup: "0", synccommon.AnnotationSyncWaveGroupDependencies: ""})
+	pod2 := testingutils.NewPod()
+	pod2.SetName("pod-2")
+	pod2.SetAnnotations(map[string]string{synccommon.AnnotationSyncWave: "0", synccommon.AnnotationSyncWaveGroup: "0", synccommon.AnnotationSyncWaveGroupDependencies: ""})
+	pod3 := testingutils.NewPod()
+	pod3.SetName("pod-3")
+	pod3.SetAnnotations(map[string]string{synccommon.AnnotationSyncWave: "3", synccommon.AnnotationSyncWaveGroup: "1", synccommon.AnnotationSyncWaveGroupDependencies: ""})
+	pod4 := testingutils.NewPod()
+	pod4.SetName("pod-4")
+	pod4.SetAnnotations(map[string]string{synccommon.AnnotationSyncWave: "1", synccommon.AnnotationSyncWaveGroup: "2", synccommon.AnnotationSyncWaveGroupDependencies: "1"})
+	pod5 := testingutils.NewPod()
+	pod5.SetName("pod-5")
+	pod5.SetAnnotations(map[string]string{synccommon.AnnotationSyncWave: "1", synccommon.AnnotationSyncWaveGroup: "3", synccommon.AnnotationSyncWaveGroupDependencies: "0,1"})
+
+	syncCtx.resources = groupResources(ReconciliationResult{
+		Live:   []*unstructured.Unstructured{nil, nil, nil, nil, nil},
+		Target: []*unstructured.Unstructured{pod1, pod2, pod3, pod4, pod5},
+	})
+
+	called := false
+	syncCtx.syncWaveHook = func(syncIdentities []synccommon.SyncIdentity, final bool) error {
+		called = true
+		assert.Equal(t, 2, len(syncIdentities))
+		assert.Equal(t, synccommon.SyncPhaseSync, string(syncIdentities[0].Phase))
+		assert.Equal(t, -1, syncIdentities[0].Wave)
+		assert.Equal(t, 0, syncIdentities[0].WaveGroup)
+		assert.Equal(t, 3, syncIdentities[1].Wave)
+		assert.Equal(t, 1, syncIdentities[1].WaveGroup)
+		assert.False(t, final)
+		return nil
+	}
+
+	originalSyncTasks, _ := syncCtx.getSyncTasks()
+	assert.Equal(t, 5, len(originalSyncTasks))
+	_, _, originalStates := syncCtx.GetState()
+	assert.Equal(t, 0, len(originalStates))
+
+	// call sync. pod-1 and pod-3 should be processed first. Verify we invoke SyncWaveHook call after applying first waves
+
+	syncCtx.Sync()
+	assert.True(t, called)
+
+	_, _, results := syncCtx.GetState()
+	assert.Equal(t, 2, len(results))
+
+	pod1Res := results[0]
+	pod1Res.HookPhase = synccommon.OperationSucceeded
+	syncCtx.syncRes[resourceResultKey(pod1Res.ResourceKey, synccommon.SyncPhaseSync)] = pod1Res
+
+	pod2Res := results[1]
+	pod2Res.HookPhase = synccommon.OperationSucceeded
+	syncCtx.syncRes[resourceResultKey(pod2Res.ResourceKey, synccommon.SyncPhaseSync)] = pod2Res
+
+	// Here I would like to see :
+	// syncCtx.resources[0].Target and syncCtx.resources[2].Target not nil because pod-1 and pod-3 have been synced
+	// syncTasks should contain 3 elements : pod-2, pod-4 and pod-5
+	// but syncCtx.getSyncTasks() returns 5 elements
+
+	/*####### /!\ THIS TEST SHOULD PASS /!\ #######
+
+		syncTasks2, _ := syncCtx.getSyncTasks()
+		assert.Equal(t, 3, len(syncTasks2))
+
+	########     /!\ BUT IT DOES NOT /!\    #####*/
+
+	// call sync again, pod-2 and pod-4 should be synced during this sync. Verify we invoke SyncWaveHook call after applying second waves
+
+	called = false
+	syncCtx.syncWaveHook = func(_ []synccommon.SyncIdentity, _ bool) error {
+		called = true
+		return nil
+	}
+	syncCtx.Sync()
+	assert.True(t, called)
+
+	_, _, results2 := syncCtx.GetState()
+	assert.Equal(t, 4, len(results2))
+
+	pod3Res := results2[2]
+	pod3Res.HookPhase = synccommon.OperationSucceeded
+	syncCtx.syncRes[resourceResultKey(pod3Res.ResourceKey, synccommon.SyncPhaseSync)] = pod3Res
+
+	pod4Res := results2[3]
+	pod4Res.HookPhase = synccommon.OperationSucceeded
+	syncCtx.syncRes[resourceResultKey(pod4Res.ResourceKey, synccommon.SyncPhaseSync)] = pod4Res
+
+	// complete last wave, then call Sync again. Verify we invoke another SyncWaveHook call after applying last wave
+	called = false
+	syncCtx.syncWaveHook = func(syncIdentities []synccommon.SyncIdentity, final bool) error {
+		called = true
+		return nil
+	}
+	syncCtx.Sync()
+	assert.True(t, called)
+
+	_, _, results3 := syncCtx.GetState()
+	assert.Equal(t, 5, len(results3))
+
+	pod5Res := results3[0]
+	pod5Res.HookPhase = synccommon.OperationSucceeded
+	syncCtx.syncRes[resourceResultKey(pod5Res.ResourceKey, synccommon.SyncPhaseSync)] = pod5Res
+
+	// no remaining wave. Verify we don't invoke another SyncWaveHook call
+
+	called = false
+	syncCtx.syncWaveHook = func(syncIdentities []synccommon.SyncIdentity, final bool) error {
+		called = true
+		return nil
+	}
+	syncCtx.Sync()
+	assert.False(t, called)
+}
+
 func TestSyncWaveHookFail(t *testing.T) {
 	syncCtx := newTestSyncCtx(nil, WithOperationSettings(false, false, false, false))
 	pod1 := testingutils.NewPod()
@@ -1857,7 +1977,7 @@ func TestSyncWaveHookFail(t *testing.T) {
 	})
 
 	called := false
-	syncCtx.syncWaveHook = func(_ synccommon.SyncPhase, _ int, _ bool) error {
+	syncCtx.syncWaveHook = func(_ []synccommon.SyncIdentity, _ bool) error {
 		called = true
 		return errors.New("intentional error")
 	}