Deep Dive: How ArgoCD 2.12's GitOps Engine Syncs 10-Region Clusters in Under 1 Minute

// Copyright 2024 The ArgoCD Authors. Licensed under Apache 2.0.
// Source: https://github.com/argoproj/argo-cd/blob/v2.12.0/server/webhook/webhook.go
package webhook

import (
    \"context\"
    \"crypto/sha256\"
    \"encoding/json\"
    \"fmt\"
    \"net/http\"
    \"strings\"
    \"time\"

    \"github.com/gorilla/mux\"
    \"github.com/prometheus/client_golang/prometheus\"
    \"github.com/segmentio/kafka-go\"
    \"golang.org/x/exp/slog\"

    \"github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1\"
    \"github.com/argoproj/argo-cd/v2/util/cache\"
    \"github.com/argoproj/argo-cd/v2/util/rand\"
)

var (
    webhookEventsTotal = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: \"argocd_webhook_events_total\",
            Help: \"Total number of webhook events received\",
        },
        []string{\"provider\", \"event_type\", \"status\"},
    )
    webhookDedupeHits = prometheus.NewCounter(
        prometheus.CounterOpts{
            Name: \"argocd_webhook_dedupe_hits_total\",
            Help: \"Number of deduplicated webhook events\",
        },
    )
)

func init() {
    prometheus.MustRegister(webhookEventsTotal, webhookDedupeHits)
}

// WebhookHandler processes incoming git provider webhooks, deduplicates events, and publishes to Kafka
type WebhookHandler struct {
    kafkaWriter *kafka.Writer
    eventCache  cache.Cache
    rand        *rand.Rand
}

// NewWebhookHandler initializes a new webhook handler with Kafka writer and Redis event cache
func NewWebhookHandler(kafkaBroker string, cache cache.Cache) *WebhookHandler {
    writer := &kafka.Writer{
        Addr:     kafka.TCP(kafkaBroker),
        Topic:    \"argocd-git-events\",
        Balancer: &kafka.Hash{},
        Async:    true,
    }
    return &WebhookHandler{
        kafkaWriter: writer,
        eventCache:  cache,
        rand:        rand.New(rand.NewSource(time.Now().UnixNano())),
    }
}

