diff --git a/controllers/gitrepository_controller.go b/controllers/gitrepository_controller.go index a47b1811..cb5c4761 100644 --- a/controllers/gitrepository_controller.go +++ b/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 + } + } +} diff --git a/controllers/helmchart_controller.go b/controllers/helmchart_controller.go index 54141406..b288a8fb 100644 --- a/controllers/helmchart_controller.go +++ b/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 + } + } +} diff --git a/controllers/helmrepository_controller.go b/controllers/helmrepository_controller.go index 925dc920..fa7ef13e 100644 --- a/controllers/helmrepository_controller.go +++ b/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 + } + } +} diff --git a/go.sum b/go.sum index 746da900..ee65bc8f 100644 --- a/go.sum +++ b/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= diff --git a/main.go b/main.go index b8e29de5..72ba800c 100644 --- a/main.go +++ b/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 {