Standardize controller for GitOps Toolkit

Signed-off-by: Stefan Prodan <[email protected]>

Author: stefanprodan

.github/dependabot.yaml

@@ -4,7 +4,7 @@ updates:
   - package-ecosystem: "gomod"
     directory: "/"
     schedule:
-      interval: "daily"
+      interval: "monthly"
     groups:
       flux-deps:
         patterns:
@@ -22,15 +22,15 @@ updates:
   - package-ecosystem: "github-actions"
     directory: "/"
     schedule:
-      interval: "daily"
+      interval: "monthly"
     groups:
       ci:
         patterns:
           - "*"
   - package-ecosystem: "docker"
     directory: "/"
     schedule:
-      interval: "daily"
+      interval: "monthly"
     groups:
       docker:
         patterns:

CONTRIBUTING.md

@@ -40,9 +40,9 @@ meeting](https://docs.google.com/document/d/1l_M0om0qUEN_NNiGgpqJ2tvsF2iioHkaARD
 ### How to run the test suite
 
 Prerequisites:
-* go >= 1.20
-* docker >= 20.10
-* kustomize >= 4.4
+* go >= 1.25
+* docker >= 28.3
+* kustomize >= 5.7
 
 You can run the unit tests by simply doing
 

README.md

@@ -1,12 +1,17 @@
 # source-watcher
 
-[![test](https://github.com/fluxcd/source-watcher/workflows/test/badge.svg)](https://github.com/fluxcd/source-watcher/actions)
+[![test](https://github.com/fluxcd/source-watcher/workflows/e2e/badge.svg)](https://github.com/fluxcd/source-watcher/actions)
+[![report](https://goreportcard.com/badge/github.com/fluxcd/source-watcher)](https://goreportcard.com/report/github.com/fluxcd/source-watcher)
+[![license](https://img.shields.io/github/license/fluxcd/source-watcher.svg)](https://github.com/fluxcd/source-watcher/blob/main/LICENSE)
+[![release](https://img.shields.io/github/release/fluxcd/source-watcher/all.svg)](https://github.com/fluxcd/source-watcher/releases)
 
-A Flux extension controller that enables advanced source composition and decomposition patterns for GitOps workflows.
+The source-watcher is a [GitOps toolkit](https://fluxcd.io/flux/components/) controller
+that extends Flux with advanced source composition and decomposition patterns.
 
 ## Overview
 
-source-watcher introduces the **ArtifactGenerator** API, which allows you to:
+The source-watcher controller implements the [ArtifactGenerator](docs/README.md) API,
+which allows Flux users to:
 
 - 🔗 **Compose** multiple Flux sources (GitRepository, OCIRepository, Bucket) into a single deployable artifact
 - 📦 **Decompose** monorepos into multiple independent artifacts with separate deployment lifecycles
@@ -99,7 +104,7 @@ spec:
 ```
 
 Each service gets its own ExternalArtifact with an independent revision.
-Changes to `services/auth/` only trigger reconciliation of the auth-service,
+Changes to `services/auth/` only trigger the reconciliation of the auth-service,
 leaving other services untouched.
 
 ### Example: Helm Chart Composition
@@ -122,7 +127,7 @@ spec:
       name: podinfo-values
   artifacts:
     - name: podinfo-composite
-      originRevision: "@chart" # Track chart version
+      originRevision: "@chart"
       copy:
         - from: "@chart/"
           to: "@artifact/"
@@ -180,10 +185,6 @@ Copy operations support glob patterns, exclusions and YAML merge:
 - **Selective Deployment** - Deploy only changed components from large repositories
   by decomposing them into dedicated artifacts.
 
-## API Reference
-
-- [ArtifactGenerator CRD](docs/spec/v1beta1/artifactgenerators.md)
-
 ## Contributing
 
 This project is Apache 2.0 licensed and accepts contributions via GitHub pull requests.

api/v1beta1/artifactgenerator_types.go

@@ -233,6 +233,10 @@ func (in *ArtifactGenerator) HasArtifactInInventory(name, namespace, digest stri
 
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
+// +kubebuilder:resource:shortName=ag
+// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp",description=""
+// +kubebuilder:printcolumn:name="Ready",type="string",JSONPath=".status.conditions[?(@.type==\"Ready\")].status",description=""
+// +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.conditions[?(@.type==\"Ready\")].message",description=""
 
 // ArtifactGenerator is the Schema for the artifactgenerators API.
 type ArtifactGenerator struct {

cmd/main.go

@@ -17,30 +17,35 @@ limitations under the License.
 package main
 
 import (
+	"fmt"
 	"os"
 	"time"
 
-	"github.com/fluxcd/pkg/runtime/acl"
-	"github.com/fluxcd/pkg/runtime/probes"
 	flag "github.com/spf13/pflag"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 	"k8s.io/utils/ptr"
-	ctrl "sigs.k8s.io/controller-runtime"
+	ctrlruntime "sigs.k8s.io/controller-runtime"
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 	ctrlcfg "sigs.k8s.io/controller-runtime/pkg/config"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
-	artcfg "github.com/fluxcd/pkg/artifact/config"
-	artdigest "github.com/fluxcd/pkg/artifact/digest"
-	artsrv "github.com/fluxcd/pkg/artifact/server"
-	artstore "github.com/fluxcd/pkg/artifact/storage"
+	"github.com/fluxcd/pkg/artifact/config"
+	"github.com/fluxcd/pkg/artifact/digest"
+	"github.com/fluxcd/pkg/artifact/server"
+	"github.com/fluxcd/pkg/artifact/storage"
+	"github.com/fluxcd/pkg/runtime/acl"
 	"github.com/fluxcd/pkg/runtime/client"
+	ctrl "github.com/fluxcd/pkg/runtime/controller"
+	"github.com/fluxcd/pkg/runtime/features"
+	"github.com/fluxcd/pkg/runtime/jitter"
+	"github.com/fluxcd/pkg/runtime/leaderelection"
 	"github.com/fluxcd/pkg/runtime/logger"
 	"github.com/fluxcd/pkg/runtime/pprof"
+	"github.com/fluxcd/pkg/runtime/probes"
 	sourcev1 "github.com/fluxcd/source-controller/api/v1"
 
 	swapi "github.com/fluxcd/source-watcher/api/v1beta1"
@@ -50,7 +55,7 @@ import (
 
 var (
 	scheme   = runtime.NewScheme()
-	setupLog = ctrl.Log.WithName("setup")
+	setupLog = ctrlruntime.Log.WithName("setup")
 )
 
 func init() {
@@ -64,66 +69,83 @@ func main() {
 	const controllerName = "source-watcher"
 
 	var (
-		metricsAddr          string
-		healthAddr           string
-		enableLeaderElection bool
-		concurrent           int
-		httpRetry            int
-		requeueDependency    time.Duration
-		artifactOptions      artcfg.Options
-		aclOptions           acl.Options
-		clientOptions        client.Options
-		logOptions           logger.Options
+		metricsAddr           string
+		healthAddr            string
+		concurrent            int
+		httpRetry             int
+		reconciliationTimeout time.Duration
+		requeueDependency     time.Duration
+		artifactOptions       config.Options
+		aclOptions            acl.Options
+		clientOptions         client.Options
+		logOptions            logger.Options
+		leaderElectionOptions leaderelection.Options
+		rateLimiterOptions    ctrl.RateLimiterOptions
+		intervalJitterOptions jitter.IntervalOptions
+		featureGates          features.FeatureGates
 	)
 
 	flag.IntVar(&concurrent, "concurrent", 10, "The number of concurrent resource reconciles.")
 	flag.StringVar(&metricsAddr, "metrics-addr", ":8080", "The address the metric endpoint binds to.")
 	flag.StringVar(&healthAddr, "health-addr", ":9440", "The address the health endpoint binds to.")
-	flag.BoolVar(&enableLeaderElection, "enable-leader-election", false,
-		"Enable leader election for controller manager. "+
-			"Enabling this will ensure there is only one active controller manager.")
 	flag.IntVar(&httpRetry, "http-retry", 9,
 		"The maximum number of retries when failing to fetch artifacts over HTTP.")
+	flag.DurationVar(&reconciliationTimeout, "reconciliation-timeout", 10*time.Minute,
+		"The maximum duration of a reconciliation.")
 	flag.DurationVar(&requeueDependency, "requeue-dependency", 5*time.Second,
 		"The interval at which failing dependencies are reevaluated.")
 
 	artifactOptions.BindFlags(flag.CommandLine)
 	aclOptions.BindFlags(flag.CommandLine)
 	clientOptions.BindFlags(flag.CommandLine)
 	logOptions.BindFlags(flag.CommandLine)
+	leaderElectionOptions.BindFlags(flag.CommandLine)
+	rateLimiterOptions.BindFlags(flag.CommandLine)
+	intervalJitterOptions.BindFlags(flag.CommandLine)
+	featureGates.BindFlags(flag.CommandLine)
 
 	flag.Parse()
 
-	ctrl.SetLogger(logger.NewLogger(logOptions))
+	ctrlruntime.SetLogger(logger.NewLogger(logOptions))
 
-	algo, err := artdigest.AlgorithmForName(artifactOptions.ArtifactDigestAlgo)
+	digestAlgo, err := digest.AlgorithmForName(artifactOptions.ArtifactDigestAlgo)
 	if err != nil {
 		setupLog.Error(err, "unable to configure canonical digest algorithm")
 		os.Exit(1)
 	}
-	artdigest.Canonical = algo
+	digest.Canonical = digestAlgo
 
-	storage, err := artstore.New(&artifactOptions)
+	artifactStorage, err := storage.New(&artifactOptions)
 	if err != nil {
 		setupLog.Error(err, "unable to configure artifact storage")
 		os.Exit(1)
 	}
-	setupLog.Info("storage setup for " + storage.BasePath)
+	setupLog.Info("storage setup for " + artifactStorage.BasePath)
 
-	ctx := ctrl.SetupSignalHandler()
-	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
+	if err := intervalJitterOptions.SetGlobalJitter(nil); err != nil {
+		setupLog.Error(err, "unable to set global jitter")
+		os.Exit(1)
+	}
+
+	ctx := ctrlruntime.SetupSignalHandler()
+	leaderElectionID := fmt.Sprintf("%s-%s", controllerName, "leader-election")
+	mgr, err := ctrlruntime.NewManager(ctrlruntime.GetConfigOrDie(), ctrlruntime.Options{
 		Scheme: scheme,
 		Metrics: metricsserver.Options{
 			BindAddress:   metricsAddr,
 			ExtraHandlers: pprof.GetHandlers(),
 		},
 		HealthProbeBindAddress:        healthAddr,
-		LeaderElection:                enableLeaderElection,
-		LeaderElectionID:              controllerName,
-		LeaderElectionReleaseOnCancel: true,
+		LeaderElection:                leaderElectionOptions.Enable,
+		LeaderElectionID:              leaderElectionID,
+		LeaderElectionReleaseOnCancel: leaderElectionOptions.ReleaseOnCancel,
+		LeaseDuration:                 &leaderElectionOptions.LeaseDuration,
+		RenewDeadline:                 &leaderElectionOptions.RenewDeadline,
+		Logger:                        ctrlruntime.Log,
 		Controller: ctrlcfg.Controller{
 			MaxConcurrentReconciles: concurrent,
 			RecoverPanic:            ptr.To(true),
+			ReconciliationTimeout:   reconciliationTimeout,
 		},
 		Client: ctrlclient.Options{
 			Cache: &ctrlclient.CacheOptions{
@@ -136,38 +158,44 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Note that the liveness check will pass beyond this point, but the readiness
+	// check will continue to fail until this controller instance is elected leader.
+	probes.SetupChecks(mgr, setupLog)
+
 	if err = (&controller.ArtifactGeneratorReconciler{
 		ControllerName:            controllerName,
 		Client:                    mgr.GetClient(),
 		APIReader:                 mgr.GetAPIReader(),
 		Scheme:                    mgr.GetScheme(),
 		EventRecorder:             mgr.GetEventRecorderFor(controllerName),
-		Storage:                   storage,
+		Storage:                   artifactStorage,
 		ArtifactFetchRetries:      httpRetry,
 		DependencyRequeueInterval: requeueDependency,
 		NoCrossNamespaceRefs:      aclOptions.NoCrossNamespaceRefs,
-	}).SetupWithManager(ctx, mgr); err != nil {
+	}).SetupWithManager(ctx, mgr, controller.ArtifactGeneratorReconcilerOptions{
+		RateLimiter: ctrl.GetRateLimiter(rateLimiterOptions),
+	}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", swapi.ArtifactGeneratorKind)
 		os.Exit(1)
 	}
 	// +kubebuilder:scaffold:builder
 
 	go func() {
-		// Block until our controller manager is elected leader. We presume our
+		// Block until our controller manager is elected leader.We presume our
 		// entire process will terminate if we lose leadership, so we don't need
 		// to handle that.
 		<-mgr.Elected()
 
 		// Start the artifact server if running as leader.
+		// This will make the readiness check pass and Kubernetes will start
+		// routing traffic from kustomize-controller and helm-controller to this instance.
 		setupLog.Info("starting storage server on " + artifactOptions.StorageAddress)
-		if err := artsrv.Start(ctx, &artifactOptions); err != nil {
+		if err := server.Start(ctx, &artifactOptions); err != nil {
 			setupLog.Error(err, "artifact server error")
 			os.Exit(1)
 		}
 	}()
 
-	probes.SetupChecks(mgr, setupLog)
-
 	setupLog.Info("starting manager")
 	if err := mgr.Start(ctx); err != nil {
 		setupLog.Error(err, "problem running manager")

config/crd/bases/source.extensions.fluxcd.io_artifactgenerators.yaml

@@ -11,10 +11,22 @@ spec:
     kind: ArtifactGenerator
     listKind: ArtifactGeneratorList
     plural: artifactgenerators
+    shortNames:
+    - ag
     singular: artifactgenerator
   scope: Namespaced
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - jsonPath: .metadata.creationTimestamp
+      name: Age
+      type: date
+    - jsonPath: .status.conditions[?(@.type=="Ready")].status
+      name: Ready
+      type: string
+    - jsonPath: .status.conditions[?(@.type=="Ready")].message
+      name: Status
+      type: string
+    name: v1beta1
     schema:
       openAPIV3Schema:
         description: ArtifactGenerator is the Schema for the artifactgenerators API.

config/rbac/role.yaml

@@ -13,7 +13,7 @@ rules:
   - delete
   - get
   - list
-  - patch
+  - patchStatus
   - update
   - watch
 - apiGroups:
@@ -28,5 +28,17 @@ rules:
   - artifactgenerators/status
   verbs:
   - get
-  - patch
+  - patchStatus
   - update
+- apiGroups:
+  - source.toolkit.fluxcd.io
+  resources:
+  - '*'
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patchStatus
+  - update
+  - watch