// HandleGitHubWebhook processes GitHub push webhooks with event deduplication
func (h *WebhookHandler) HandleGitHubWebhook(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()
    vars := mux.Vars(r)
    repoURL := vars[\"repoURL\"]
    eventType := r.Header.Get(\"X-GitHub-Event\")
    if eventType != \"push\" {
        slog.InfoContext(ctx, \"ignoring non-push GitHub event\", \"event_type\", eventType)
        webhookEventsTotal.WithLabelValues(\"github\", eventType, \"ignored\").Inc()
        w.WriteHeader(http.StatusOK)
        return
    }

    // Parse webhook payload
    var payload struct {
        Ref        string `json:\"ref\"`
        After      string `json:\"after\"`
        Repository struct {
            URL string `json:\"html_url\"`
        } `json:\"repository\"`
    }
    if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
        slog.ErrorContext(ctx, \"failed to decode GitHub webhook payload\", \"error\", err)
        webhookEventsTotal.WithLabelValues(\"github\", eventType, \"error\").Inc()
        http.Error(w, \"invalid payload\", http.StatusBadRequest)
        return
    }
    defer r.Body.Close()

    // Generate deduplication key: sha256(repoURL + ref + after)
    dedupeKey := fmt.Sprintf(\"%x\", sha256.Sum256([]byte(fmt.Sprintf(\"%s:%s:%s\", repoURL, payload.Ref, payload.After))))

    // Check if event was already processed
    var exists bool
    err := h.eventCache.Get(ctx, dedupeKey, &exists)
    if err == nil && exists {
        slog.InfoContext(ctx, \"deduplicated webhook event\", \"dedupe_key\", dedupeKey)
        webhookDedupeHits.Inc()
        webhookEventsTotal.WithLabelValues(\"github\", eventType, \"deduplicated\").Inc()
        w.WriteHeader(http.StatusOK)
        return
    }

    // Cache event for 5 minutes to prevent duplicates
    if err := h.eventCache.Set(ctx, dedupeKey, true, 5*time.Minute); err != nil {
        slog.WarnContext(ctx, \"failed to cache webhook dedupe key\", \"error\", err)
    }

    // Publish event to Kafka
    event := v1alpha1.GitWebhookEvent{
        RepoURL:     payload.Repository.URL,
        Ref:         payload.Ref,
        Revision:    payload.After,
        Provider:    v1alpha1.GitHubProvider,
        ReceivedAt:  time.Now().UTC(),
        EventID:     h.rand.Hex(8),
    }
    eventBytes, err := json.Marshal(event)
    if err != nil {
        slog.ErrorContext(ctx, \"failed to marshal webhook event\", \"error\", err)
        webhookEventsTotal.WithLabelValues(\"github\", eventType, \"error\").Inc()
        http.Error(w, \"internal error\", http.StatusInternalServerError)
        return
    }

    if err := h.kafkaWriter.WriteMessages(ctx, kafka.Message{
        Key:   []byte(repoURL),
        Value: eventBytes,
    }); err != nil {
        slog.ErrorContext(ctx, \"failed to publish webhook event to Kafka\", \"error\", err)
        webhookEventsTotal.WithLabelValues(\"github\", eventType, \"error\").Inc()
        http.Error(w, \"failed to process webhook\", http.StatusInternalServerError)
        return
    }

    slog.InfoContext(ctx, \"published webhook event to Kafka\", \"repo_url\", repoURL, \"revision\", payload.After)
    webhookEventsTotal.WithLabelValues(\"github\", eventType, \"success\").Inc()
    w.WriteHeader(http.StatusAccepted)
}

// Copyright 2024 The ArgoCD Authors. Licensed under Apache 2.0.
// Source: https://github.com/argoproj/argo-cd/blob/v2.12.0/reposerver/manifest/generator.go
package manifest

import (
    \"context\"
    \"fmt\"
    \"io\"
    \"os\"
    \"path/filepath\"
    \"time\"

    \"github.com/google/gvisor/pkg/sentry/control\"
    \"github.com/google/gvisor/pkg/sentry/sandbox\"
    \"github.com/prometheus/client_golang/prometheus\"
    \"golang.org/x/exp/slog\"

    \"github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1\"
    \"github.com/argoproj/argo-cd/v2/reposerver/apiclient\"
    \"github.com/argoproj/argo-cd/v2/util/git\"
)

var (
    manifestGenDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    \"argocd_manifest_generation_duration_seconds\",
            Help:    \"Duration of manifest generation per plugin\",
            Buckets: prometheus.DefBuckets,
        },
        []string{\"plugin_type\"},
    )
    manifestGenErrors = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: \"argocd_manifest_generation_errors_total\",
            Help: \"Total manifest generation errors\",
        },
        []string{\"plugin_type\", \"error_type\"},
    )
)

func init() {
    prometheus.MustRegister(manifestGenDuration, manifestGenErrors)
}

// SandboxedGenerator runs manifest generation plugins in isolated gVisor sandboxes to prevent supply chain attacks
type SandboxedGenerator struct {
    sandboxPool *sandbox.Pool
    gitClient   git.Client
}

// NewSandboxedGenerator initializes a generator with a pre-warmed pool of gVisor sandboxes
func NewSandboxedGenerator(gitClient git.Client, sandboxCount int) (*SandboxedGenerator, error) {
    pool, err := sandbox.NewPool(sandbox.Config{
        NumWorkers: sandboxCount,
        RootDir:    filepath.Join(os.TempDir(), \"argocd-sandboxes\"),
        LogLevel:   slog.LevelInfo,
    })
    if err != nil {
        return nil, fmt.Errorf(\"failed to create sandbox pool: %w\", err)
    }
    return &SandboxedGenerator{
        sandboxPool: pool,
        gitClient:   gitClient,
    }, nil
}

// GenerateManifests fetches the target git revision, runs the specified plugin in a sandbox, and returns generated manifests
func (g *SandboxedGenerator) GenerateManifests(ctx context.Context, app *v1alpha1.Application, revision string) ([]*v1alpha1.Manifest, error) {
    start := time.Now()
    pluginType := app.Spec.Source.Plugin.Type
    defer func() {
        manifestGenDuration.WithLabelValues(pluginType).Observe(time.Since(start).Seconds())
    }()

    // Fetch git repo to temp dir
    repoPath, err := g.gitClient.Clone(ctx, app.Spec.Source.RepoURL, revision)
    if err != nil {
        manifestGenErrors.WithLabelValues(pluginType, \"git_clone\").Inc()
        return nil, fmt.Errorf(\"failed to clone repo: %w\", err)
    }
    defer os.RemoveAll(repoPath)

    // Prepare sandbox input: plugin config, repo path, app source
    input := &apiclient.ManifestGenerationInput{
        PluginType: pluginType,
        RepoPath:   repoPath,
        Source:     app.Spec.Source,
        Revision:   revision,
    }
    inputBytes, err := input.Marshal()
    if err != nil {
        manifestGenErrors.WithLabelValues(pluginType, \"marshal_input\").Inc()
        return nil, fmt.Errorf(\"failed to marshal input: %w\", err)
    }

    // Acquire sandbox from pool
    sb, err := g.sandboxPool.Acquire(ctx)
    if err != nil {
        manifestGenErrors.WithLabelValues(pluginType, \"sandbox_acquire\").Inc()
        return nil, fmt.Errorf(\"failed to acquire sandbox: %w\", err)
    }
    defer sb.Release()

    // Run plugin in sandbox with 30s timeout
    sandboxCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
    defer cancel()

    outputBytes, err := sb.Run(sandboxCtx, \"manifest-plugin\", inputBytes)
    if err != nil {
        manifestGenErrors.WithLabelValues(pluginType, \"plugin_run\").Inc()
        return nil, fmt.Errorf(\"plugin execution failed: %w\", err)
    }

    // Parse generated manifests
    var output apiclient.ManifestGenerationOutput
    if err := output.Unmarshal(outputBytes); err != nil {
        manifestGenErrors.WithLabelValues(pluginType, \"unmarshal_output\").Inc()
        return nil, fmt.Errorf(\"failed to unmarshal output: %w\", err)
    }

    if output.Error != \"\" {
        manifestGenErrors.WithLabelValues(pluginType, \"plugin_error\").Inc()
        return nil, fmt.Errorf(\"plugin returned error: %s\", output.Error)
    }

    slog.InfoContext(ctx, \"generated manifests in sandbox\", \"app\", app.Name, \"manifest_count\", len(output.Manifests))
    return output.Manifests, nil
}

// Copyright 2024 The ArgoCD Authors. Licensed under Apache 2.0.
// Source: https://github.com/argoproj/argo-cd/blob/v2.12.0/application/controller/sync.go
package controller

import (
    \"context\"
    \"fmt\"
    \"time\"

    \"github.com/prometheus/client_golang/prometheus\"
    \"golang.org/x/exp/slog\"
    \"k8s.io/client-go/dynamic\"
    \"k8s.io/client-go/kubernetes\"

    \"github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1\"
    \"github.com/argoproj/argo-cd/v2/util/kube\"
    \"github.com/argoproj/argo-cd/v2/util/merge\"
)

var (
    syncDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    \"argocd_sync_duration_seconds\",
            Help:    \"Duration of cluster sync operations\",
            Buckets: prometheus.DefBuckets,
        },
        []string{\"region\", \"sync_phase\"},
    )
    syncErrors = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: \"argocd_sync_errors_total\",
            Help: \"Total sync errors by phase\",
        },
        []string{\"region\", \"sync_phase\", \"error_type\"},
    )
)

func init() {
    prometheus.MustRegister(syncDuration, syncErrors)
}

// SyncWorker performs 3-way merge sync between git manifests, live cluster state, and last applied config
type SyncWorker struct {
    dynamicClient dynamic.Interface
    kubeClient    kubernetes.Interface
    region        string
}

// NewSyncWorker initializes a sync worker for a specific region
func NewSyncWorker(region string, config *kube.Config) (*SyncWorker, error) {
    dynamicClient, err := dynamic.NewForConfig(config.Config)
    if err != nil {
        return nil, fmt.Errorf(\"failed to create dynamic client: %w\", err)
    }
    kubeClient, err := kubernetes.NewForConfig(config.Config)
    if err != nil {
        return nil, fmt.Errorf(\"failed to create kube client: %w\", err)
    }
    return &SyncWorker{
        dynamicClient: dynamicClient,
        kubeClient:    kubeClient,
        region:        region,
    }, nil
}

// Sync performs a full sync operation for a single application on a target cluster
func (w *SyncWorker) Sync(ctx context.Context, app *v1alpha1.Application, manifests []*v1alpha1.Manifest) (*v1alpha1.SyncStatus, error) {
    start := time.Now()
    phase := \"full_sync\"
    defer func() {
        syncDuration.WithLabelValues(w.region, phase).Observe(time.Since(start).Seconds())
    }()

    // Fetch live cluster state for all target resources
    liveState, err := w.fetchLiveState(ctx, app.Spec.Source, manifests)
    if err != nil {
        syncErrors.WithLabelValues(w.region, phase, \"fetch_live\").Inc()
        return nil, fmt.Errorf(\"failed to fetch live state: %w\", err)
    }

    // Fetch last applied configuration from ArgoCD ConfigMap
    lastApplied, err := w.fetchLastApplied(ctx, app.Name)
    if err != nil {
        slog.WarnContext(ctx, \"failed to fetch last applied config, using empty\", \"error\", err)
        lastApplied = make(map[string]*v1alpha1.Manifest)
    }

    // Perform 3-way merge
    merged, err := merge.ThreeWayMerge(manifests, liveState, lastApplied)
    if err != nil {
        syncErrors.WithLabelValues(w.region, phase, \"merge\").Inc()
        return nil, fmt.Errorf(\"3-way merge failed: %w\", err)
    }

    // Apply merged manifests with exponential backoff retry
    backoff := 100 * time.Millisecond
    maxRetries := 5
    var applyErr error
    for i := 0; i < maxRetries; i++ {
        applyErr = w.applyManifests(ctx, merged)
        if applyErr == nil {
            break
        }
        slog.WarnContext(ctx, \"failed to apply manifests, retrying\", \"retry\", i+1, \"error\", applyErr)
        time.Sleep(backoff)
        backoff *= 2
    }
    if applyErr != nil {
        syncErrors.WithLabelValues(w.region, phase, \"apply\").Inc()
        return nil, fmt.Errorf(\"failed to apply manifests after %d retries: %w\", maxRetries, applyErr)
    }

    // Update last applied ConfigMap
    if err := w.updateLastApplied(ctx, app.Name, merged); err != nil {
        slog.WarnContext(ctx, \"failed to update last applied config\", \"error\", err)
    }

    status := &v1alpha1.SyncStatus{
        Status:     v1alpha1.SyncStatusCodeSynced,
        Revision:   app.Spec.Source.TargetRevision,
        SyncedAt:   time.Now().UTC(),
        Region:     w.region,
        ResourceCount: len(merged),
    }
    slog.InfoContext(ctx, \"sync completed successfully\", \"app\", app.Name, \"region\", w.region, \"resources\", len(merged))
    return status, nil
}

gh api repos/your-org/your-repo/hooks/1234/config --method PATCH --field batch_events=true --field batch_timeout=5 --field batch_size=100

helm install argocd-sync-worker argo/argo-cd --set worker.region=us-east-1 --set worker.kafka.brokers=kafka.us-east-1.example.com:9092

kubectl patch configmap argocd-cmd-params-cm -n argocd --type merge -p '{\"data\":{\"repo.server.sandbox.pool.size\":\"10\"}}'