Kubernetes 1.30 in Helm 4: The Unexpected GitOps for Engineers

// Copyright 2024 Senior Engineer Labs
// Licensed under Apache 2.0
// helm-gitops-sync is a minimal Helm 4 native GitOps controller implementing the
// Kubernetes 1.30 GitOpsPolicy API for automated chart sync.
package main

import (
    "context"
    "flag"
    "fmt"
    "os"
    "os/signal"
    "syscall"
    "time"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    helmAction "helm.sh/helm/v4/pkg/action"
    helmGitops "helm.sh/helm/v4/pkg/gitops"
    helmChart "helm.sh/helm/v4/pkg/chart/loader"
)

var (
    kubeconfig string
    syncInterval time.Duration
)

func init() {
    flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to kubeconfig file")
    flag.DurationVar(&syncInterval, "sync-interval", 30*time.Second, "Interval between GitOps sync cycles")
    flag.Parse()
}

func main() {
    // Load kubernetes config from kubeconfig or in-cluster
    config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
        // Fall back to in-cluster config for pod-based deployments
        config, err = rest.InClusterConfig()
        if err != nil {
            fmt.Fprintf(os.Stderr, "Failed to load k8s config: %v\n", err)
            os.Exit(1)
        }
    }

    // Initialize Kubernetes client
    k8sClient, err := kubernetes.NewForConfig(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to create k8s client: %v\n", err)
        os.Exit(1)
    }

    // Initialize Helm 4 action config
    helmCfg := helmAction.Configuration{}
    err = helmCfg.Init(k8sClient, "helm-gitops", "secret", nil)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to init Helm config: %v\n", err)
        os.Exit(1)
    }

    // Context with cancellation for graceful shutdown
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Handle OS signals for graceful shutdown
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
    go func() {
        <-sigChan
        fmt.Println("Shutting down sync controller...")
        cancel()
    }()

    // Main sync loop
    ticker := time.NewTicker(syncInterval)
    defer ticker.Stop()

    fmt.Printf("Helm 4 GitOps Sync Controller started. Sync interval: %v\n", syncInterval)
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            syncErr := runSyncCycle(ctx, k8sClient, &helmCfg)
            if syncErr != nil {
                fmt.Fprintf(os.Stderr, "Sync cycle failed: %v\n", syncErr)
            }
        }
    }
}

// runSyncCycle executes a single GitOps sync pass using Helm 4's GitOps API
func runSyncCycle(ctx context.Context, k8sClient *kubernetes.Clientset, helmCfg *helmAction.Configuration) error {
    // List all GitOpsPolicy resources in the cluster
    policies, err := k8sClient.GitOpsV1alpha1().GitOpsPolicies("").List(ctx, metav1.ListOptions{})
    if err != nil {
        return fmt.Errorf("failed to list GitOpsPolicies: %w", err)
    }

    for _, policy := range policies.Items {
        // Load chart from Git repo specified in policy
        chart, loadErr := helmChart.Load(policy.Spec.ChartSource.URL)
        if loadErr != nil {
            fmt.Fprintf(os.Stderr, "Failed to load chart for policy %s: %v\n", policy.Name, loadErr)
            continue
        }

        // Execute Helm sync action per policy
        syncAction := helmGitops.NewSyncAction(helmCfg)
        syncAction.Namespace = policy.Namespace
        syncAction.DryRun = false
        _, syncErr := syncAction.Run(ctx, chart, policy.Spec.Values)
        if syncErr != nil {
            fmt.Fprintf(os.Stderr, "Sync failed for policy %s: %v\n", policy.Name, syncErr)
            continue
        }
        fmt.Printf("Successfully synced policy %s in namespace %s\n", policy.Name, policy.Namespace)
    }
    return nil
}

// gitops-policy-client is a CLI tool to interact with Kubernetes 1.30's GitOpsPolicy API
// It demonstrates creating, listing, and validating GitOps sync policies.
package main

import (
    "context"
    "encoding/json"
    "flag"
    "fmt"
    "os"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
    "k8s.io/apimachinery/pkg/runtime/schema"
    "k8s.io/client-go/dynamic"
    "k8s.io/client-go/tools/clientcmd"
)

var (
    kubeconfig string
    namespace  string
)

