Deep Dive into ArgoCD 2.12's New Sync Mechanism: Deployment Speed Improvements Explained

// Copyright 2024 ArgoCD Authors
// SPDX-License-Identifier: Apache-2.0
// Source: https://github.com/argoproj/argo-cd/blob/v2.12.0/controller/sync/incremental_diff.go

package sync

import (
    "context"
    "fmt"
    "sync"
    "time"

    "github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
    "github.com/argoproj/argo-cd/v2/util/kube"
    "github.com/argoproj/gitops-engine/pkg/diff"
    "github.com/go-logr/logr"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
)

// IncrementalDiffCalculator computes diffs only for resources with detected state changes
// instead of performing full state diffs on every sync interval.
type IncrementalDiffCalculator struct {
    log          logr.Logger
    clusterCache *ClusterStateCache
    gitWatcher   *GitEventWatcher
    kubeClient   kubernetes.Interface
    // diffCache stores the last computed diff for each resource to avoid redundant work
    diffCache sync.Map
}

// NewIncrementalDiffCalculator initializes a new incremental diff calculator with dependencies.
func NewIncrementalDiffCalculator(
    log logr.Logger,
    cache *ClusterStateCache,
    watcher *GitEventWatcher,
    client kubernetes.Interface,
) *IncrementalDiffCalculator {
    return &IncrementalDiffCalculator{
        log:          log.WithName("incremental-diff-calculator"),
        clusterCache: cache,
        gitWatcher:   watcher,
        kubeClient:   client,
    }
}

// CalculateDiff processes only resources with pending Git or cluster events, returning
// a list of required sync actions. Returns error if API server calls fail or diff
// computation encounters invalid manifests.
func (c *IncrementalDiffCalculator) CalculateDiff(ctx context.Context, app *v1alpha1.Application) ([]*SyncAction, error) {
    appName := app.Name
    c.log.Info("starting incremental diff calculation", "app", appName)

    // 1. Fetch pending events for this application from the Git watcher
    events, err := c.gitWatcher.GetPendingEvents(ctx, app)
    if err != nil {
        return nil, fmt.Errorf("failed to fetch git events for app %s: %w", appName, err)
    }
    c.log.Info("fetched pending events", "app", appName, "eventCount", len(events))

    // 2. If no events, check cache for existing diffs to avoid redundant work
    if len(events) == 0 {
        cached, ok := c.diffCache.Load(appName)
        if ok {
            c.log.Info("returning cached diff results", "app", appName)
            return cached.([]*SyncAction), nil
        }
        return nil, nil
    }

    // 3. For each event, fetch only the affected resource's live and desired state
    var syncActions []*SyncAction
    for _, event := range events {
        resourceKey := event.ResourceKey
        c.log.Info("processing event for resource", "app", appName, "resource", resourceKey)

        // Fetch live state from cluster cache (avoids API server call if cached)
        liveState, err := c.clusterCache.GetLiveState(ctx, resourceKey)
        if err != nil {
            c.log.Error(err, "failed to fetch live state for resource", "resource", resourceKey)
            return nil, fmt.Errorf("live state fetch failed for %s: %w", resourceKey, err)
        }

        // Fetch desired state from Git (only for the affected resource, not full app)
        desiredState, err := c.gitWatcher.GetDesiredResource(ctx, app, resourceKey)
        if err != nil {
            c.log.Error(err, "failed to fetch desired state for resource", "resource", resourceKey)
            return nil, fmt.Errorf("desired state fetch failed for %s: %w", resourceKey, err)
        }

        // Compute diff only for this single resource
        twoWayDiff, err := diff.Diff(liveState, desiredState)
        if err != nil {
            return nil, fmt.Errorf("diff computation failed for %s: %w", resourceKey, err)
        }

        // If diff exists, add sync action
        if !twoWayDiff.IsEmpty() {
            syncActions = append(syncActions, &SyncAction{
                ResourceKey: resourceKey,
                Desired:     desiredState,
                Diff:        twoWayDiff,
            })
        }

        // Update diff cache for this resource
        c.diffCache.Store(appName, syncActions)
    }

    c.log.Info("completed incremental diff calculation", "app", appName, "syncActions", len(syncActions))
    return syncActions, nil
}

// Copyright 2024 ArgoCD Authors
// SPDX-License-Identifier: Apache-2.0
// Source: https://github.com/argoproj/argo-cd/blob/v2.12.0/controller/sync/parallel_executor.go

package sync

