Skip to content

Commit 06c5782

Browse files
stefanprodanstefanprodan
authored
Merge pull request #69 from fluxcd/events
Implement event recording
2 parents 136124a + c9681ae commit 06c5782

File tree

7 files changed

+347
-121
lines changed

7 files changed

+347
-121
lines changed

‎config/manager/deployment.yaml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ spec:
1515
app: source-controller
1616
annotations:
1717
prometheus.io/scrape: "true"
18-
prometheus.io/port: "9090"
18+
prometheus.io/port: "8080"
1919
spec:
2020
terminationGracePeriodSeconds: 10
2121
containers:
@@ -26,9 +26,9 @@ spec:
2626
allowPrivilegeEscalation: false
2727
readOnlyRootFilesystem: true
2828
ports:
29-
- containerPort: 8080
30-
name: http
3129
- containerPort: 9090
30+
name: http
31+
- containerPort: 8080
3232
name: http-prom
3333
args:
3434
- --enable-leader-election

‎controllers/gitrepository_controller.go‎

Lines changed: 51 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
"fmt"
2222
"io/ioutil"
2323
"os"
24+
"strings"
2425

2526
"github.com/go-git/go-git/v5/plumbing/object"
2627
"github.com/go-git/go-git/v5/plumbing/transport"
@@ -29,20 +30,25 @@ import (
2930
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
3031
"k8s.io/apimachinery/pkg/runtime"
3132
"k8s.io/apimachinery/pkg/types"
33+
kuberecorder "k8s.io/client-go/tools/record"
34+
"k8s.io/client-go/tools/reference"
3235
ctrl "sigs.k8s.io/controller-runtime"
3336
"sigs.k8s.io/controller-runtime/pkg/client"
3437
"sigs.k8s.io/controller-runtime/pkg/controller"
3538

39+
"github.com/fluxcd/pkg/recorder"
3640
sourcev1 "github.com/fluxcd/source-controller/api/v1alpha1"
3741
intgit "github.com/fluxcd/source-controller/internal/git"
3842
)
3943

4044
// GitRepositoryReconciler reconciles a GitRepository object
4145
type GitRepositoryReconciler struct {
4246
client.Client
43-
Log logr.Logger
44-
Scheme *runtime.Scheme
45-
Storage *Storage
47+
Log logr.Logger
48+
Scheme *runtime.Scheme
49+
Storage *Storage
50+
EventRecorder kuberecorder.EventRecorder
51+
ExternalEventRecorder *recorder.EventRecorder
4652
}
4753