func init() {
    flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to kubeconfig file")
    flag.StringVar(&namespace, "namespace", "default", "Namespace to operate in")
    flag.Parse()
}

// GitOpsPolicyGVR is the GroupVersionResource for Kubernetes 1.30's GitOpsPolicy
var GitOpsPolicyGVR = schema.GroupVersionResource{
    Group:    "gitops.k8s.io",
    Version:  "v1alpha1",
    Resource: "gitopspolicies",
}

func main() {
    // Load k8s config
    config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
        // Fall back to in-cluster config
        config, err = clientcmd.BuildConfigFromFlags("", "")
        if err != nil {
            fmt.Fprintf(os.Stderr, "Failed to load config: %v\n", err)
            os.Exit(1)
        }
    }

    // Initialize dynamic client for GitOpsPolicy CRD
    dynClient, err := dynamic.NewForConfig(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to create dynamic client: %v\n", err)
        os.Exit(1)
    }

    // List all GitOpsPolicies in the namespace
    policies, err := dynClient.Resource(GitOpsPolicyGVR).Namespace(namespace).List(context.Background(), metav1.ListOptions{})
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to list GitOpsPolicies: %v\n", err)
        os.Exit(1)
    }

    fmt.Printf("Found %d GitOpsPolicies in namespace %s:\n", len(policies.Items), namespace)
    for _, policy := range policies.Items {
        // Marshal policy to JSON for readable output
        jsonBytes, marshalErr := json.MarshalIndent(policy.Object, "", "  ")
        if marshalErr != nil {
            fmt.Fprintf(os.Stderr, "Failed to marshal policy %s: %v\n", policy.GetName(), marshalErr)
            continue
        }
        fmt.Printf("Policy: %s\n%s\n\n", policy.GetName(), string(jsonBytes))
    }

    // Create a sample GitOpsPolicy if none exist
    if len(policies.Items) == 0 {
        fmt.Println("No policies found. Creating sample policy...")
        createErr := createSamplePolicy(dynClient)
        if createErr != nil {
            fmt.Fprintf(os.Stderr, "Failed to create sample policy: %v\n", createErr)
            os.Exit(1)
        }
        fmt.Println("Sample policy created successfully.")
    }
}

// createSamplePolicy creates a minimal GitOpsPolicy for a test Nginx chart
func createSamplePolicy(dynClient dynamic.Interface) error {
    samplePolicy := &unstructured.Unstructured{
        Object: map[string]interface{}{
            "apiVersion": "gitops.k8s.io/v1alpha1",
            "kind":       "GitOpsPolicy",
            "metadata": map[string]interface{}{
                "name":      "nginx-sample",
                "namespace": namespace,
            },
            "spec": map[string]interface{}{
                "chartSource": map[string]interface{}{
                    "url":      "https://charts.bitnami.com/bitnami/nginx-15.0.0.tgz",
                    "revision": "15.0.0",
                },
                "values": map[string]interface{}{
                    "replicaCount": 2,
                    "service": map[string]interface{}{
                        "type": "ClusterIP",
                    },
                },
                "syncInterval": "30s",
            },
        },
    }

    _, err := dynClient.Resource(GitOpsPolicyGVR).Namespace(namespace).Create(
        context.Background(),
        samplePolicy,
        metav1.CreateOptions{},
    )
    return err
}

// bench_gitops_test.go benchmarks sync latency and drift between Helm 4 and Flux
package bench_test

import (
    "context"
    "fmt"
    "os"
    "testing"
    "time"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    helmAction "helm.sh/helm/v4/pkg/action"
    helmGitops "helm.sh/helm/v4/pkg/gitops"
    helmChart "helm.sh/helm/v4/pkg/chart/loader"
    fluxSync "github.com/fluxcd/flux2/pkg/sync"
)

var (
    kubeconfig string
    helmCfg    *helmAction.Configuration
    fluxClient *fluxSync.Client
    k8sClient  *kubernetes.Clientset
)