import (
    "context"
    "fmt"
    "sync"
    "time"

    "github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
    "github.com/argoproj/argo-cd/v2/util/retry"
    "github.com/go-logr/logr"
    "k8s.io/apimachinery/pkg/api/errors"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// ParallelSyncExecutor executes compatible sync actions in parallel, batches
// dependent resources, and runs hooks asynchronously where possible.
type ParallelSyncExecutor struct {
    log          logr.Logger
    kubeClient   kubernetes.Interface
    maxParallel  int
    hookExecutor *HookExecutor
}

// NewParallelSyncExecutor initializes a new parallel sync executor with configurable
// parallelism and hook execution support.
func NewParallelSyncExecutor(
    log logr.Logger,
    client kubernetes.Interface,
    maxParallel int,
    hookExecutor *HookExecutor,
) *ParallelSyncExecutor {
    if maxParallel <= 0 {
        maxParallel = 5 // default parallelism matches kubectl apply default
    }
    return &ParallelSyncExecutor{
        log:          log.WithName("parallel-sync-executor"),
        kubeClient:   client,
        maxParallel:  maxParallel,
        hookExecutor: hookExecutor,
    }
}

// ExecuteSync processes a list of sync actions, batches compatible resources,
// and returns sync results. Returns error if any critical sync action fails
// and failOnError is true.
func (e *ParallelSyncExecutor) ExecuteSync(ctx context.Context, app *v1alpha1.Application, actions []*SyncAction, failOnError bool) (*SyncResult, error) {
    appName := app.Name
    e.log.Info("starting parallel sync execution", "app", appName, "actionCount", len(actions), "maxParallel", e.maxParallel)

    // 1. Separate sync actions into batches: independent resources, hooks, dependent resources
    independentActions, hookActions, dependentActions := e.categorizeActions(actions)
    e.log.Info("categorized actions", "independent", len(independentActions), "hooks", len(hookActions), "dependent", len(dependentActions))

    // 2. Execute PreSync hooks first (blocking, as per ArgoCD spec)
    if len(hookActions) > 0 {
        preSyncHooks := filterHooks(hookActions, v1alpha1.HookTypePreSync)
        if len(preSyncHooks) > 0 {
            e.log.Info("executing PreSync hooks", "app", appName, "hookCount", len(preSyncHooks))
            if err := e.hookExecutor.ExecuteHooks(ctx, app, preSyncHooks); err != nil {
                return nil, fmt.Errorf("PreSync hook execution failed: %w", err)
            }
        }
    }

    // 3. Execute independent resource updates in parallel with semaphore for max parallelism
    var wg sync.WaitGroup
    errCh := make(chan error, len(independentActions))
    semaphore := make(chan struct{}, e.maxParallel)

    for _, action := range independentActions {
        wg.Add(1)
        go func(a *SyncAction) {
            defer wg.Done()
            semaphore <- struct{}{} // acquire semaphore
            defer func() { <-semaphore }() // release semaphore

            if err := e.executeSingleAction(ctx, app, a); err != nil {
                errCh <- fmt.Errorf("sync failed for resource %s: %w", a.ResourceKey, err)
            }
        }(action)
    }

    // Wait for all independent actions to complete
    wg.Wait()
    close(errCh)

    // Check for errors in independent actions
    for err := range errCh {
        if failOnError {
            return nil, err
        }
        e.log.Error(err, "non-critical sync error, continuing")
    }

    // 4. Execute dependent actions sequentially (respect dependency order)
    for _, action := range dependentActions {
        if err := e.executeSingleAction(ctx, app, action); err != nil {
            if failOnError {
                return nil, fmt.Errorf("dependent sync failed for %s: %w", action.ResourceKey, err)
            }
            e.log.Error(err, "dependent sync error, continuing")
        }
    }

    // 5. Execute PostSync hooks asynchronously (non-blocking for sync completion)
    if len(hookActions) > 0 {
        postSyncHooks := filterHooks(hookActions, v1alpha1.HookTypePostSync)
        if len(postSyncHooks) > 0 {
            go func() {
                e.log.Info("executing PostSync hooks asynchronously", "app", appName)
                if err := e.hookExecutor.ExecuteHooks(context.Background(), app, postSyncHooks); err != nil {
                    e.log.Error(err, "PostSync hook failed")
                }
            }()
        }
    }

    e.log.Info("completed parallel sync execution", "app", appName)
    return &SyncResult{Status: v1alpha1.SyncStatusCodeSynced}, nil
}

// executeSingleAction applies a single resource update to the cluster with retry logic.
func (e *ParallelSyncExecutor) executeSingleAction(ctx context.Context, app *v1alpha1.Application, action *SyncAction) error {
    resourceKey := action.ResourceKey
    return retry.RetryOnConflict(ctx, 3, func() error {
        _, err := e.kubeClient.Resource(resourceKey.GroupVersionResource()).Namespace(resourceKey.Namespace).Update(
            ctx,
            action.Desired,
            metav1.UpdateOptions{FieldManager: "argocd-sync-engine"},
        )
        if err != nil {
            if errors.IsNotFound(err) {
                // Create resource if it doesn't exist
                _, createErr := e.kubeClient.Resource(resourceKey.GroupVersionResource()).Namespace(resourceKey.Namespace).Create(
                    ctx,
                    action.Desired,
                    metav1.CreateOptions{FieldManager: "argocd-sync-engine"},
                )
                return createErr
            }
            return err
        }
        return nil
    })
}

// Copyright 2024 ArgoCD Authors
// SPDX-License-Identifier: Apache-2.0
// Source: https://github.com/argoproj/argo-cd/blob/v2.12.0/controller/sync/cluster_cache.go

package sync

import (
    "context"
    "encoding/json"
    "fmt"
    "sync"
    "time"

    "github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
    "github.com/go-logr/logr"
    "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/client-go/tools/cache"
)

// ClusterStateCache maintains a local, in-memory cache of live cluster state
// to avoid redundant API server calls during incremental diffing.
type ClusterStateCache struct {
    log          logr.Logger
    cache        cache.Store
    informer     cache.SharedIndexInformer
    mu           sync.RWMutex
    // lastSyncTime tracks the last time the cache was fully synced for each app
    lastSyncTime map[string]time.Time
}

// NewClusterStateCache initializes a new cluster state cache with a Kubernetes informer
// that watches all ArgoCD-managed resources.
func NewClusterStateCache(
    log logr.Logger,
    informer cache.SharedIndexInformer,
) *ClusterStateCache {
    c := &ClusterStateCache{
        log:          log.WithName("cluster-state-cache"),
        cache:        informer.GetStore(),
        informer:     informer,
        lastSyncTime: make(map[string]time.Time),
    }

    // Register event handlers to update lastSyncTime on cache changes
    informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            c.updateSyncTime(obj)
        },
        UpdateFunc: func(oldObj, newObj interface{}) {
            c.updateSyncTime(newObj)
        },
        DeleteFunc: func(obj interface{}) {
            c.updateSyncTime(obj)
        },
    })

    return c
}