4854
// +kubebuilder:rbac:groups=source.fluxcd.io,resources=gitrepositories,verbs=get;list;watch;create;update;patch;delete
@@ -51,42 +57,48 @@ type GitRepositoryReconciler struct {
5157
func (r *GitRepositoryReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
5258
ctx := context.Background()
5359

54-
var repo sourcev1.GitRepository
55-
if err := r.Get(ctx, req.NamespacedName, &repo); err != nil {
60+
var repository sourcev1.GitRepository
61+
if err := r.Get(ctx, req.NamespacedName, &repository); err != nil {
5662
return ctrl.Result{}, client.IgnoreNotFound(err)
5763
}
5864

59-
log := r.Log.WithValues(repo.Kind, req.NamespacedName)
65+
log := r.Log.WithValues(repository.Kind, req.NamespacedName)
6066

6167
// set initial status
62-
if reset, status := r.shouldResetStatus(repo); reset {
68+
if reset, status := r.shouldResetStatus(repository); reset {
6369
log.Info("Initializing Git repository")
64-
repo.Status = status
65-
if err := r.Status().Update(ctx, &repo); err != nil {
70+
repository.Status = status
71+
if err := r.Status().Update(ctx, &repository); err != nil {
6672
log.Error(err, "unable to update GitRepository status")
6773
return ctrl.Result{Requeue: true}, err
6874
}
6975
} else {
70-
repo = sourcev1.GitRepositoryProgressing(repo)
71-
if err := r.Status().Update(ctx, &repo); err != nil {
76+
repository = sourcev1.GitRepositoryProgressing(repository)
77+
if err := r.Status().Update(ctx, &repository); err != nil {
7278
log.Error(err, "unable to update GitRepository status")
7379
return ctrl.Result{Requeue: true}, err
7480
}
7581
}
7682

7783
// try to remove old artifacts
78-
if err := r.gc(repo); err != nil {
84+
if err := r.gc(repository); err != nil {
7985
log.Error(err, "artifacts GC failed")
8086
}
8187

8288
// try git sync
83-
syncedRepo, err := r.sync(ctx, *repo.DeepCopy())
89+
syncedRepo, err := r.sync(ctx, *repository.DeepCopy())
8490
if err != nil {
8591
log.Error(err, "Git repository sync failed")
92+
r.event(repository, recorder.EventSeverityError, err.Error())
8693
if err := r.Status().Update(ctx, &syncedRepo); err != nil {
8794
log.Error(err, "unable to update GitRepository status")
8895
}
8996
return ctrl.Result{Requeue: true}, err
97+
} else {
98+
// emit revision change event
99+
if repository.Status.Artifact == nil || syncedRepo.Status.Artifact.Revision != repository.Status.Artifact.Revision {
100+
r.event(syncedRepo, recorder.EventSeverityInfo, sourcev1.GitRepositoryReadyMessage(syncedRepo))
101+
}
90102
}
91103

92104
// update status
@@ -98,7 +110,7 @@ func (r *GitRepositoryReconciler) Reconcile(req ctrl.Request) (ctrl.Result, erro
98110
log.Info("Git repository sync succeeded", "msg", sourcev1.GitRepositoryReadyMessage(syncedRepo))
99111

100112
// requeue repository
101-
return ctrl.Result{RequeueAfter: repo.GetInterval().Duration}, nil
113+
return ctrl.Result{RequeueAfter: repository.GetInterval().Duration}, nil
102114
}
103115

104116
type GitRepositoryReconcilerOptions struct {
@@ -260,3 +272,28 @@ func (r *GitRepositoryReconciler) gc(repository sourcev1.GitRepository) error {
260272
}
261273
return nil
262274
}
275+
276+
// emit Kubernetes event and forward event to notification controller if configured
277+
func (r *GitRepositoryReconciler) event(repository sourcev1.GitRepository, severity, msg string) {
278+
if r.EventRecorder != nil {
279+
r.EventRecorder.Eventf(&repository, "Normal", severity, msg)
280+
}
281+
if r.ExternalEventRecorder != nil {
282+
objRef, err := reference.GetReference(r.Scheme, &repository)
283+
if err != nil {
284+
r.Log.WithValues(
285+
strings.ToLower(repository.Kind),
286+
fmt.Sprintf("%s/%s", repository.GetNamespace(), repository.GetName()),
287+
).Error(err, "unable to send event")
288+
return
289+
}
290+
291+
if err := r.ExternalEventRecorder.Eventf(*objRef, nil, severity, severity, msg); err != nil {
292+
r.Log.WithValues(
293+
strings.ToLower(repository.Kind),
294+
fmt.Sprintf("%s/%s", repository.GetNamespace(), repository.GetName()),
295+
).Error(err, "unable to send event")
296+
return
297+
}
298+
}
299+
}

‎controllers/helmchart_controller.go‎

Lines changed: 41 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
"fmt"
2222
"io/ioutil"
2323
"net/url"
24+
"strings"
2425

2526
"github.com/go-logr/logr"
2627
"helm.sh/helm/v3/pkg/getter"
@@ -29,22 +30,27 @@ import (
2930
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
3031
"k8s.io/apimachinery/pkg/runtime"
3132
"k8s.io/apimachinery/pkg/types"
33+
kuberecorder "k8s.io/client-go/tools/record"
34+
"k8s.io/client-go/tools/reference"
3235
ctrl "sigs.k8s.io/controller-runtime"
3336
"sigs.k8s.io/controller-runtime/pkg/client"
3437
"sigs.k8s.io/controller-runtime/pkg/controller"
3538
"sigs.k8s.io/yaml"
3639

40+
"github.com/fluxcd/pkg/recorder"
3741
sourcev1 "github.com/fluxcd/source-controller/api/v1alpha1"
3842
"github.com/fluxcd/source-controller/internal/helm"
3943
)
4044

4145
// HelmChartReconciler reconciles a HelmChart object
4246
type HelmChartReconciler struct {
4347
client.Client
44-
Log logr.Logger
45-
Scheme *runtime.Scheme
46-
Storage *Storage
47-
Getters getter.Providers
48+
Log logr.Logger
49+
Scheme *runtime.Scheme
50+
Storage *Storage
51+
Getters getter.Providers
52+
EventRecorder kuberecorder.EventRecorder
53+
ExternalEventRecorder *recorder.EventRecorder
4854
}
4955

5056
// +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) {
101107
pulledChart, err := r.sync(ctx, repository, *chart.DeepCopy())
102108
if err != nil {
103109
log.Error(err, "Helm chart sync failed")
110+
r.event(chart, recorder.EventSeverityError, err.Error())
104111
if err := r.Status().Update(ctx, &pulledChart); err != nil {
105112
log.Error(err, "unable to update HelmChart status")
106113
}
107114
return ctrl.Result{Requeue: true}, err
115+
} else {
116+
// emit version change event
117+
if chart.Status.Artifact == nil || pulledChart.Status.Artifact.Revision != chart.Status.Artifact.Revision {
118+
r.event(pulledChart, recorder.EventSeverityInfo, sourcev1.HelmChartReadyMessage(pulledChart))
119+
}
108120
}
109121

110122
// update status
@@ -326,3 +338,28 @@ func (r *HelmChartReconciler) setOwnerRef(ctx context.Context, chart *sourcev1.H
326338
)))
327339
return r.Update(ctx, chart)
328340
}
341+
342+
// emit Kubernetes event and forward event to notification controller if configured
343+
func (r *HelmChartReconciler) event(chart sourcev1.HelmChart, severity, msg string) {
344+
if r.EventRecorder != nil {
345+
r.EventRecorder.Eventf(&chart, "Normal", severity, msg)
346+
}
347+
if r.ExternalEventRecorder != nil {
348+
objRef, err := reference.GetReference(r.Scheme, &chart)
349+
if err != nil {
350+
r.Log.WithValues(
351+
strings.ToLower(chart.Kind),
352+
fmt.Sprintf("%s/%s", chart.GetNamespace(), chart.GetName()),
353+
).Error(err, "unable to send event")
354+
return
355+
}
356+
357+
if err := r.ExternalEventRecorder.Eventf(*objRef, nil, severity, severity, msg); err != nil {
358+
r.Log.WithValues(
359+
strings.ToLower(chart.Kind),
360+
fmt.Sprintf("%s/%s", chart.GetNamespace(), chart.GetName()),
361+
).Error(err, "unable to send event")
362+
return
363+
}
364+
}
365+
}

‎controllers/helmrepository_controller.go‎

Lines changed: 41 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222
"io/ioutil"
2323
"net/url"
2424
"path"
25+
"strings"
2526

2627
"github.com/go-logr/logr"
2728
"helm.sh/helm/v3/pkg/getter"
@@ -30,22 +31,27 @@ import (
3031
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
3132
"k8s.io/apimachinery/pkg/runtime"
3233
"k8s.io/apimachinery/pkg/types"
34+
kuberecorder "k8s.io/client-go/tools/record"
35+
"k8s.io/client-go/tools/reference"
3336
ctrl "sigs.k8s.io/controller-runtime"
3437
"sigs.k8s.io/controller-runtime/pkg/client"
3538
"sigs.k8s.io/controller-runtime/pkg/controller"
3639
"sigs.k8s.io/yaml"
3740

41+
"github.com/fluxcd/pkg/recorder"
3842
sourcev1 "github.com/fluxcd/source-controller/api/v1alpha1"
3943
"github.com/fluxcd/source-controller/internal/helm"
4044
)
4145

4246
// HelmRepositoryReconciler reconciles a HelmRepository object
4347
type HelmRepositoryReconciler struct {
4448
client.Client
45-
Log logr.Logger
46-
Scheme *runtime.Scheme
47-
Storage *Storage
48-
Getters getter.Providers
49+
Log logr.Logger
50+
Scheme *runtime.Scheme
51+
Storage *Storage
52+
Getters getter.Providers
53+
EventRecorder kuberecorder.EventRecorder
54+
ExternalEventRecorder *recorder.EventRecorder
4955
}
5056

5157
// +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
8793
syncedRepo, err := r.sync(ctx, *repository.DeepCopy())
8894
if err != nil {
8995
log.Error(err, "Helm repository sync failed")
96+
r.event(repository, recorder.EventSeverityError, err.Error())
9097
if err := r.Status().Update(ctx, &syncedRepo); err != nil {
9198
log.Error(err, "unable to update HelmRepository status")
9299
}
93100
return ctrl.Result{Requeue: true}, err
101+
} else {
102+
// emit revision change event
103+
if repository.Status.Artifact == nil || syncedRepo.Status.Artifact.Revision != repository.Status.Artifact.Revision {
104+
r.event(syncedRepo, recorder.EventSeverityInfo, sourcev1.HelmRepositoryReadyMessage(syncedRepo))
105+
}
94106
}
95107

96108
// update status
@@ -255,3 +267,28 @@ func (r *HelmRepositoryReconciler) gc(repository sourcev1.HelmRepository) error
255267
}
256268
return nil
257269
}
270+
271+
// emit Kubernetes event and forward event to notification controller if configured
272+
func (r *HelmRepositoryReconciler) event(repository sourcev1.HelmRepository, severity, msg string) {
273+
if r.EventRecorder != nil {
274+
r.EventRecorder.Eventf(&repository, "Normal", severity, msg)
275+
}
276+
if r.ExternalEventRecorder != nil {
277+
objRef, err := reference.GetReference(r.Scheme, &repository)
278+
if err != nil {
279+
r.Log.WithValues(
280+
strings.ToLower(repository.Kind),
281+
fmt.Sprintf("%s/%s", repository.GetNamespace(), repository.GetName()),
282+
).Error(err, "unable to send event")
283+
return
284+
}
285+
286+
if err := r.ExternalEventRecorder.Eventf(*objRef, nil, severity, severity, msg); err != nil {
287+
r.Log.WithValues(
288+
strings.ToLower(repository.Kind),
289+
fmt.Sprintf("%s/%s", repository.GetNamespace(), repository.GetName()),
290+
).Error(err, "unable to send event")
291+
return
292+
}
293+
}
294+
}

‎go.mod‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -4,21 +4,20 @@ go 1.13
44

55
require (
66
github.com/blang/semver v3.5.0+incompatible
7-
github.com/fluxcd/pkg v0.0.1
7+
github.com/fluxcd/pkg v0.0.2
88
github.com/go-git/go-billy/v5 v5.0.0
99
github.com/go-git/go-git/v5 v5.1.0
1010
github.com/go-logr/logr v0.1.0
1111
github.com/onsi/ginkgo v1.11.0
1212
github.com/onsi/gomega v1.8.1
1313
github.com/sosedoff/gitkit v0.2.1-0.20191202022816-7182d43c6254
14-
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
15-
helm.sh/helm/v3 v3.1.2
16-
k8s.io/api v0.17.2
17-
k8s.io/apimachinery v0.17.2
18-
k8s.io/client-go v0.17.2
14+
helm.sh/helm/v3 v3.2.4
15+
k8s.io/api v0.18.4
16+
k8s.io/apimachinery v0.18.4
17+
k8s.io/client-go v0.18.4
1918
rsc.io/letsencrypt v0.0.3 // indirect
20-
sigs.k8s.io/controller-runtime v0.5.0
21-
sigs.k8s.io/yaml v1.1.0
19+
sigs.k8s.io/controller-runtime v0.6.0
20+
sigs.k8s.io/yaml v1.2.0
2221
)
2322

2423
// TODO(hidde): drop when PR is accepted:

0 commit comments

Comments
 (0)