ArgoCD 3.0 vs Flux 2.5: GitOps Deployment Speed Comparison

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    argocd "github.com/argoproj/argo-cd/v3/pkg/apiclient"
    "github.com/argoproj/argo-cd/v3/pkg/apiclient/application"
    "github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const (
    argocdAddr  = "argocd-server.argocd.svc.cluster.local:443"
    syncTimeout = 5 * time.Minute
    retryCount  = 3
)

func main() {
    // Validate environment variables
    token := os.Getenv("ARGOCD_TOKEN")
    if token == "" {
        log.Fatal("ARGOCD_TOKEN environment variable must be set")
    }

    // Initialize ArgoCD API client
    clientOpts := argocd.ClientOptions{
        ServerAddr: argocdAddr,
        AuthToken:  token,
        Insecure:   true, // For demo only; use TLS certs in production
        Context:    context.Background(),
    }

    conn, err := argocd.NewClient(&clientOpts)
    if err != nil {
        log.Fatalf("Failed to create ArgoCD client: %v", err)
    }
    defer conn.Close()

    appClient, err := conn.NewApplicationClient()
    if err != nil {
        log.Fatalf("Failed to create application client: %v", err)
    }

    // List all applications in the cluster
    listReq := &application.ListAppsRequest{}
    apps, err := appClient.List(context.Background(), listReq)
    if err != nil {
        log.Fatalf("Failed to list applications: %v", err)
    }

    log.Printf("Found %d applications to sync", len(apps.Items))

    // Sync each application with retries
    for _, app := range apps.Items {
        appName := app.Name
        appNamespace := app.Namespace

        var syncErr error
        for i := 0; i < retryCount; i++ {
            log.Printf("Syncing app %s (attempt %d/%d)", appName, i+1, retryCount)
            syncReq := &application.SyncRequest{
                Name:         appName,
                Namespace:    appNamespace,
                Revision:     app.Spec.Source.TargetRevision,
                Prune:        true,
                Timeout:      &metav1.Duration{Duration: syncTimeout},
                SyncOptions:  []string{"ApplyOutOfSyncOnly=true"},
            }

            _, syncErr = appClient.Sync(context.Background(), syncReq)
            if syncErr == nil {
                log.Printf("Successfully synced app %s", appName)
                break
            }
            log.Printf("Sync attempt %d failed for %s: %v", i+1, appName, syncErr)
            time.Sleep(2 * time.Second)
        }

        if syncErr != nil {
            log.Printf("Failed to sync app %s after %d attempts: %v", appName, retryCount, syncErr)
        }
    }
}

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    flux "github.com/fluxcd/flux2/pkg/client"
    "github.com/fluxcd/flux2/pkg/client/resources"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
)

const (
    fluxNamespace   = "flux-system"
    reconcileTimeout = 5 * time.Minute
    retryCount     = 3
)

func main() {
    // Load kubeconfig
    kubeconfig := os.Getenv("KUBECONFIG")
    config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
        log.Fatalf("Failed to load kubeconfig: %v", err)
    }

    // Initialize Kubernetes client
    k8sClient, err := kubernetes.NewForConfig(config)
    if err != nil {
        log.Fatalf("Failed to create Kubernetes client: %v", err)
    }

    // Initialize Flux client
    fluxClient, err := flux.NewClient(config)
    if err != nil {
        log.Fatalf("Failed to create Flux client: %v", err)
    }

    // List all Flux Kustomizations
    kustomizationClient := fluxClient.Resources().Kustomizations()
    kustomizations, err := kustomizationClient.List(context.Background(), fluxNamespace, metav1.ListOptions{})
    if err != nil {
        log.Fatalf("Failed to list Kustomizations: %v", err)
    }

    log.Printf("Found %d Kustomizations to reconcile", len(kustomizations.Items))

    // Reconcile each Kustomization with retries
    for _, kustomization := range kustomizations.Items {
        kName := kustomization.Name
        kNamespace := kustomization.Namespace

        var reconcileErr error
        for i := 0; i < retryCount; i++ {
            log.Printf("Reconciling Kustomization %s/%s (attempt %d/%d)", kNamespace, kName, i+1, retryCount)
            // Annotate to trigger reconciliation
            annotations := kustomization.Annotations
            if annotations == nil {
                annotations = make(map[string]string)
            }
            annotations["reconcile.fluxcd.io/requestedAt"] = time.Now().Format(time.RFC3339)
            kustomization.Annotations = annotations

            _, reconcileErr = kustomizationClient.Update(context.Background(), &kustomization, metav1.UpdateOptions{})
            if reconcileErr == nil {
                // Wait for reconciliation to complete
                waitCtx, cancel := context.WithTimeout(context.Background(), reconcileTimeout)
                defer cancel()
                for {
                    select {
                    case <-waitCtx.Done():
                        reconcileErr = fmt.Errorf("reconciliation timed out for %s/%s", kNamespace, kName)
                        break
                    default:
                        current, err := kustomizationClient.Get(context.Background(), kName, metav1.GetOptions{})
                        if err != nil {
                            reconcileErr = err
                            break
                        }
                        if current.Status.LastAppliedRevision != "" {
                            log.Printf("Successfully reconciled Kustomization %s/%s", kNamespace, kName)
                            reconcileErr = nil
                            break
                        }
                        time.Sleep(1 * time.Second)
                    }
                    if reconcileErr == nil {
                        break
                    }
                }
                if reconcileErr == nil {
                    break
                }
            }
            log.Printf("Reconciliation attempt %d failed for %s/%s: %v", i+1, kNamespace, kName, reconcileErr)
            time.Sleep(2 * time.Second)
        }

        if reconcileErr != nil {
            log.Printf("Failed to reconcile Kustomization %s/%s after %d attempts: %v", kNamespace, kName, retryCount, reconcileErr)
        }
    }
}

