fix: sync should apply Namespaces and CRDs before resources that depend on them (argoproj#225)

* fix: sync should apply Namespaces and CRDs before resources that depend on them

Signed-off-by: Alexander Matyushentsev <[email protected]>

Author: Alexander Matyushentsev

pkg/sync/sync_context.go

@@ -625,16 +625,6 @@ func (sc *syncContext) getSyncTasks() (_ syncTasks, successful bool) {
 		task.liveObj = sc.liveObj(task.targetObj)
 	}
 
-	// enrich tasks with the result
-	for _, task := range tasks {
-		result, ok := sc.syncRes[task.resultKey()]
-		if ok {
-			task.syncStatus = result.Status
-			task.operationState = result.HookPhase
-			task.message = result.Message
-		}
-	}
-
 	// check permissions
 	for _, task := range tasks {
 		serverRes, err := kube.ServerResourceForGroupVersionKind(sc.disco, task.groupVersionKind())
@@ -683,7 +673,17 @@ func (sc *syncContext) getSyncTasks() (_ syncTasks, successful bool) {
 		}
 	}
 
-	sort.Sort(tasks)
+	tasks.Sort()
+
+	// finally enrich tasks with the result
+	for _, task := range tasks {
+		result, ok := sc.syncRes[task.resultKey()]
+		if ok {
+			task.syncStatus = result.Status
+			task.operationState = result.HookPhase
+			task.message = result.Message
+		}
+	}
 
 	return tasks, successful
 }

pkg/sync/sync_task.go

