Adds GitOps Toolkit EventRecorder to ArtifactGenerator reconciler

cmd/main.go

@@ -27,6 +27,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/utils/ptr"
 	ctrlruntime "sigs.k8s.io/controller-runtime"
 	ctrlcache "sigs.k8s.io/controller-runtime/pkg/cache"
@@ -41,6 +42,7 @@ import (
 	gotkacl "github.com/fluxcd/pkg/runtime/acl"
 	gotkclient "github.com/fluxcd/pkg/runtime/client"
 	gotkctrl "github.com/fluxcd/pkg/runtime/controller"
+	gotkevents "github.com/fluxcd/pkg/runtime/events"
 	gotkfeatures "github.com/fluxcd/pkg/runtime/features"
 	gotkjitter "github.com/fluxcd/pkg/runtime/jitter"
 	gotkelection "github.com/fluxcd/pkg/runtime/leaderelection"
@@ -184,13 +186,15 @@ func main() {
 	// check will continue to fail until this controller instance is elected leader.
 	gotkprobes.SetupChecks(mgr, setupLog)
 
+	eventRecorder := mustSetupEventRecorder(mgr, eventsAddr, controllerName)
+
 	// Register the ArtifactGenerator controller with the manager.
 	if err = (&controller.ArtifactGeneratorReconciler{
 		ControllerName:            controllerName,
 		Client:                    mgr.GetClient(),
 		APIReader:                 mgr.GetAPIReader(),
 		Scheme:                    mgr.GetScheme(),
-		EventRecorder:             mgr.GetEventRecorderFor(controllerName),
+		EventRecorder:             eventRecorder,
 		Storage:                   artifactStorage,
 		ArtifactFetchRetries:      httpRetry,
 		DependencyRequeueInterval: requeueDependency,
@@ -225,3 +229,12 @@ func main() {
 		os.Exit(1)
 	}
 }
+
+func mustSetupEventRecorder(mgr ctrlruntime.Manager, eventsAddr, controllerName string) record.EventRecorder {
+	eventRecorder, err := gotkevents.NewRecorder(mgr, ctrlruntime.Log, eventsAddr, controllerName)
+	if err != nil {
+		setupLog.Error(err, "unable to create event recorder")
+		os.Exit(1)
+	}
+	return eventRecorder
+}

go.mod

@@ -10,6 +10,7 @@ replace github.com/opencontainers/go-digest => github.com/opencontainers/go-dige
 
 require (
 	github.com/bmatcuk/doublestar/v4 v4.9.2
+	github.com/fluxcd/pkg/apis/event v0.22.0
 	github.com/fluxcd/pkg/apis/meta v1.25.0
 	github.com/fluxcd/pkg/artifact v0.7.0
 	github.com/fluxcd/pkg/http/fetch v0.22.0

go.sum

@@ -65,6 +65,8 @@ github.com/fluxcd/cli-utils v0.37.1-flux.1 h1:WnG2mHxCPZMj/soIq/S/1zvbrGCJN3GJGb
 github.com/fluxcd/cli-utils v0.37.1-flux.1/go.mod h1:aND5wX3LuTFtB7eUT7vsWr8mmxRVSPR2Wkvbn0SqPfw=
 github.com/fluxcd/pkg/apis/acl v0.9.0 h1:wBpgsKT+jcyZEcM//OmZr9RiF8klL3ebrDp2u2ThsnA=
 github.com/fluxcd/pkg/apis/acl v0.9.0/go.mod h1:TttNS+gocsGLwnvmgVi3/Yscwqrjc17+vhgYfqkfrV4=
+github.com/fluxcd/pkg/apis/event v0.22.0 h1:nCW0TnneMnscSnj9NlaSKcvyC+436MbY1GyKn/4YnII=
+github.com/fluxcd/pkg/apis/event v0.22.0/go.mod h1:Hoi4DejaNKVahGkRXqGBjT9h1aKmhc7RCYcsgoTieqc=
 github.com/fluxcd/pkg/apis/meta v1.25.0 h1:fmZgMoe7yITGfhFqdOs7w2GOu3Y/2Vvz4+4p/eay3eA=
 github.com/fluxcd/pkg/apis/meta v1.25.0/go.mod h1:1D92RqAet0/n/cH5S0khBXweirHWkw9rCO0V4NCY6xc=
 github.com/fluxcd/pkg/artifact v0.7.0 h1:gACHm8Xn8K9RcZ+TuJKCYf3Av3Le3iBW7tpnxVDuHmw=

internal/controller/artifactgenerator_controller.go

@@ -35,6 +35,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
+	eventv1 "github.com/fluxcd/pkg/apis/event/v1beta1"
 	gotkmeta "github.com/fluxcd/pkg/apis/meta"
 	gotkstroage "github.com/fluxcd/pkg/artifact/storage"
 	gotkfetch "github.com/fluxcd/pkg/http/fetch"
@@ -102,7 +103,7 @@ func (r *ArtifactGeneratorReconciler) Reconcile(ctx context.Context, req ctrl.Re
 	// Pause reconciliation if the object has the reconcile annotation set to 'disabled'.
 	if obj.IsDisabled() {
 		log.Error(errors.New("can't reconcile"), msgReconciliationDisabled)
-		r.Event(obj, corev1.EventTypeWarning, swapi.ReconciliationDisabledReason, msgReconciliationDisabled)
+		r.Event(obj, eventv1.EventTypeTrace, swapi.ReconciliationDisabledReason, msgReconciliationDisabled)
 		return ctrl.Result{}, nil
 	}
 
@@ -120,6 +121,7 @@ func (r *ArtifactGeneratorReconciler) reconcile(ctx context.Context,
 	obj *swapi.ArtifactGenerator,
 	patcher *gotkpatch.SerialPatcher) (ctrl.Result, error) {
 	log := ctrl.LoggerFrom(ctx)
+	oldObj := obj.DeepCopy()
 
 	// Create a temporary directory to fetch sources and build artifacts.
 	tmpDir, err := builder.MkdirTempAbs("", "ag-")
@@ -157,7 +159,7 @@ func (r *ArtifactGeneratorReconciler) reconcile(ctx context.Context,
 	if !hasDrifted {
 		msg := fmt.Sprintf("No drift detected, %d artifact(s) up to date", len(obj.Status.Inventory))
 		log.Info(msg)
-		r.Event(obj, corev1.EventTypeNormal, gotkmeta.ReadyCondition, msg)
+		r.Event(obj, eventv1.EventTypeTrace, gotkmeta.ReadyCondition, msg)
 		return ctrl.Result{RequeueAfter: obj.GetRequeueAfter()}, nil
 	}
 
@@ -252,11 +254,30 @@ func (r *ArtifactGeneratorReconciler) reconcile(ctx context.Context,
 		gotkmeta.ReadyCondition,
 		gotkmeta.SucceededReason,
 		"%s", msg)
-	r.Event(obj, corev1.EventTypeNormal, gotkmeta.ReadyCondition, msg)
+	r.Event(obj, eventv1.EventTypeTrace, gotkmeta.ReadyCondition, msg)
+
+	r.notify(oldObj, obj, eaRefs)
 
 	return ctrl.Result{RequeueAfter: gotkjitter.JitteredIntervalDuration(obj.GetRequeueAfter())}, nil
 }
 
+// notify emits notification related to the result of reconciliation. It will only send events if
+// there is a least one external artifact update
+func (r *ArtifactGeneratorReconciler) notify(oldObj, newObj *swapi.ArtifactGenerator, eaRefs []swapi.ExternalArtifactReference) {
+	eaChanged := make([]string, 0)
+
+	for _, eaRef := range eaRefs {
+		if !oldObj.HasArtifactInInventory(eaRef.Name, eaRef.Namespace, eaRef.Digest) {
+			eaChanged = append(eaChanged, fmt.Sprintf("%s/%s (%s)", eaRef.Namespace, eaRef.Name, eaRef.Digest))
+		}
+	}
+
+	if len(eaChanged) > 0 {
+		msg := fmt.Sprintf("external artifacts reconciled: %s", strings.Join(eaChanged, "\n"))
+		r.Event(newObj, corev1.EventTypeNormal, gotkmeta.ReadyCondition, msg)
+	}
+}
+
 // observeSources retrieves the current state of sources,
 // including their artifact URLs, digests, and revisions.
 // It returns a map of source alias to observed state.
@@ -464,7 +485,7 @@ func (r *ArtifactGeneratorReconciler) reconcileExternalArtifact(ctx context.Cont
 		msg := fmt.Sprintf("%s/%s/%s reconciled with revision %s",
 			ea.Kind, ea.Namespace, ea.Name, artifact.Revision)
 		log.Info(msg)
-		r.Event(obj, corev1.EventTypeNormal, gotkmeta.ReadyCondition, msg)
+		r.Event(obj, eventv1.EventTypeTrace, gotkmeta.ReadyCondition, msg)
 	}
 
 	return &swapi.ExternalArtifactReference{