selective sync (argoproj#213)

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

Author: kshamajain99

.gitignore

@@ -3,3 +3,4 @@
 .vscode
 .idea
 coverage.out
+vendor/

pkg/sync/sync_context.go

@@ -2,6 +2,7 @@ package sync
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"sort"
 	"strconv"
@@ -23,6 +24,7 @@ import (
 	"k8s.io/klog/v2/klogr"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 
+	"github.com/argoproj/gitops-engine/pkg/diff"
 	"github.com/argoproj/gitops-engine/pkg/health"
 	"github.com/argoproj/gitops-engine/pkg/sync/common"
 	"github.com/argoproj/gitops-engine/pkg/sync/hook"
@@ -129,6 +131,18 @@ func WithPruneLast(enabled bool) SyncOpt {
 	}
 }
 
+// WithResourceModificationChecker sets resource modification result
+func WithResourceModificationChecker(enabled bool, diffResults *diff.DiffResultList) SyncOpt {
+	return func(ctx *syncContext) {
+		ctx.applyOutOfSyncOnly = enabled
+		if enabled {
+			ctx.modificationResult = groupDiffResults(diffResults)
+		} else {
+			ctx.modificationResult = nil
+		}
+	}
+}
+
 // WithNamespaceCreation will create non-exist namespace
 func WithNamespaceCreation(createNamespace bool, namespaceModifier func(*unstructured.Unstructured) bool) SyncOpt {
 	return func(ctx *syncContext) {
@@ -217,6 +231,25 @@ func groupResources(reconciliationResult ReconciliationResult) map[kubeutil.Reso
 	return resources
 }
 
+// generates a map of resource and its modification result based on diffResultList
+func groupDiffResults(diffResultList *diff.DiffResultList) map[kubeutil.ResourceKey]bool {
+	modifiedResources := make(map[kube.ResourceKey]bool)
+	for _, res := range diffResultList.Diffs {
+		var obj unstructured.Unstructured
+		var err error
+		if string(res.NormalizedLive) != "null" {
+			err = json.Unmarshal(res.NormalizedLive, &obj)
+		} else {
+			err = json.Unmarshal(res.PredictedLive, &obj)
+		}
+		if err != nil {
+			continue
+		}
+		modifiedResources[kube.GetResourceKey(&obj)] = res.Modified
+	}
+	return modifiedResources
+}
+
 const (
 	crdReadinessTimeout = time.Duration(3) * time.Second
 )
@@ -281,6 +314,10 @@ type syncContext struct {
 	namespaceModifier func(*unstructured.Unstructured) bool
 
 	syncWaveHook common.SyncWaveHook
+
+	applyOutOfSyncOnly bool
+	// stores whether the resource is modified or not
+	modificationResult map[kube.ResourceKey]bool
 }
 
 func (sc *syncContext) setRunningPhase(tasks []*syncTask, isPendingDeletion bool) {
@@ -516,6 +553,13 @@ func (sc *syncContext) getSyncTasks() (_ syncTasks, successful bool) {
 			continue
 		}
 
+		if sc.applyOutOfSyncOnly {
+			if modified, ok := sc.modificationResult[k]; !modified && ok {
+				sc.log.WithValues("resource key", k).V(1).Info("Skipping as resource was not modified")
+				continue
+			}
+		}
+
 		for _, phase := range syncPhases(obj) {
 			resourceTasks = append(resourceTasks, &syncTask{phase: phase, targetObj: resource.Target, liveObj: resource.Live})
 		}

pkg/sync/sync_context_test.go

@@ -1,6 +1,7 @@
 package sync
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"reflect"
@@ -18,6 +19,7 @@ import (
 	testcore "k8s.io/client-go/testing"
 	"k8s.io/klog/v2/klogr"
 
+	"github.com/argoproj/gitops-engine/pkg/diff"
 	"github.com/argoproj/gitops-engine/pkg/health"
 	synccommon "github.com/argoproj/gitops-engine/pkg/sync/common"
 	"github.com/argoproj/gitops-engine/pkg/utils/kube"
@@ -1056,3 +1058,78 @@ func TestPruneLast(t *testing.T) {
 		assert.Equal(t, 3, tasks.lastWave())
 	})
 }
+
+func diffResultList() *diff.DiffResultList {
+	pod1 := NewPod()
+	pod1.SetName("pod-1")
+	pod2 := NewPod()
+	pod2.SetName("pod-2")
+	pod3 := NewPod()
+	pod3.SetName("pod-3")
+
+	diffResultList := diff.DiffResultList{
+		Modified: true,
+		Diffs:    []diff.DiffResult{},
+	}
+
+	podBytes, _ := json.Marshal(pod1)
+	diffResultList.Diffs = append(diffResultList.Diffs, diff.DiffResult{NormalizedLive: []byte("null"), PredictedLive: podBytes, Modified: true})
+
+	podBytes, _ = json.Marshal(pod2)
+	diffResultList.Diffs = append(diffResultList.Diffs, diff.DiffResult{NormalizedLive: podBytes, PredictedLive: []byte("null"), Modified: true})
+
+	podBytes, _ = json.Marshal(pod3)
+	diffResultList.Diffs = append(diffResultList.Diffs, diff.DiffResult{NormalizedLive: podBytes, PredictedLive: podBytes, Modified: false})
+
+	return &diffResultList
+}
+
+func TestApplyOutOfSyncOnly(t *testing.T) {
+	pod1 := NewPod()
+	pod1.SetName("pod-1")
+	pod2 := NewPod()
+	pod2.SetName("pod-2")
+	pod3 := NewPod()
+	pod3.SetName("pod-3")
+	syncCtx := newTestSyncCtx()
+
+	t.Run("applyOutOfSyncOnly=false", func(t *testing.T) {
+		syncCtx.applyOutOfSyncOnly = true
+		syncCtx.modificationResult = nil
+		syncCtx.resources = groupResources(ReconciliationResult{
+			Live:   []*unstructured.Unstructured{nil, pod2, pod3},
+			Target: []*unstructured.Unstructured{pod1, nil, pod3},
+		})
+		tasks, successful := syncCtx.getSyncTasks()
+
+		assert.True(t, successful)
+		assert.Len(t, tasks, 3)
+	})
+
+	syncCtx = newTestSyncCtx(WithResourceModificationChecker(true, diffResultList()))
+	t.Run("applyOutOfSyncOnly=true", func(t *testing.T) {
+		syncCtx.applyOutOfSyncOnly = true
+		syncCtx.resources = groupResources(ReconciliationResult{
+			Live:   []*unstructured.Unstructured{nil, pod2, pod3},
+			Target: []*unstructured.Unstructured{pod1, nil, pod3},
+		})
+		tasks, successful := syncCtx.getSyncTasks()
+
+		assert.True(t, successful)
+		assert.Len(t, tasks, 2)
+	})
+
+	pod4 := NewPod()
+	pod4.SetName("pod-4")
+	t.Run("applyOutOfSyncOnly=true and missing resource key", func(t *testing.T) {
+		syncCtx.applyOutOfSyncOnly = true
+		syncCtx.resources = groupResources(ReconciliationResult{
+			Live:   []*unstructured.Unstructured{nil, pod2, pod3, pod4},
+			Target: []*unstructured.Unstructured{pod1, nil, pod3, pod4},
+		})
+		tasks, successful := syncCtx.getSyncTasks()
+
+		assert.True(t, successful)
+		assert.Len(t, tasks, 3)
+	})
+}