Skip to content

Commit 3fd7d4e

Browse files
adri1197adri1197
committed
Migrate event recorder to event/v1
Encapsulate event recording under fluxcd/pkg/runtime/events and extract the duplicated eventLogf helper into reconcile.EventLogf in internal/reconcile/event.go Signed-off-by: Adrian Fernandez De La Torre <adri1197@gmail.com>
1 parent 8721122 commit 3fd7d4e

17 files changed

Lines changed: 359 additions & 302 deletions

‎internal/controller/bucket_controller.go‎

Lines changed: 12 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -32,9 +32,7 @@ import (
3232
"golang.org/x/sync/errgroup"
3333
"golang.org/x/sync/semaphore"
3434
corev1 "k8s.io/api/core/v1"
35-
"k8s.io/apimachinery/pkg/runtime"
3635
"k8s.io/apimachinery/pkg/types"
37-
kuberecorder "k8s.io/client-go/tools/record"
3836
"k8s.io/client-go/util/workqueue"
3937
ctrl "sigs.k8s.io/controller-runtime"
4038
"sigs.k8s.io/controller-runtime/pkg/client"
@@ -43,14 +41,15 @@ import (
4341
"sigs.k8s.io/controller-runtime/pkg/predicate"
4442
"sigs.k8s.io/controller-runtime/pkg/reconcile"
4543

46-
eventv1 "github.com/fluxcd/pkg/apis/event/v1beta1"
44+
eventv1 "github.com/fluxcd/pkg/apis/event/v1"
4745
"github.com/fluxcd/pkg/apis/meta"
4846
intdigest "github.com/fluxcd/pkg/artifact/digest"
4947
"github.com/fluxcd/pkg/artifact/storage"
5048
"github.com/fluxcd/pkg/auth"
5149
"github.com/fluxcd/pkg/cache"
5250
"github.com/fluxcd/pkg/runtime/conditions"
5351
helper "github.com/fluxcd/pkg/runtime/controller"
52+
"github.com/fluxcd/pkg/runtime/events"
5453
"github.com/fluxcd/pkg/runtime/jitter"
5554
"github.com/fluxcd/pkg/runtime/patch"
5655
"github.com/fluxcd/pkg/runtime/predicates"
@@ -126,7 +125,7 @@ var bucketFailConditions = []string{
126125
// BucketReconciler reconciles a v1.Bucket object.
127126
type BucketReconciler struct {
128127
client.Client
129-
kuberecorder.EventRecorder
128+
events.EventRecorder
130129
helper.Metrics
131130

132131
Storage *storage.Storage
@@ -348,13 +347,13 @@ func (r *BucketReconciler) notify(ctx context.Context, oldObj, newObj *sourcev1.
348347

349348
// Notify on new artifact and failure recovery.
350349
if !oldObj.GetArtifact().HasDigest(newObj.GetArtifact().Digest) {
351-
r.AnnotatedEventf(newObj, annotations, corev1.EventTypeNormal,
352-
"NewArtifact", "%s", message)
350+
r.AnnotatedEventf(newObj, nil, annotations, corev1.EventTypeNormal,
351+
"NewArtifact", eventv1.ActionApplied, "%s", message)
353352
ctrl.LoggerFrom(ctx).Info(message)
354353
} else {
355354
if sreconcile.FailureRecovery(oldObj, newObj, bucketFailConditions) {
356-
r.AnnotatedEventf(newObj, annotations, corev1.EventTypeNormal,
357-
meta.SucceededReason, "%s", message)
355+
r.AnnotatedEventf(newObj, nil, annotations, corev1.EventTypeNormal,
356+
meta.SucceededReason, eventv1.ActionReconciled, "%s", message)
358357
ctrl.LoggerFrom(ctx).Info(message)
359358
}
360359
}
@@ -388,7 +387,7 @@ func (r *BucketReconciler) reconcileStorage(ctx context.Context, sp *patch.Seria
388387
// matches the actual artifact
389388
if !artifactMissing {
390389
if err := r.Storage.VerifyArtifact(*artifact); err != nil {
391-
r.Eventf(obj, corev1.EventTypeWarning, "ArtifactVerificationFailed", "failed to verify integrity of artifact: %s", err.Error())
390+
r.Eventf(obj, nil, corev1.EventTypeWarning, "ArtifactVerificationFailed", eventv1.ActionFailed, "failed to verify integrity of artifact: %s", err.Error())
392391

393392
if err = r.Storage.Remove(*artifact); err != nil {
394393
return sreconcile.ResultEmpty, fmt.Errorf("failed to remove artifact after digest mismatch: %w", err)
@@ -525,7 +524,7 @@ func (r *BucketReconciler) reconcileArtifact(ctx context.Context, sp *patch.Seri
525524
if curArtifact := obj.GetArtifact(); curArtifact != nil && curArtifact.Revision != "" {
526525
curRev := digest.Digest(curArtifact.Revision)
527526
if curRev.Validate() == nil && index.Digest(curRev.Algorithm()) == curRev {
528-
r.eventLogf(ctx, obj, eventv1.EventTypeTrace, sourcev1.ArtifactUpToDateReason, "artifact up-to-date with remote revision: '%s'", artifact.Revision)
527+
sreconcile.EventLogf(ctx, r, obj, eventv1.EventTypeTrace, sourcev1.ArtifactUpToDateReason, "artifact up-to-date with remote revision: '%s'", artifact.Revision)
529528
return sreconcile.ResultSuccess, nil
530529
}
531530
}
@@ -582,7 +581,7 @@ func (r *BucketReconciler) reconcileArtifact(ctx context.Context, sp *patch.Seri
582581
// Update symlink on a "best effort" basis
583582
url, err := r.Storage.Symlink(artifact, "latest.tar.gz")
584583
if err != nil {
585-
r.eventLogf(ctx, obj, eventv1.EventTypeTrace, sourcev1.SymlinkUpdateFailedReason,
584+
sreconcile.EventLogf(ctx, r, obj, eventv1.EventTypeTrace, sourcev1.SymlinkUpdateFailedReason,
586585
"failed to update status URL symlink: %s", err)
587586
}
588587
if url != "" {
@@ -626,7 +625,7 @@ func (r *BucketReconciler) garbageCollect(ctx context.Context, obj *sourcev1.Buc
626625
"GarbageCollectionFailed",
627626
)
628627
} else if deleted != "" {
629-
r.eventLogf(ctx, obj, eventv1.EventTypeTrace, "GarbageCollectionSucceeded",
628+
sreconcile.EventLogf(ctx, r, obj, eventv1.EventTypeTrace, "GarbageCollectionSucceeded",
630629
"garbage collected artifacts for deleted resource")
631630
}
632631
obj.Status.Artifact = nil
@@ -641,40 +640,14 @@ func (r *BucketReconciler) garbageCollect(ctx context.Context, obj *sourcev1.Buc
641640
)
642641
}
643642
if len(delFiles) > 0 {
644-
r.eventLogf(ctx, obj, eventv1.EventTypeTrace, "GarbageCollectionSucceeded",
643+
sreconcile.EventLogf(ctx, r, obj, eventv1.EventTypeTrace, "GarbageCollectionSucceeded",
645644
"garbage collected %d artifacts", len(delFiles))
646645
return nil
647646
}
648647
}
649648
return nil
650649
}
651650

652-
// eventLogf records events, and logs at the same time.
653-
//
654-
// This log is different from the debug log in the EventRecorder, in the sense
655-
// that this is a simple log. While the debug log contains complete details
656-
// about the event.
657-
func (r *BucketReconciler) eventLogf(ctx context.Context, obj runtime.Object, eventType string, reason string, messageFmt string, args ...interface{}) {
658-
r.annotatedEventLogf(ctx, obj, nil, eventType, reason, messageFmt, args...)
659-
}
660-
661-
// annotatedEventLogf records annotated events, and logs at the same time.
662-
//
663-
// This log is different from the debug log in the EventRecorder, in the sense
664-
// that this is a simple log. While the debug log contains complete details
665-
// about the event.
666-
func (r *BucketReconciler) annotatedEventLogf(ctx context.Context,
667-
obj runtime.Object, annotations map[string]string, eventType string, reason string, messageFmt string, args ...interface{}) {
668-
msg := fmt.Sprintf(messageFmt, args...)
669-
// Log and emit event.
670-
if eventType == corev1.EventTypeWarning {
671-
ctrl.LoggerFrom(ctx).Error(errors.New(reason), msg)
672-
} else {
673-
ctrl.LoggerFrom(ctx).Info(msg)
674-
}
675-
r.AnnotatedEventf(obj, annotations, eventType, reason, "%s", msg)
676-
}
677-
678651
// fetchEtagIndex fetches the current etagIndex for the in the obj specified
679652
// bucket using the given provider, while filtering them using .sourceignore
680653
// rules. After fetching an object, the etag value in the index is updated to

‎internal/controller/bucket_controller_test.go‎

Lines changed: 37 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -30,19 +30,22 @@ import (
3030

3131
. "github.com/onsi/gomega"
3232
corev1 "k8s.io/api/core/v1"
33+
eventsv1 "k8s.io/api/events/v1"
3334
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
34-
"k8s.io/client-go/tools/record"
3535
ctrl "sigs.k8s.io/controller-runtime"
3636
"sigs.k8s.io/controller-runtime/pkg/client"
3737
fakeclient "sigs.k8s.io/controller-runtime/pkg/client/fake"
3838

3939
kstatus "github.com/fluxcd/cli-utils/pkg/kstatus/status"
40+
41+
eventv1 "github.com/fluxcd/pkg/apis/event/v1"
4042
"github.com/fluxcd/pkg/apis/meta"
4143
intdigest "github.com/fluxcd/pkg/artifact/digest"
4244
"github.com/fluxcd/pkg/artifact/storage"
4345
"github.com/fluxcd/pkg/auth"
4446
"github.com/fluxcd/pkg/runtime/conditions"
4547
conditionscheck "github.com/fluxcd/pkg/runtime/conditions/check"
48+
"github.com/fluxcd/pkg/runtime/events"
4649
"github.com/fluxcd/pkg/runtime/jitter"
4750
"github.com/fluxcd/pkg/runtime/patch"
4851

@@ -85,7 +88,7 @@ func TestBucketReconciler_deleteBeforeFinalizer(t *testing.T) {
8588

8689
r := &BucketReconciler{
8790
Client: k8sClient,
88-
EventRecorder: record.NewFakeRecorder(32),
91+
EventRecorder: events.NewFakeRecorder(32, false),
8992
Storage: testStorage,
9093
}
9194
// NOTE: Only a real API server responds with an error in this scenario.
@@ -383,7 +386,7 @@ func TestBucketReconciler_reconcileStorage(t *testing.T) {
383386
WithScheme(testEnv.GetScheme()).
384387
WithStatusSubresource(&sourcev1.Bucket{}).
385388
Build(),
386-
EventRecorder: record.NewFakeRecorder(32),
389+
EventRecorder: events.NewFakeRecorder(32, false),
387390
Storage: testStorage,
388391
patchOptions: getPatchOptions(bucketReadyCondition.Owned, "sc"),
389392
}
@@ -918,7 +921,7 @@ func TestBucketReconciler_reconcileSource_generic(t *testing.T) {
918921
}
919922

920923
r := &BucketReconciler{
921-
EventRecorder: record.NewFakeRecorder(32),
924+
EventRecorder: events.NewFakeRecorder(32, false),
922925
Client: clientBuilder.Build(),
923926
Storage: testStorage,
924927
patchOptions: getPatchOptions(bucketReadyCondition.Owned, "sc"),
@@ -1385,7 +1388,7 @@ func TestBucketReconciler_reconcileSource_gcs(t *testing.T) {
13851388
}
13861389

13871390
r := &BucketReconciler{
1388-
EventRecorder: record.NewFakeRecorder(32),
1391+
EventRecorder: events.NewFakeRecorder(32, false),
13891392
Client: clientBuilder.Build(),
13901393
Storage: testStorage,
13911394
patchOptions: getPatchOptions(bucketReadyCondition.Owned, "sc"),
@@ -1589,7 +1592,7 @@ func TestBucketReconciler_reconcileArtifact(t *testing.T) {
15891592

15901593
r := &BucketReconciler{
15911594
Client: clientBuilder.Build(),
1592-
EventRecorder: record.NewFakeRecorder(32),
1595+
EventRecorder: events.NewFakeRecorder(32, false),
15931596
Storage: testStorage,
15941597
patchOptions: getPatchOptions(bucketReadyCondition.Owned, "sc"),
15951598
}
@@ -1712,7 +1715,7 @@ func TestBucketReconciler_statusConditions(t *testing.T) {
17121715
}
17131716

17141717
ctx := context.TODO()
1715-
summarizeHelper := summarize.NewHelper(record.NewFakeRecorder(32), serialPatcher)
1718+
summarizeHelper := summarize.NewHelper(events.NewFakeRecorder(32, false), serialPatcher)
17161719
summarizeOpts := []summarize.Option{
17171720
summarize.WithConditions(bucketReadyCondition),
17181721
summarize.WithReconcileResult(sreconcile.ResultSuccess),
@@ -1739,7 +1742,7 @@ func TestBucketReconciler_notify(t *testing.T) {
17391742
resErr error
17401743
oldObjBeforeFunc func(obj *sourcev1.Bucket)
17411744
newObjBeforeFunc func(obj *sourcev1.Bucket)
1742-
wantEvent string
1745+
wantEvent *eventsv1.Event
17431746
}{
17441747
{
17451748
name: "error - no event",
@@ -1753,7 +1756,12 @@ func TestBucketReconciler_notify(t *testing.T) {
17531756
newObjBeforeFunc: func(obj *sourcev1.Bucket) {
17541757
obj.Status.Artifact = &meta.Artifact{Revision: "xxx", Digest: "yyy"}
17551758
},
1756-
wantEvent: "Normal NewArtifact stored artifact with 2 fetched files from",
1759+
wantEvent: &eventsv1.Event{
1760+
Type: "Normal",
1761+
Reason: "NewArtifact",
1762+
Action: eventv1.ActionApplied,
1763+
Note: "stored artifact with 2 fetched files from",
1764+
},
17571765
},
17581766
{
17591767
name: "recovery from failure",
@@ -1768,7 +1776,12 @@ func TestBucketReconciler_notify(t *testing.T) {
17681776
obj.Status.Artifact = &meta.Artifact{Revision: "xxx", Digest: "yyy"}
17691777
conditions.MarkTrue(obj, meta.ReadyCondition, meta.SucceededReason, "ready")
17701778
},
1771-
wantEvent: "Normal Succeeded stored artifact with 2 fetched files from",
1779+
wantEvent: &eventsv1.Event{
1780+
Type: "Normal",
1781+
Reason: "Succeeded",
1782+
Action: eventv1.ActionReconciled,
1783+
Note: "stored artifact with 2 fetched files from",
1784+
},
17721785
},
17731786
{
17741787
name: "recovery and new artifact",
@@ -1783,7 +1796,12 @@ func TestBucketReconciler_notify(t *testing.T) {
17831796
obj.Status.Artifact = &meta.Artifact{Revision: "aaa", Digest: "bbb"}
17841797
conditions.MarkTrue(obj, meta.ReadyCondition, meta.SucceededReason, "ready")
17851798
},
1786-
wantEvent: "Normal NewArtifact stored artifact with 2 fetched files from",
1799+
wantEvent: &eventsv1.Event{
1800+
Type: "Normal",
1801+
Reason: "NewArtifact",
1802+
Action: eventv1.ActionApplied,
1803+
Note: "stored artifact with 2 fetched files from",
1804+
},
17871805
},
17881806
{
17891807
name: "no updates",
@@ -1804,7 +1822,7 @@ func TestBucketReconciler_notify(t *testing.T) {
18041822
t.Run(tt.name, func(t *testing.T) {
18051823
g := NewWithT(t)
18061824

1807-
recorder := record.NewFakeRecorder(32)
1825+
recorder := events.NewFakeRecorder(32, false)
18081826

18091827
oldObj := &sourcev1.Bucket{
18101828
Spec: sourcev1.BucketSpec{
@@ -1832,12 +1850,15 @@ func TestBucketReconciler_notify(t *testing.T) {
18321850

18331851
select {
18341852
case x, ok := <-recorder.Events:
1835-
g.Expect(ok).To(Equal(tt.wantEvent != ""), "unexpected event received")
1836-
if tt.wantEvent != "" {
1837-
g.Expect(x).To(ContainSubstring(tt.wantEvent))
1853+
g.Expect(ok).To(Equal(tt.wantEvent != nil), "unexpected event received")
1854+
if tt.wantEvent != nil {
1855+
g.Expect(x.Type).To(Equal(tt.wantEvent.Type))
1856+
g.Expect(x.Reason).To(Equal(tt.wantEvent.Reason))
1857+
g.Expect(x.Action).To(Equal(tt.wantEvent.Action))
1858+
g.Expect(x.Note).To(ContainSubstring(tt.wantEvent.Note))
18381859
}
18391860
default:
1840-
if tt.wantEvent != "" {
1861+
if tt.wantEvent != nil {
18411862
t.Errorf("expected some event to be emitted")
18421863
}
18431864
}

0 commit comments

Comments
 (0)