Implement event recording

- emit Kubernetes events for artifact acquisition actions
- forward events to notification controller

controllers/gitrepository_controller.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"strings"
 
 	"github.com/go-git/go-git/v5/plumbing/object"
 	"github.com/go-git/go-git/v5/plumbing/transport"
@@ -29,10 +30,13 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	kuberecorder "k8s.io/client-go/tools/record"
+	"k8s.io/client-go/tools/reference"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 
+	"github.com/fluxcd/pkg/recorder"
 	sourcev1 "github.com/fluxcd/source-controller/api/v1alpha1"
 	intgit "github.com/fluxcd/source-controller/internal/git"
 )
@@ -40,9 +44,11 @@ import (
 // GitRepositoryReconciler reconciles a GitRepository object
 type GitRepositoryReconciler struct {
 	client.Client
-	Log     logr.Logger
-	Scheme  *runtime.Scheme
-	Storage *Storage
+	Log                   logr.Logger
+	Scheme                *runtime.Scheme
+	Storage               *Storage
+	EventRecorder         kuberecorder.EventRecorder
+	ExternalEventRecorder *recorder.EventRecorder
 }
 
 // +kubebuilder:rbac:groups=source.fluxcd.io,resources=gitrepositories,verbs=get;list;watch;create;update;patch;delete
@@ -51,42 +57,48 @@ type GitRepositoryReconciler struct {
 func (r *GitRepositoryReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 	ctx := context.Background()
 
-	var repo sourcev1.GitRepository
-	if err := r.Get(ctx, req.NamespacedName, &repo); err != nil {
+	var repository sourcev1.GitRepository
+	if err := r.Get(ctx, req.NamespacedName, &repository); err != nil {
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	log := r.Log.WithValues(repo.Kind, req.NamespacedName)
+	log := r.Log.WithValues(repository.Kind, req.NamespacedName)
 
 	// set initial status
-	if reset, status := r.shouldResetStatus(repo); reset {
+	if reset, status := r.shouldResetStatus(repository); reset {
 		log.Info("Initializing Git repository")
-		repo.Status = status
-		if err := r.Status().Update(ctx, &repo); err != nil {
+		repository.Status = status
+		if err := r.Status().Update(ctx, &repository); err != nil {
 			log.Error(err, "unable to update GitRepository status")
 			return ctrl.Result{Requeue: true}, err
 		}
 	} else {
-		repo = sourcev1.GitRepositoryProgressing(repo)
-		if err := r.Status().Update(ctx, &repo); err != nil {
+		repository = sourcev1.GitRepositoryProgressing(repository)
+		if err := r.Status().Update(ctx, &repository); err != nil {
 			log.Error(err, "unable to update GitRepository status")
 			return ctrl.Result{Requeue: true}, err
 		}
 	}
 
 	// try to remove old artifacts
-	if err := r.gc(repo); err != nil {
+	if err := r.gc(repository); err != nil {
 		log.Error(err, "artifacts GC failed")
 	}
 
 	// try git sync
-	syncedRepo, err := r.sync(ctx, *repo.DeepCopy())
+	syncedRepo, err := r.sync(ctx, *repository.DeepCopy())
 	if err != nil {
 		log.Error(err, "Git repository sync failed")
+		r.event(repository, recorder.EventSeverityError, err.Error())
 		if err := r.Status().Update(ctx, &syncedRepo); err != nil {
 			log.Error(err, "unable to update GitRepository status")
 		}
 		return ctrl.Result{Requeue: true}, err
+	} else {
+		// emit revision change event
+		if repository.Status.Artifact == nil || syncedRepo.Status.Artifact.Revision != repository.Status.Artifact.Revision {
+			r.event(syncedRepo, recorder.EventSeverityInfo, sourcev1.GitRepositoryReadyMessage(syncedRepo))
+		}
 	}
 
 	// update status
@@ -98,7 +110,7 @@ func (r *GitRepositoryReconciler) Reconcile(req ctrl.Request) (ctrl.Result, erro
 	log.Info("Git repository sync succeeded", "msg", sourcev1.GitRepositoryReadyMessage(syncedRepo))
 
 	// requeue repository
-	return ctrl.Result{RequeueAfter: repo.GetInterval().Duration}, nil
+	return ctrl.Result{RequeueAfter: repository.GetInterval().Duration}, nil
 }
 
 type GitRepositoryReconcilerOptions struct {
@@ -260,3 +272,28 @@ func (r *GitRepositoryReconciler) gc(repository sourcev1.GitRepository) error {
 	}
 	return nil
 }
+
+// emit Kubernetes event and forward event to notification controller if configured
+func (r *GitRepositoryReconciler) event(repository sourcev1.GitRepository, severity, msg string) {
+	if r.EventRecorder != nil {
+		r.EventRecorder.Eventf(&repository, "Normal", severity, msg)
+	}
+	if r.ExternalEventRecorder != nil {
+		objRef, err := reference.GetReference(r.Scheme, &repository)
+		if err != nil {
+			r.Log.WithValues(
+				strings.ToLower(repository.Kind),
+				fmt.Sprintf("%s/%s", repository.GetNamespace(), repository.GetName()),
+			).Error(err, "unable to send event")
+			return
+		}
+
+		if err := r.ExternalEventRecorder.Eventf(*objRef, nil, severity, severity, msg); err != nil {
+			r.Log.WithValues(
+				strings.ToLower(repository.Kind),
+				fmt.Sprintf("%s/%s", repository.GetNamespace(), repository.GetName()),
+			).Error(err, "unable to send event")
+			return
+		}
+	}
+}

controllers/helmchart_controller.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/url"
+	"strings"
 
 	"github.com/go-logr/logr"
 	"helm.sh/helm/v3/pkg/getter"
@@ -29,11 +30,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	kuberecorder "k8s.io/client-go/tools/record"
+	"k8s.io/client-go/tools/reference"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/yaml"
 
+	"github.com/fluxcd/pkg/recorder"
 	sourcev1 "github.com/fluxcd/source-controller/api/v1alpha1"
 	"github.com/fluxcd/source-controller/internal/helm"
 )
@@ -41,10 +45,12 @@ import (
 // HelmChartReconciler reconciles a HelmChart object
 type HelmChartReconciler struct {
 	client.Client
-	Log     logr.Logger
-	Scheme  *runtime.Scheme
-	Storage *Storage
-	Getters getter.Providers
+	Log                   logr.Logger
+	Scheme                *runtime.Scheme
+	Storage               *Storage
+	Getters               getter.Providers
+	EventRecorder         kuberecorder.EventRecorder
+	ExternalEventRecorder *recorder.EventRecorder
 }
 
 // +kubebuilder:rbac:groups=source.fluxcd.io,resources=helmcharts,verbs=get;list;watch;create;update;patch;delete
@@ -101,10 +107,16 @@ func (r *HelmChartReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 	pulledChart, err := r.sync(ctx, repository, *chart.DeepCopy())
 	if err != nil {
 		log.Error(err, "Helm chart sync failed")
+		r.event(chart, recorder.EventSeverityError, err.Error())
 		if err := r.Status().Update(ctx, &pulledChart); err != nil {
 			log.Error(err, "unable to update HelmChart status")
 		}
 		return ctrl.Result{Requeue: true}, err
+	} else {
+		// emit version change event
+		if chart.Status.Artifact == nil || pulledChart.Status.Artifact.Revision != chart.Status.Artifact.Revision {
+			r.event(pulledChart, recorder.EventSeverityInfo, sourcev1.HelmChartReadyMessage(pulledChart))
+		}
 	}
 
 	// update status
@@ -326,3 +338,28 @@ func (r *HelmChartReconciler) setOwnerRef(ctx context.Context, chart *sourcev1.H
 	)))
 	return r.Update(ctx, chart)
 }
+
+// emit Kubernetes event and forward event to notification controller if configured
+func (r *HelmChartReconciler) event(chart sourcev1.HelmChart, severity, msg string) {
+	if r.EventRecorder != nil {
+		r.EventRecorder.Eventf(&chart, "Normal", severity, msg)
+	}
+	if r.ExternalEventRecorder != nil {
+		objRef, err := reference.GetReference(r.Scheme, &chart)
+		if err != nil {
+			r.Log.WithValues(
+				strings.ToLower(chart.Kind),
+				fmt.Sprintf("%s/%s", chart.GetNamespace(), chart.GetName()),
+			).Error(err, "unable to send event")
+			return
+		}
+
+		if err := r.ExternalEventRecorder.Eventf(*objRef, nil, severity, severity, msg); err != nil {
+			r.Log.WithValues(
+				strings.ToLower(chart.Kind),
+				fmt.Sprintf("%s/%s", chart.GetNamespace(), chart.GetName()),
+			).Error(err, "unable to send event")
+			return
+		}
+	}
+}

controllers/helmrepository_controller.go

@@ -22,6 +22,7 @@ import (
 	"io/ioutil"
 	"net/url"
 	"path"
+	"strings"
 
 	"github.com/go-logr/logr"
 	"helm.sh/helm/v3/pkg/getter"
@@ -30,11 +31,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	kuberecorder "k8s.io/client-go/tools/record"
+	"k8s.io/client-go/tools/reference"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/yaml"
 
+	"github.com/fluxcd/pkg/recorder"
 	sourcev1 "github.com/fluxcd/source-controller/api/v1alpha1"
 	"github.com/fluxcd/source-controller/internal/helm"
 )
@@ -42,10 +46,12 @@ import (
 // HelmRepositoryReconciler reconciles a HelmRepository object
 type HelmRepositoryReconciler struct {
 	client.Client
-	Log     logr.Logger
-	Scheme  *runtime.Scheme
-	Storage *Storage
-	Getters getter.Providers
+	Log                   logr.Logger
+	Scheme                *runtime.Scheme
+	Storage               *Storage
+	Getters               getter.Providers
+	EventRecorder         kuberecorder.EventRecorder
+	ExternalEventRecorder *recorder.EventRecorder
 }
 
 // +kubebuilder:rbac:groups=source.fluxcd.io,resources=helmrepositories,verbs=get;list;watch;create;update;patch;delete
@@ -87,10 +93,16 @@ func (r *HelmRepositoryReconciler) Reconcile(req ctrl.Request) (ctrl.Result, err
 	syncedRepo, err := r.sync(ctx, *repository.DeepCopy())
 	if err != nil {
 		log.Error(err, "Helm repository sync failed")
+		r.event(repository, recorder.EventSeverityError, err.Error())
 		if err := r.Status().Update(ctx, &syncedRepo); err != nil {
 			log.Error(err, "unable to update HelmRepository status")
 		}
 		return ctrl.Result{Requeue: true}, err
+	} else {
+		// emit revision change event
+		if repository.Status.Artifact == nil || syncedRepo.Status.Artifact.Revision != repository.Status.Artifact.Revision {
+			r.event(syncedRepo, recorder.EventSeverityInfo, sourcev1.HelmRepositoryReadyMessage(syncedRepo))
+		}
 	}
 
 	// update status
@@ -255,3 +267,28 @@ func (r *HelmRepositoryReconciler) gc(repository sourcev1.HelmRepository) error
 	}
 	return nil
 }
+
+// emit Kubernetes event and forward event to notification controller if configured
+func (r *HelmRepositoryReconciler) event(repository sourcev1.HelmRepository, severity, msg string) {
+	if r.EventRecorder != nil {
+		r.EventRecorder.Eventf(&repository, "Normal", severity, msg)
+	}
+	if r.ExternalEventRecorder != nil {
+		objRef, err := reference.GetReference(r.Scheme, &repository)
+		if err != nil {
+			r.Log.WithValues(
+				strings.ToLower(repository.Kind),
+				fmt.Sprintf("%s/%s", repository.GetNamespace(), repository.GetName()),
+			).Error(err, "unable to send event")
+			return
+		}
+
+		if err := r.ExternalEventRecorder.Eventf(*objRef, nil, severity, severity, msg); err != nil {
+			r.Log.WithValues(
+				strings.ToLower(repository.Kind),
+				fmt.Sprintf("%s/%s", repository.GetNamespace(), repository.GetName()),
+			).Error(err, "unable to send event")
+			return
+		}
+	}
+}

go.sum

@@ -327,11 +327,13 @@ github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t
 github.com/grpc-ecosystem/grpc-gateway v1.9.5/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
 github.com/hashicorp/errwrap v0.0.0-20141028054710-7554cd9344ce/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-cleanhttp v0.5.1 h1:dH3aiDG9Jvb5r5+bYHsikaOUIpcM0xvgMXVoDkXMzJM=
 github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
 github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
 github.com/hashicorp/go-multierror v0.0.0-20161216184304-ed905158d874/go.mod h1:JMRHfdO9jKNzS/+BTlxCjKNQHg/jZAft8U7LloJvN7I=
 github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
 github.com/hashicorp/go-retryablehttp v0.6.4/go.mod h1:vAew36LZh98gCBJNLH42IQ1ER/9wtLZZ8meHqQvEYWY=
+github.com/hashicorp/go-retryablehttp v0.6.6 h1:HJunrbHTDDbBb/ay4kxa1n+dLmttUlnP3V9oNE4hmsM=
 github.com/hashicorp/go-retryablehttp v0.6.6/go.mod h1:vAew36LZh98gCBJNLH42IQ1ER/9wtLZZ8meHqQvEYWY=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1 h1:0hERBMJE1eitiLkihrMvRVBYAkpHzc/J3QdDN+dAcgU=

main.go

@@ -33,6 +33,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
+	"github.com/fluxcd/pkg/recorder"
 	sourcev1 "github.com/fluxcd/source-controller/api/v1alpha1"
 	"github.com/fluxcd/source-controller/controllers"
 	// +kubebuilder:scaffold:imports
@@ -59,6 +60,7 @@ func init() {
 func main() {
 	var (
 		metricsAddr          string
+		eventsAddr           string
 		enableLeaderElection bool
 		storagePath          string
 		storageAddr          string
@@ -67,6 +69,7 @@ func main() {
 	)
 
 	flag.StringVar(&metricsAddr, "metrics-addr", ":9090", "The address the metric endpoint binds to.")
+	flag.StringVar(&eventsAddr, "events-addr", "", "The address of the events receiver.")
 	flag.BoolVar(&enableLeaderElection, "enable-leader-election", false,
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.")
@@ -79,6 +82,16 @@ func main() {
 
 	ctrl.SetLogger(zap.New(zap.UseDevMode(!logJSON)))
 
+	var eventRecorder *recorder.EventRecorder
+	if eventsAddr != "" {
+		if er, err := recorder.NewEventRecorder(eventsAddr, "source-controller"); err != nil {
+			setupLog.Error(err, "unable to create event recorder")
+			os.Exit(1)
+		} else {
+			eventRecorder = er
+		}
+	}
+
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
 		Scheme:             scheme,
 		MetricsBindAddress: metricsAddr,
@@ -97,10 +110,12 @@ func main() {
 	go startFileServer(storage.BasePath, storageAddr, setupLog)
 
 	if err = (&controllers.GitRepositoryReconciler{
-		Client:  mgr.GetClient(),
-		Log:     ctrl.Log.WithName("controllers").WithName("GitRepository"),
-		Scheme:  mgr.GetScheme(),
-		Storage: storage,
+		Client:                mgr.GetClient(),
+		Log:                   ctrl.Log.WithName("controllers").WithName("GitRepository"),
+		Scheme:                mgr.GetScheme(),
+		Storage:               storage,
+		EventRecorder:         mgr.GetEventRecorderFor("source-controller"),
+		ExternalEventRecorder: eventRecorder,
 	}).SetupWithManagerAndOptions(mgr, controllers.GitRepositoryReconcilerOptions{
 		MaxConcurrentReconciles: concurrent,
 	}); err != nil {
@@ -108,11 +123,13 @@ func main() {
 		os.Exit(1)
 	}
 	if err = (&controllers.HelmRepositoryReconciler{
-		Client:  mgr.GetClient(),
-		Log:     ctrl.Log.WithName("controllers").WithName("HelmRepository"),
-		Scheme:  mgr.GetScheme(),
-		Storage: storage,
-		Getters: getters,
+		Client:                mgr.GetClient(),
+		Log:                   ctrl.Log.WithName("controllers").WithName("HelmRepository"),
+		Scheme:                mgr.GetScheme(),
+		Storage:               storage,
+		Getters:               getters,
+		EventRecorder:         mgr.GetEventRecorderFor("source-controller"),
+		ExternalEventRecorder: eventRecorder,
 	}).SetupWithManagerAndOptions(mgr, controllers.HelmRepositoryReconcilerOptions{
 		MaxConcurrentReconciles: concurrent,
 	}); err != nil {
@@ -120,11 +137,13 @@ func main() {
 		os.Exit(1)
 	}
 	if err = (&controllers.HelmChartReconciler{
-		Client:  mgr.GetClient(),
-		Log:     ctrl.Log.WithName("controllers").WithName("HelmChart"),
-		Scheme:  mgr.GetScheme(),
-		Storage: storage,
-		Getters: getters,
+		Client:                mgr.GetClient(),
+		Log:                   ctrl.Log.WithName("controllers").WithName("HelmChart"),
+		Scheme:                mgr.GetScheme(),
+		Storage:               storage,
+		Getters:               getters,
+		EventRecorder:         mgr.GetEventRecorderFor("source-controller"),
+		ExternalEventRecorder: eventRecorder,
 	}).SetupWithManagerAndOptions(mgr, controllers.HelmChartReconcilerOptions{
 		MaxConcurrentReconciles: concurrent,
 	}); err != nil {