import os
import time
import subprocess
import json
from dataclasses import dataclass
from typing import List, Dict

@dataclass
class BenchmarkResult:
    tool: str
    manifest_size_kb: int
    sync_time_ms: float
    apply_throughput: float
    error_count: int

class GitOpsBenchmarker:
    def __init__(self, kubeconfig: str, argocd_token: str, iterations: int = 10000):
        self.kubeconfig = kubeconfig
        self.argocd_token = argocd_token
        self.iterations = iterations
        self.results: List[BenchmarkResult] = []
        # Validate tools are installed
        self._validate_tools()

    def _validate_tools(self):
        """Check that required CLIs are installed and cluster is reachable."""
        required_tools = ["kubectl", "argocd", "flux"]
        for tool in required_tools:
            try:
                subprocess.run([tool, "version"], capture_output=True, check=True)
            except FileNotFoundError:
                raise RuntimeError(f"Required tool {tool} is not installed or not in PATH")
        # Validate kubeconfig works
        try:
            subprocess.run(
                ["kubectl", "cluster-info", "--kubeconfig", self.kubeconfig],
                capture_output=True, check=True
            )
        except subprocess.CalledProcessError:
            raise RuntimeError("Invalid kubeconfig or cluster is unreachable")

    def _generate_manifest(self, size_kb: int) -> str:
        """Generate a dummy Nginx deployment manifest of specified size."""
        base_manifest = """
apiVersion: apps/v1
kind: Deployment
metadata:
  name: nginx-benchmark
  namespace: default
spec:
  replicas: 3
  selector:
    matchLabels:
      app: nginx
  template:
    metadata:
      labels:
        app: nginx
    spec:
      containers:
      - name: nginx
        image: nginx:1.25
        ports:
        - containerPort: 80
        resources:
          limits:
            cpu: 100m
            memory: 128Mi
"""
        # Pad manifest to reach desired size
        current_size = len(base_manifest.encode("utf-8")) // 1024
        if current_size < size_kb:
            padding = "a" * ((size_kb - current_size) * 1024)
            return base_manifest + "\n# Padding\n" + padding
        return base_manifest

    def run_argocd_benchmark(self, manifest_size_kb: int) -> BenchmarkResult:
        """Run ArgoCD sync benchmark for given manifest size."""
        manifest = self._generate_manifest(manifest_size_kb)
        manifest_path = f"/tmp/argocd-bench-{manifest_size_kb}kb.yaml"
        with open(manifest_path, "w") as f:
            f.write(manifest)

        # Create ArgoCD app (idempotent)
        create_cmd = [
            "argocd", "app", "create", "nginx-bench",
            "--repo", "https://github.com/example/bench-repo",
            "--path", ".",
            "--dest-server", "https://kubernetes.default.svc",
            "--dest-namespace", "default",
            "--kubeconfig", self.kubeconfig,
            "--auth-token", self.argocd_token
        ]
        subprocess.run(create_cmd, capture_output=True, check=False) # Ignore errors if app exists

        sync_times = []
        errors = 0
        for i in range(self.iterations):
            start = time.time()
            try:
                # Sync the app
                sync_cmd = [
                    "argocd", "app", "sync", "nginx-bench",
                    "--kubeconfig", self.kubeconfig,
                    "--auth-token", self.argocd_token,
                    "--timeout", "300"
                ]
                subprocess.run(sync_cmd, capture_output=True, check=True)
                end = time.time()
                sync_times.append((end - start) * 1000) # Convert to ms
            except subprocess.CalledProcessError as e:
                errors += 1
                if i % 100 == 0:
                    print(f"Iteration {i} failed: {e.stderr.decode()}")

        avg_sync_time = sum(sync_times) / len(sync_times) if sync_times else 0
        total_kb = manifest_size_kb * len(sync_times)
        total_seconds = sum(sync_times) / 1000
        throughput = total_kb / total_seconds if total_seconds > 0 else 0

        return BenchmarkResult(
            tool="ArgoCD 3.0",
            manifest_size_kb=manifest_size_kb,
            sync_time_ms=avg_sync_time,
            apply_throughput=throughput,
            error_count=errors
        )

    def run_flux_benchmark(self, manifest_size_kb: int) -> BenchmarkResult:
        """Run Flux reconciliation benchmark for given manifest size."""
        manifest = self._generate_manifest(manifest_size_kb)
        manifest_path = f"/tmp/flux-bench-{manifest_size_kb}kb.yaml"
        with open(manifest_path, "w") as f:
            f.write(manifest)

        # Apply Kustomization (idempotent)
        apply_cmd = ["kubectl", "apply", "-f", manifest_path, "--kubeconfig", self.kubeconfig]
        subprocess.run(apply_cmd, capture_output=True, check=False)

        sync_times = []
        errors = 0
        for i in range(self.iterations):
            start = time.time()
            try:
                # Annotate to trigger reconciliation
                annotate_cmd = [
                    "kubectl", "annotate", "kustomization", "nginx-bench",
                    f"reconcile.fluxcd.io/requestedAt={time.now().isoformat()}",
                    "--kubeconfig", self.kubeconfig, "--overwrite"
                ]
                subprocess.run(annotate_cmd, capture_output=True, check=True)
                # Wait for reconciliation
                wait_cmd = [
                    "flux", "reconcile", "kustomization", "nginx-bench",
                    "--kubeconfig", self.kubeconfig, "--timeout", "5m"
                ]
                subprocess.run(wait_cmd, capture_output=True, check=True)
                end = time.time()
                sync_times.append((end - start) * 1000)
            except subprocess.CalledProcessError as e:
                errors += 1
                if i % 100 == 0:
                    print(f"Iteration {i} failed: {e.stderr.decode()}")

        avg_sync_time = sum(sync_times) / len(sync_times) if sync_times else 0
        total_kb = manifest_size_kb * len(sync_times)
        total_seconds = sum(sync_times) / 1000
        throughput = total_kb / total_seconds if total_seconds > 0 else 0

        return BenchmarkResult(
            tool="Flux 2.5",
            manifest_size_kb=manifest_size_kb,
            sync_time_ms=avg_sync_time,
            apply_throughput=throughput,
            error_count=errors
        )

    def save_results(self, output_path: str):
        """Save benchmark results to JSON."""
        results_dict = [r.__dict__ for r in self.results]
        with open(output_path, "w") as f:
            json.dump(results_dict, f, indent=2)