@@ -23,22 +23,7 @@ type syncTask struct {
 	syncStatus     common.ResultCode
 	operationState common.OperationPhase
 	message        string
-}
-
-// isDependsOn returns true if given task depends on current task and should be executed after
-func (t *syncTask) isDependsOn(other *syncTask) bool {
-	otherObj := other.obj()
-	thisGVK := t.obj().GroupVersionKind()
-	otherGVK := otherObj.GroupVersionKind()
-
-	if isCRDOfGroupKind(thisGVK.Group, thisGVK.Kind, otherObj) {
-		return true
-	}
-
-	if otherGVK.Group == "" && otherGVK.Kind == kube.NamespaceKind && otherObj.GetName() == t.obj().GetNamespace() {
-		return true
-	}
-	return false
+	waveOverride   *int
 }
 
 func ternary(val bool, a, b string) string {
@@ -73,6 +58,9 @@ func (t *syncTask) obj() *unstructured.Unstructured {
 }
 
 func (t *syncTask) wave() int {
+	if t.waveOverride != nil {
+		return *t.waveOverride
+	}
 	return syncwaves.Wave(t.obj())
 }
 

pkg/sync/sync_tasks.go

@@ -1,9 +1,14 @@
 package sync
 
 import (
+	"fmt"
+	"sort"
 	"strings"
 
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+
 	"github.com/argoproj/gitops-engine/pkg/sync/common"
+	"github.com/argoproj/gitops-engine/pkg/utils/kube"
 )
 
 // kindOrder represents the correct order of Kubernetes resources within a manifest
@@ -74,12 +79,6 @@ func (s syncTasks) Less(i, j int) bool {
 	tA := s[i]
 	tB := s[j]
 
-	if tA.isDependsOn(tB) {
-		return false
-	} else if tB.isDependsOn(tA) {
-		return true
-	}
-
 	d := syncPhaseOrder[tA.phase] - syncPhaseOrder[tB.phase]
 	if d != 0 {
 		return d < 0
@@ -103,6 +102,73 @@ func (s syncTasks) Less(i, j int) bool {
 	return a.GetName() < b.GetName()
 }
 
+func (s syncTasks) Sort() {
+	sort.Sort(s)
+	// make sure namespaces are created before resources referencing namespaces
+	s.adjustDeps(func(obj *unstructured.Unstructured) (string, bool) {
+		return obj.GetName(), obj.GetKind() == kube.NamespaceKind && obj.GroupVersionKind().Group == ""
+	}, func(obj *unstructured.Unstructured) (string, bool) {
+		return obj.GetNamespace(), obj.GetNamespace() != ""
+	})
+	// make sure CRDs are created before CRs
+	s.adjustDeps(func(obj *unstructured.Unstructured) (string, bool) {
+		if kube.IsCRD(obj) {
+			crdGroup, ok, err := unstructured.NestedString(obj.Object, "spec", "group")
+			if err != nil || !ok {
+				return "", false
+			}
+			crdKind, ok, err := unstructured.NestedString(obj.Object, "spec", "names", "kind")
+			if err != nil || !ok {
+				return "", false
+			}
+			return fmt.Sprintf("%s/%s", crdGroup, crdKind), true
+		}
+		return "", false
+	}, func(obj *unstructured.Unstructured) (string, bool) {
+		gk := obj.GroupVersionKind()
+		return fmt.Sprintf("%s/%s", gk.Group, gk.Kind), true
+	})
+}
+
+// adjust order of tasks and bubble up tasks which are dependencies of other tasks
+// (e.g. namespace sync should happen before resources that resides in that namespace)
+func (s syncTasks) adjustDeps(isDep func(obj *unstructured.Unstructured) (string, bool), doesRefDep func(obj *unstructured.Unstructured) (string, bool)) {
+	// store dependency key and first occurrence of resource referencing the dependency
+	firstIndexByDepKey := map[string]int{}
+
+	for i, t := range s {
+		if t.targetObj == nil {
+			continue
+		}
+
+		if depKey, ok := isDep(t.targetObj); ok {
+			// if tasks is a dependency then insert if before first task that reference it
+			if index, ok := firstIndexByDepKey[depKey]; ok {
+				// wave and sync phase of dependency resource must be same as wave and phase of resource that depend on it
+				wave := s[index].wave()
+				t.waveOverride = &wave
+				t.phase = s[index].phase
+
+				for j := i; j > index; j-- {
+					s[j] = s[j-1]
+				}
+				s[index] = t
+				// increase previously collected indexes by 1
+				for ns, firstIndex := range firstIndexByDepKey {
+					if firstIndex >= index {
+						firstIndexByDepKey[ns] = firstIndex + 1
+					}
+				}
+			}
+		} else if depKey, ok := doesRefDep(t.targetObj); ok {
+			// if task is referencing the dependency then store first index of it
+			if _, ok := firstIndexByDepKey[depKey]; !ok {
+				firstIndexByDepKey[depKey] = i
+			}
+		}
+	}
+}
+
 func (s syncTasks) Filter(predicate func(task *syncTask) bool) (tasks syncTasks) {
 	for _, task := range s {
 		if predicate(task) {

pkg/sync/sync_tasks_test.go

@@ -33,7 +33,6 @@ func TestAnySyncTasks(t *testing.T) {
 		return task.name() == "does-not-exist"
 	})
 	assert.False(t, res)
-
 }
 
 func TestAllSyncTasks(t *testing.T) {
@@ -369,37 +368,76 @@ func TestSyncNamespaceAgainstCRD(t *testing.T) {
 }
 
 func TestSyncTasksSort_NamespaceAndObjectInNamespace(t *testing.T) {
-	deployment := &syncTask{
+	hook1 := &syncTask{
 		phase: common.SyncPhasePreSync,
 		targetObj: &unstructured.Unstructured{
 			Object: map[string]interface{}{
 				"kind": "Job",
 				"metadata": map[string]interface{}{
-					"namespace": "myNamespace",
-					"name":      "mySyncHookJob",
+					"namespace": "myNamespace1",
+					"name":      "mySyncHookJob1",
 				},
 			},
 		}}
-	namespace := &syncTask{
+	hook2 := &syncTask{
+		phase: common.SyncPhasePreSync,
+		targetObj: &unstructured.Unstructured{
+			Object: map[string]interface{}{
+				"kind": "Job",
+				"metadata": map[string]interface{}{
+					"namespace": "myNamespace2",
+					"name":      "mySyncHookJob2",
+				},
+			},
+		}}
+	namespace1 := &syncTask{
 		targetObj: &unstructured.Unstructured{
 			Object: map[string]interface{}{
 				"kind": "Namespace",
 				"metadata": map[string]interface{}{
-					"name": "myNamespace",
+					"name": "myNamespace1",
+					"annotations": map[string]string{
+						"argocd.argoproj.io/sync-wave": "1",
+					},
+				},
+			},
+		},
+	}
+	namespace2 := &syncTask{
+		targetObj: &unstructured.Unstructured{
+			Object: map[string]interface{}{
+				"kind": "Namespace",
+				"metadata": map[string]interface{}{
+					"name": "myNamespace2",
+					"annotations": map[string]string{
+						"argocd.argoproj.io/sync-wave": "2",
+					},
 				},
 			},
 		},
 	}
 
-	unsorted := syncTasks{deployment, namespace}
-	sort.Sort(unsorted)
+	unsorted := syncTasks{hook1, hook2, namespace1, namespace2}
+	unsorted.Sort()
 
-	assert.Equal(t, syncTasks{namespace, deployment}, unsorted)
+	assert.Equal(t, syncTasks{namespace1, hook1, namespace2, hook2}, unsorted)
+	assert.Equal(t, 0, namespace1.wave())
+	assert.Equal(t, common.SyncPhase(common.SyncPhasePreSync), namespace1.phase)
+	assert.Equal(t, 0, namespace2.wave())
+	assert.Equal(t, common.SyncPhase(common.SyncPhasePreSync), namespace2.phase)
 }
 
 func TestSyncTasksSort_CRDAndCR(t *testing.T) {
-	crd := &syncTask{
+	cr := &syncTask{
 		phase: common.SyncPhasePreSync,
+		targetObj: &unstructured.Unstructured{
+			Object: map[string]interface{}{
+				"kind":       "Workflow",
+				"apiVersion": "argoproj.io/v1",
+			},
+		},
+	}
+	crd := &syncTask{
 		targetObj: &unstructured.Unstructured{
 			Object: map[string]interface{}{
 				"apiVersion": "apiextensions.k8s.io/v1",
@@ -412,17 +450,9 @@ func TestSyncTasksSort_CRDAndCR(t *testing.T) {
 				},
 			},
 		}}
-	cr := &syncTask{
-		targetObj: &unstructured.Unstructured{
-			Object: map[string]interface{}{
-				"kind":       "Workflow",
-				"apiVersion": "argoproj.io/v1",
-			},
-		},
-	}
 
 	unsorted := syncTasks{cr, crd}
-	sort.Sort(unsorted)
+	unsorted.Sort()
 
 	assert.Equal(t, syncTasks{crd, cr}, unsorted)
 }