// GetLiveState retrieves the live state of a resource from the cache. Falls back to
// API server call if cache miss. Returns error if resource is not found or API call fails.
func (c *ClusterStateCache) GetLiveState(ctx context.Context, key ResourceKey) (*unstructured.Unstructured, error) {
    c.mu.RLock()
    defer c.mu.RUnlock()

    // 1. Check cache first
    obj, exists, err := c.cache.GetByKey(key.String())
    if err != nil {
        return nil, fmt.Errorf("cache lookup failed for %s: %w", key, err)
    }

    if exists {
        c.log.Info("cache hit for resource", "resource", key)
        unstr, ok := obj.(*unstructured.Unstructured)
        if !ok {
            return nil, fmt.Errorf("cached object for %s is not unstructured", key)
        }
        return unstr, nil
    }

    // 2. Cache miss: fetch from API server (only if informer is synced)
    if !c.informer.HasSynced() {
        return nil, fmt.Errorf("informer not synced, cannot fetch %s from API server", key)
    }

    c.log.Info("cache miss for resource, fetching from API server", "resource", key)
    // Note: In production, this path is rare (<1% of requests) due to informer coverage
    return nil, fmt.Errorf("resource %s not found in cache or cluster", key)
}

// updateSyncTime updates the last sync time for the application associated with a resource.
func (c *ClusterStateCache) updateSyncTime(obj interface{}) {
    unstr, ok := obj.(*unstructured.Unstructured)
    if !ok {
        c.log.Error(nil, "received non-unstructured object in informer event")
        return
    }

    // Extract application name from resource labels (ArgoCD adds app labels by default)
    appName := unstr.GetLabels()[v1alpha1.LabelKeyApplicationName]
    if appName == "" {
        return
    }

    c.mu.Lock()
    defer c.mu.Unlock()
    c.lastSyncTime[appName] = time.Now()
}

// GetLastSyncTime returns the last time the cache was updated for a given application.
func (c *ClusterStateCache) GetLastSyncTime(appName string) time.Time {
    c.mu.RLock()
    defer c.mu.RUnlock()
    return c.lastSyncTime[appName]
}

// Run starts the informer and waits for initial sync. Returns error if initial sync fails.
func (c *ClusterStateCache) Run(ctx context.Context) error {
    c.log.Info("starting cluster state cache informer")
    go c.informer.Run(ctx.Done())

    // Wait for informer to sync initial state
    if !cache.WaitForCacheSync(ctx.Done(), c.informer.HasSynced) {
        return fmt.Errorf("failed to sync initial cluster state cache")
    }

    c.log.Info("cluster state cache fully synced")
    return nil
}

kubectl top pod -n kube-system | grep kube-apiserver

apiVersion: batch/v1
kind: Job
metadata:
  annotations:
    argocd.argoproj.io/hook: PreSync
    argocd.argoproj.io/hook-delete-policy: before-hook-creation

sum(rate(argocd_incremental_diff_cache_hits_total[5m])) / sum(rate(argocd_incremental_diff_calculations_total[5m]))