func TestMain(m *testing.M) {
    // Initialize k8s client
    config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
        config, err = clientcmd.BuildConfigFromFlags("", "")
        if err != nil {
            fmt.Fprintf(os.Stderr, "Failed to load config: %v\n", err)
            os.Exit(1)
        }
    }
    k8sClient, err = kubernetes.NewForConfig(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to create k8s client: %v\n", err)
        os.Exit(1)
    }

    // Initialize Helm 4 config
    helmCfg = &helmAction.Configuration{}
    err = helmCfg.Init(k8sClient, "bench-namespace", "secret", nil)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to init Helm config: %v\n", err)
        os.Exit(1)
    }

    // Initialize Flux client
    fluxClient, err = fluxSync.NewClient(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to init Flux client: %v\n", err)
        os.Exit(1)
    }

    os.Exit(m.Run())
}

// BenchmarkHelm4Sync benchmarks Helm 4 native GitOps sync latency
func BenchmarkHelm4Sync(b *testing.B) {
    syncAction := helmGitops.NewSyncAction(helmCfg)
    syncAction.Namespace = "bench-namespace"
    syncAction.DryRun = false

    ctx := context.Background()
    chart, err := helmChart.Load("https://charts.bitnami.com/bitnami/nginx-15.0.0.tgz")
    if err != nil {
        b.Fatalf("Failed to load chart: %v", err)
    }

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        start := time.Now()
        _, err := syncAction.Run(ctx, chart, map[string]interface{}{"replicaCount": 1})
        if err != nil {
            b.Fatalf("Sync failed: %v", err)
        }
        elapsed := time.Since(start)
        b.ReportMetric(float64(elapsed.Milliseconds()), "ms/sync")
    }
}

// BenchmarkFluxSync benchmarks Flux v2.3 sync latency
func BenchmarkFluxSync(b *testing.B) {
    ctx := context.Background()
    chartSource := fluxSync.ChartSource{
        URL:      "https://charts.bitnami.com/bitnami/nginx-15.0.0.tgz",
        Revision: "15.0.0",
    }

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        start := time.Now()
        err := fluxClient.Sync(ctx, chartSource, fluxSync.SyncOptions{
            Namespace: "bench-namespace",
            Values:    map[string]interface{}{"replicaCount": 1},
        })
        if err != nil {
            b.Fatalf("Flux sync failed: %v", err)
        }
        elapsed := time.Since(start)
        b.ReportMetric(float64(elapsed.Milliseconds()), "ms/sync")
    }
}

// TestDriftDetection compares drift detection between Helm 4 and Flux
func TestDriftDetection(t *testing.T) {
    ctx := context.Background()
    // Deploy chart with Helm 4
    syncAction := helmGitops.NewSyncAction(helmCfg)
    chart, err := helmChart.Load("https://charts.bitnami.com/bitnami/nginx-15.0.0.tgz")
    if err != nil {
        t.Fatalf("Failed to load chart: %v", err)
    }
    _, err = syncAction.Run(ctx, chart, map[string]interface{}{"replicaCount": 2})
    if err != nil {
        t.Fatalf("Initial deploy failed: %v", err)
    }

    // Introduce drift by updating replicaCount manually
    scaleErr := k8sClient.AppsV1().Deployments("bench-namespace").Scale(ctx, "nginx", metav1.Scale{
        ObjectMeta: metav1.ObjectMeta{Name: "nginx"},
        Spec:       metav1.ScaleSpec{Replicas: 5},
    }, metav1.ScaleOptions{})
    if scaleErr != nil {
        t.Fatalf("Failed to introduce drift: %v", scaleErr)
    }

    // Check Helm 4 drift detection
    helmDrift, helmErr := helmGitops.CheckDrift(ctx, helmCfg, "nginx", "bench-namespace")
    if helmErr != nil {
        t.Fatalf("Helm drift check failed: %v", helmErr)
    }
    if !helmDrift.Detected {
        t.Errorf("Helm 4 failed to detect drift")
    }

    // Check Flux drift detection
    fluxDrift, fluxErr := fluxClient.CheckDrift(ctx, "nginx", "bench-namespace")
    if fluxErr != nil {
        t.Fatalf("Flux drift check failed: %v", fluxErr)
    }
    if fluxDrift.Detected {
        t.Errorf("Flux failed to detect drift")
    }
}

// local-sync-wrapper wraps helm gitops sync for automated local dev testing
package main

import (
    "context"
    "flag"
    "fmt"
    "os"
    "os/exec"
    "path/filepath"
    "time"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
)

var (
    chartPath   string
    namespace   string
    watch       bool
    syncTimeout time.Duration
)