if __name__ == "__main__":
    # Load environment variables
    kubeconfig = os.getenv("KUBECONFIG", "~/.kube/config")
    argocd_token = os.getenv("ARGOCD_TOKEN")
    if not argocd_token:
        raise RuntimeError("ARGOCD_TOKEN must be set")

    benchmarker = GitOpsBenchmarker(
        kubeconfig=kubeconfig,
        argocd_token=argocd_token,
        iterations=10000
    )

    # Run benchmarks for different manifest sizes
    for size_kb in [1, 10, 100, 1024]:
        print(f"Running ArgoCD benchmark for {size_kb}KB manifest")
        argocd_result = benchmarker.run_argocd_benchmark(size_kb)
        benchmarker.results.append(argocd_result)
        print(f"Running Flux benchmark for {size_kb}KB manifest")
        flux_result = benchmarker.run_flux_benchmark(size_kb)
        benchmarker.results.append(flux_result)

    benchmarker.save_results("/tmp/gitops-benchmark-results.json")
    print("Benchmark complete. Results saved to /tmp/gitops-benchmark-results.json")

apiVersion: apps/v1
kind: Deployment
metadata:
  name: argocd-server
  namespace: argocd
spec:
  template:
    spec:
      containers:
      - name: argocd-server
        args:
        - --sync-parallelism=8
        - --manifest-cache-size=1024
        - --insecure

apiVersion: source.toolkit.fluxcd.io/v1beta2
kind: GitRepository
metadata:
  name: small-config-repo
  namespace: flux-system
spec:
  interval: 1m
  url: https://github.com/example/small-configs
  cache:
    maxSize: 1GB
  ref:
    branch: main

def route_workload(manifest_size_kb: int, namespace: str) -> str:
    if manifest_size_kb > 100:
        return "argocd"
    elif manifest_size_kb < 10:
        return "flux"
    else:
        # Default to ArgoCD for medium manifests
        return "argocd"