Add contextual loggin based recorder (#170)

Signed-off-by: Jian Qiu <[email protected]>

Author: qiujian16

pkg/basecontroller/events/event_recorder.go

@@ -0,0 +1,22 @@
+package events
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	eventsv1 "k8s.io/client-go/kubernetes/typed/events/v1"
+	kevents "k8s.io/client-go/tools/events"
+)
+
+// NewEventRecorder creates a new event recorder for the given controller, it will also log the events
+func NewEventRecorder(ctx context.Context, scheme *runtime.Scheme,
+	eventsClient eventsv1.EventsV1Interface, controllerName string) (kevents.EventRecorder, error) {
+	broadcaster := kevents.NewBroadcaster(&kevents.EventSinkImpl{Interface: eventsClient})
+	err := broadcaster.StartRecordingToSinkWithContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+	broadcaster.StartStructuredLogging(0)
+	recorder := broadcaster.NewRecorder(scheme, controllerName)
+	return recorder, nil
+}

pkg/basecontroller/events/logging_recorder.go

@@ -0,0 +1,54 @@
+package events
+
+import (
+	"context"
+	"fmt"
+
+	"k8s.io/klog/v2"
+)
+
+// ContextualLoggingEventRecorder implements a recorder with contextual logging
+type ContextualLoggingEventRecorder struct {
+	component string
+}
+
+// NewContextualLoggingEventRecorder provides event recorder that will log all recorded events via klog.
+func NewContextualLoggingEventRecorder(component string) Recorder {
+	return &ContextualLoggingEventRecorder{
+		component: component,
+	}
+}
+
+func (r *ContextualLoggingEventRecorder) ComponentName() string {
+	return r.component
+}
+
+func (r *ContextualLoggingEventRecorder) ForComponent(component string) Recorder {
+	newRecorder := *r
+	newRecorder.component = component
+	return &newRecorder
+}
+
+func (r *ContextualLoggingEventRecorder) Shutdown() {}
+
+func (r *ContextualLoggingEventRecorder) WithComponentSuffix(suffix string) Recorder {
+	return r.ForComponent(fmt.Sprintf("%s-%s", r.ComponentName(), suffix))
+}
+
+func (r *ContextualLoggingEventRecorder) Event(ctx context.Context, reason, message string) {
+	logger := klog.FromContext(ctx)
+	logger.Info(fmt.Sprintf("INFO: %s", message), "component", r.component, "reason", reason)
+}
+
+func (r *ContextualLoggingEventRecorder) Eventf(ctx context.Context, reason, messageFmt string, args ...interface{}) {
+	r.Event(ctx, reason, fmt.Sprintf(messageFmt, args...))
+}
+
+func (r *ContextualLoggingEventRecorder) Warning(ctx context.Context, reason, message string) {
+	logger := klog.FromContext(ctx)
+	logger.Info(fmt.Sprintf("WARNING: %s", message), "component", r.component, "reason", reason)
+}
+
+func (r *ContextualLoggingEventRecorder) Warningf(ctx context.Context, reason, messageFmt string, args ...interface{}) {
+	r.Warning(ctx, reason, fmt.Sprintf(messageFmt, args...))
+}

pkg/basecontroller/events/recorder.go