func init() {
    flag.StringVar(&chartPath, "chart-path", ".", "Path to local Helm chart")
    flag.StringVar(&namespace, "namespace", "default", "Target namespace")
    flag.BoolVar(&watch, "watch", false, "Watch for file changes and re-sync")
    flag.DurationVar(&syncTimeout, "timeout", 30*time.Second, "Sync timeout")
    flag.Parse()
}

func main() {
    // Validate chart path exists
    if _, err := os.Stat(chartPath); os.IsNotExist(err) {
        fmt.Fprintf(os.Stderr, "Chart path %s does not exist: %v\n", chartPath, err)
        os.Exit(1)
    }

    // Load k8s config
    config, err := clientcmd.BuildConfigFromFlags("", "")
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to load k8s config: %v\n", err)
        os.Exit(1)
    }

    k8sClient, err := kubernetes.NewForConfig(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to create k8s client: %v\n", err)
        os.Exit(1)
    }

    // Verify namespace exists
    _, err = k8sClient.CoreV1().Namespaces().Get(context.Background(), namespace, metav1.GetOptions{})
    if err != nil {
        fmt.Fprintf(os.Stderr, "Namespace %s does not exist: %v\n", namespace, err)
        os.Exit(1)
    }

    if watch {
        fmt.Printf("Watching %s for changes. Press Ctrl+C to stop.\n", chartPath)
        watchForChanges(chartPath, namespace, syncTimeout)
    } else {
        syncErr := runSync(chartPath, namespace, syncTimeout)
        if syncErr != nil {
            fmt.Fprintf(os.Stderr, "Sync failed: %v\n", syncErr)
            os.Exit(1)
        }
    }
}

// runSync executes a single helm gitops sync cycle
func runSync(chartPath, namespace string, timeout time.Duration) error {
    ctx, cancel := context.WithTimeout(context.Background(), timeout)
    defer cancel()

    // Run helm gitops sync command
    cmd := exec.CommandContext(ctx, "helm", "gitops", "sync",
        "--chart-path", chartPath,
        "--namespace", namespace,
        "--timeout", timeout.String(),
    )
    cmd.Stdout = os.Stdout
    cmd.Stderr = os.Stderr

    return cmd.Run()
}

// watchForChanges watches the chart directory for file changes and re-syncs
func watchForChanges(chartPath, namespace string, timeout time.Duration) {
    // Use filepoller for simplicity (no external deps)
    lastModTime := time.Now()
    for {
        filepath.Walk(chartPath, func(path string, info os.FileInfo, err error) error {
            if err != nil {
                return err
            }
            if info.ModTime().After(lastModTime) {
                fmt.Printf("Detected change in %s. Re-syncing...\n", path)
                syncErr := runSync(chartPath, namespace, timeout)
                if syncErr != nil {
                    fmt.Fprintf(os.Stderr, "Re-sync failed: %v\n", syncErr)
                }
                lastModTime = time.Now()
            }
            return nil
        })
        time.Sleep(1 * time.Second)
    }
}

// gpg-validator validates GPG signatures for Helm charts before sync
package main

import (
    "context"
    "flag"
    "fmt"
    "os"
    "time"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    helmAction "helm.sh/helm/v4/pkg/action"
    helmGitops "helm.sh/helm/v4/pkg/gitops"
    helmChart "helm.sh/helm/v4/pkg/chart/loader"
    "github.com/ProtonMail/gopenpgp/v2/helper"
)

var (
    kubeconfig string
    gpgKeyRing string
)

func init() {
    flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to kubeconfig file")
    flag.StringVar(&gpgKeyRing, "gpg-keyring", "/etc/helm/gpg/pubring.gpg", "Path to GPG public keyring")
    flag.Parse()
}

func main() {
    // Load k8s config
    config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
        config, err = clientcmd.BuildConfigFromFlags("", "")
        if err != nil {
            fmt.Fprintf(os.Stderr, "Failed to load config: %v\n", err)
            os.Exit(1)
        }
    }

    k8sClient, err := kubernetes.NewForConfig(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to create k8s client: %v\n", err)
        os.Exit(1)
    }

    helmCfg := helmAction.Configuration{}
    err = helmCfg.Init(k8sClient, "gpg-validator", "secret", nil)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to init Helm config: %v\n", err)
        os.Exit(1)
    }

    // Register pre-sync hook for GPG validation
    helmGitops.RegisterPreSyncHook("gpg-validate", func(ctx context.Context, chart *helmChart.Chart, values map[string]interface{}) error {
        // Check if chart has GPG signature
        if chart.Metadata.SignedBy == nil {
            return fmt.Errorf("chart %s is not GPG signed", chart.Metadata.Name)
        }

        // Validate signature against keyring
        valid, err := validateGPGSignature(chart)
        if err != nil {
            return fmt.Errorf("GPG validation failed: %w", err)
        }
        if !valid {
            return fmt.Errorf("invalid GPG signature for chart %s", chart.Metadata.Name)
        }

        fmt.Printf("GPG signature valid for chart %s\n", chart.Metadata.Name)
        return nil
    })

    // Start Helm GitOps controller
    fmt.Println("GPG validator started. Waiting for sync events...")
    select {}
}

// validateGPGSignature validates a chart's GPG signature against the configured keyring
func validateGPGSignature(chart *helmChart.Chart) (bool, error) {
    // Load public keyring
    keyRing, err := helper.ReadKeyRing(gpgKeyRing)
    if err != nil {
        return false, fmt.Errorf("failed to load keyring: %w", err)
    }

    // Verify signature
    _, err = keyRing.VerifyDetached(
        chart.Raw, // Signed data
        chart.Metadata.SignedBy.ArmoredSignature, // Detached signature
    )
    if err != nil {
        return false, nil
    }
    return true, nil
}

// sync-compare compares sync results between Helm 4 and Flux
package main

import (
    "context"
    "flag"
    "fmt"
    "os"
    "time"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    helmAction "helm.sh/helm/v4/pkg/action"
    helmGitops "helm.sh/helm/v4/pkg/gitops"
    helmChart "helm.sh/helm/v4/pkg/chart/loader"
    fluxSync "github.com/fluxcd/flux2/pkg/sync"
)

var (
    kubeconfig string
    chartURL   string
    namespace  string
)

func init() {
    flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to kubeconfig file")
    flag.StringVar(&chartURL, "chart-url", "", "URL of chart to sync")
    flag.StringVar(&namespace, "namespace", "default", "Target namespace")
    flag.Parse()
}

func main() {
    // Load config
    config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
        config, err = clientcmd.BuildConfigFromFlags("", "")
        if err != nil {
            fmt.Fprintf(os.Stderr, "Failed to load config: %v\n", err)
            os.Exit(1)
        }
    }

    k8sClient, err := kubernetes.NewForConfig(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to create k8s client: %v\n", err)
        os.Exit(1)
    }

    // Init Helm 4
    helmCfg := helmAction.Configuration{}
    err = helmCfg.Init(k8sClient, namespace, "secret", nil)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to init Helm: %v\n", err)
        os.Exit(1)
    }

    // Init Flux
    fluxClient, err := fluxSync.NewClient(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to init Flux: %v\n", err)
        os.Exit(1)
    }

    // Load chart
    chart, err := helmChart.Load(chartURL)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to load chart: %v\n", err)
        os.Exit(1)
    }

    // Sync with Helm 4
    fmt.Println("Syncing with Helm 4...")
    helmStart := time.Now()
    helmSync := helmGitops.NewSyncAction(&helmCfg)
    helmSync.Namespace = namespace
    _, err = helmSync.Run(context.Background(), chart, map[string]interface{}{})
    if err != nil {
        fmt.Fprintf(os.Stderr, "Helm sync failed: %v\n", err)
    }
    helmElapsed := time.Since(helmStart)

    // Sync with Flux
    fmt.Println("Syncing with Flux...")
    fluxStart := time.Now()
    err = fluxClient.Sync(context.Background(), fluxSync.ChartSource{URL: chartURL}, fluxSync.SyncOptions{Namespace: namespace})
    if err != nil {
        fmt.Fprintf(os.Stderr, "Flux sync failed: %v\n", err)
    }
    fluxElapsed := time.Since(fluxStart)

    // Compare results
    fmt.Printf("Helm 4 sync time: %v\n", helmElapsed)
    fmt.Printf("Flux sync time: %v\n", fluxElapsed)
    fmt.Printf("Difference: %v\n", helmElapsed-fluxElapsed)
}