@@ -13,10 +13,10 @@ import (
 
 // Recorder is a simple event recording interface.
 type Recorder interface {
-	Event(reason, message string)
-	Eventf(reason, messageFmt string, args ...interface{})
-	Warning(reason, message string)
-	Warningf(reason, messageFmt string, args ...interface{})
+	Event(ctx context.Context, reason, message string)
+	Eventf(ctx context.Context, reason, messageFmt string, args ...interface{})
+	Warning(ctx context.Context, reason, message string)
+	Warningf(ctx context.Context, reason, messageFmt string, args ...interface{})
 
 	// ForComponent allows to fiddle the component name before sending the event to sink.
 	// Making more unique components will prevent the spam filter in upstream event sink from dropping
@@ -26,9 +26,6 @@ type Recorder interface {
 	// WithComponentSuffix is similar to ForComponent except it just suffix the current component name instead of overriding.
 	WithComponentSuffix(componentNameSuffix string) Recorder
 
-	// WithContext allows to set a context for event create API calls.
-	WithContext(ctx context.Context) Recorder
-
 	// ComponentName returns the current source component name for the event.
 	// This allows to suffix the original component name with 'sub-component'.
 	ComponentName() string
@@ -50,9 +47,6 @@ type recorder struct {
 	eventClient       corev1client.EventInterface
 	involvedObjectRef *corev1.ObjectReference
 	sourceComponent   string
-
-	// TODO: This is not the right way to pass the context, but there is no other way without breaking event interface
-	ctx context.Context
 }
 
 func (r *recorder) ComponentName() string {
@@ -67,44 +61,31 @@ func (r *recorder) ForComponent(componentName string) Recorder {
 	return &newRecorderForComponent
 }
 
-func (r *recorder) WithContext(ctx context.Context) Recorder {
-	r.ctx = ctx
-	return r
-}
-
 func (r *recorder) WithComponentSuffix(suffix string) Recorder {
 	return r.ForComponent(fmt.Sprintf("%s-%s", r.ComponentName(), suffix))
 }
 
 // Eventf emits the normal type event and allow formatting of message.
-func (r *recorder) Eventf(reason, messageFmt string, args ...interface{}) {
-	r.Event(reason, fmt.Sprintf(messageFmt, args...))
+func (r *recorder) Eventf(ctx context.Context, reason, messageFmt string, args ...interface{}) {
+	r.Event(ctx, reason, fmt.Sprintf(messageFmt, args...))
 }
 
 // Warningf emits the warning type event and allow formatting of message.
-func (r *recorder) Warningf(reason, messageFmt string, args ...interface{}) {
-	r.Warning(reason, fmt.Sprintf(messageFmt, args...))
+func (r *recorder) Warningf(ctx context.Context, reason, messageFmt string, args ...interface{}) {
+	r.Warning(ctx, reason, fmt.Sprintf(messageFmt, args...))
 }
 
 // Event emits the normal type event.
-func (r *recorder) Event(reason, message string) {
+func (r *recorder) Event(ctx context.Context, reason, message string) {
 	event := makeEvent(r.involvedObjectRef, r.sourceComponent, corev1.EventTypeNormal, reason, message)
-	ctx := context.Background()
-	if r.ctx != nil {
-		ctx = r.ctx
-	}
 	if _, err := r.eventClient.Create(ctx, event, metav1.CreateOptions{}); err != nil {
 		klog.Warningf("Error creating event %+v: %v", event, err)
 	}
 }
 
 // Warning emits the warning type event.
-func (r *recorder) Warning(reason, message string) {
+func (r *recorder) Warning(ctx context.Context, reason, message string) {
 	event := makeEvent(r.involvedObjectRef, r.sourceComponent, corev1.EventTypeWarning, reason, message)
-	ctx := context.Background()
-	if r.ctx != nil {
-		ctx = r.ctx
-	}
 	if _, err := r.eventClient.Create(ctx, event, metav1.CreateOptions{}); err != nil {
 		klog.Warningf("Error creating event %+v: %v", event, err)
 	}

pkg/basecontroller/factory/controller_context.go

@@ -2,17 +2,18 @@ package factory
 
 import (
 	"fmt"
-
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/events"
 )
 
 // syncContext implements SyncContext and provide user access to queue and object that caused
 // the sync to be triggered.
 type syncContext struct {
-	queue workqueue.TypedRateLimitingInterface[string]
+	queue    workqueue.TypedRateLimitingInterface[string]
+	recorder events.Recorder
 }
 
 var _ SyncContext = syncContext{}
@@ -24,13 +25,18 @@ func NewSyncContext(name string) SyncContext {
 			workqueue.DefaultTypedControllerRateLimiter[string](),
 			workqueue.TypedRateLimitingQueueConfig[string]{Name: name},
 		),
+		recorder: events.NewContextualLoggingEventRecorder(name),
 	}
 }
 
 func (c syncContext) Queue() workqueue.TypedRateLimitingInterface[string] {
 	return c.queue
 }
 
+func (c syncContext) Recorder() events.Recorder {
+	return c.recorder
+}
+
 // eventHandler provides default event handler that is added to an informers passed to controller factory.
 func (c syncContext) eventHandler(queueKeysFunc ObjectQueueKeysFunc, filter EventFilterFunc) cache.ResourceEventHandler {
 	resourceEventHandler := cache.ResourceEventHandlerFuncs{

pkg/basecontroller/factory/interfaces.go

@@ -2,6 +2,7 @@ package factory
 
 import (
 	"context"
+	"open-cluster-management.io/sdk-go/pkg/basecontroller/events"
 
 	"k8s.io/client-go/util/workqueue"
 )
@@ -34,6 +35,9 @@ type SyncContext interface {
 	// Queue gives access to controller queue. This can be used for manual requeue, although if a Sync() function return
 	// an error, the object is automatically re-queued. Use with caution.
 	Queue() workqueue.TypedRateLimitingInterface[string]
+
+	// Recorder returns a recorder to record events.
+	Recorder() events.Recorder
 }
 
 // SyncFunc is a function that contain main controller logic.