update the event recorder for a passive clock

Author: deads2k

pkg/controller/controllercmd/builder.go

@@ -3,6 +3,7 @@ package controllercmd
 import (
 	"context"
 	"fmt"
+	"k8s.io/utils/clock"
 	"os"
 	"strings"
 	"sync"
@@ -40,6 +41,9 @@ type StartFunc func(context.Context, *ControllerContext) error
 type ControllerContext struct {
 	ComponentConfig *unstructured.Unstructured
 
+	// Clock is a potentially fake clock that must be used to run controllers.
+	Clock clock.Clock
+
 	// KubeConfig provides the REST config with no content type (it will default to JSON).
 	// Use this config for CR resources.
 	KubeConfig *rest.Config
@@ -71,6 +75,7 @@ type ControllerBuilder struct {
 	fileObserverReactorFn   func(file string, action fileobserver.ActionType) error
 	eventRecorderOptions    record.CorrelatorOptions
 	componentOwnerReference *corev1.ObjectReference
+	clock                   clock.Clock
 
 	startFunc          StartFunc
 	componentName      string
@@ -121,10 +126,11 @@ func (i infrastructureStatusTopologyDetector) DetectTopology(ctx context.Context
 var _ TopologyDetector = (*infrastructureStatusTopologyDetector)(nil)
 
 // NewController returns a builder struct for constructing the command you want to run
-func NewController(componentName string, startFunc StartFunc) *ControllerBuilder {
+func NewController(componentName string, startFunc StartFunc, clock clock.Clock) *ControllerBuilder {
 	return &ControllerBuilder{
 		startFunc:        startFunc,
 		componentName:    componentName,
+		clock:            clock,
 		observerInterval: defaultObserverInterval,
 		nonZeroExitFn: func(args ...interface{}) {
 			klog.Warning(args...)
@@ -266,7 +272,7 @@ func (b *ControllerBuilder) Run(ctx context.Context, config *unstructured.Unstru
 			klog.Warningf("unable to get owner reference (falling back to namespace): %v", err)
 		}
 	}
-	eventRecorder := events.NewKubeRecorderWithOptions(kubeClient.CoreV1().Events(namespace), b.eventRecorderOptions, b.componentName, controllerRef)
+	eventRecorder := events.NewKubeRecorderWithOptions(kubeClient.CoreV1().Events(namespace), b.eventRecorderOptions, b.componentName, controllerRef, b.clock)
 
 	utilruntime.PanicHandlers = append(utilruntime.PanicHandlers, func(c context.Context, r interface{}) {
 		eventRecorder.Warningf(fmt.Sprintf("%sPanic", strings.Title(b.componentName)), "Panic observed: %v", r)
@@ -336,6 +342,7 @@ func (b *ControllerBuilder) Run(ctx context.Context, config *unstructured.Unstru
 
 	controllerContext := &ControllerContext{
 		ComponentConfig:   config,
+		Clock:             b.clock,
 		KubeConfig:        clientConfig,
 		ProtoKubeConfig:   protoConfig,
 		EventRecorder:     eventRecorder,

pkg/controller/controllercmd/cmd.go

@@ -3,6 +3,7 @@ package controllercmd
 import (
 	"context"
 	"fmt"
+	"k8s.io/utils/clock"
 	"math/rand"
 	"os"
 	"path/filepath"
@@ -41,6 +42,7 @@ type ControllerCommandConfig struct {
 	componentName string
 	startFunc     StartFunc
 	version       version.Info
+	clock         clock.Clock
 
 	basicFlags *ControllerFlags
 
@@ -76,11 +78,12 @@ type ControllerCommandConfig struct {
 
 // NewControllerConfig returns a new ControllerCommandConfig which can be used to wire up all the boiler plate of a controller
 // TODO add more methods around wiring health checks and the like
-func NewControllerCommandConfig(componentName string, version version.Info, startFunc StartFunc) *ControllerCommandConfig {
+func NewControllerCommandConfig(componentName string, version version.Info, startFunc StartFunc, clock clock.Clock) *ControllerCommandConfig {
 	return &ControllerCommandConfig{
 		startFunc:     startFunc,
 		componentName: componentName,
 		version:       version,
+		clock:         clock,
 
 		basicFlags: NewControllerFlags(),
 
@@ -322,7 +325,7 @@ func (c *ControllerCommandConfig) StartController(ctx context.Context) error {
 	config.LeaderElection.RenewDeadline = c.RenewDeadline
 	config.LeaderElection.RetryPeriod = c.RetryPeriod
 
-	builder := NewController(c.componentName, c.startFunc).
+	builder := NewController(c.componentName, c.startFunc, c.clock).
 		WithKubeConfigFile(c.basicFlags.KubeConfigFile, nil).
 		WithComponentNamespace(c.basicFlags.Namespace).
 		WithLeaderElection(config.LeaderElection, c.basicFlags.Namespace, c.componentName+"-lock").

pkg/operator/events/recorder.go

@@ -4,11 +4,10 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"os"
-	"time"
-
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/clock"
+	"os"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -145,11 +144,12 @@ func guessControllerReferenceForNamespace(ctx context.Context, client corev1clie
 }
 
 // NewRecorder returns new event recorder.
-func NewRecorder(client corev1client.EventInterface, sourceComponentName string, involvedObjectRef *corev1.ObjectReference) Recorder {
+func NewRecorder(client corev1client.EventInterface, sourceComponentName string, involvedObjectRef *corev1.ObjectReference, clock clock.PassiveClock) Recorder {
 	return &recorder{
 		eventClient:       client,
 		involvedObjectRef: involvedObjectRef,
 		sourceComponent:   sourceComponentName,
+		clock:             clock,
 	}
 }
 
@@ -158,6 +158,7 @@ type recorder struct {
 	eventClient       corev1client.EventInterface
 	involvedObjectRef *corev1.ObjectReference
 	sourceComponent   string
+	clock             clock.PassiveClock
 
 	// TODO: This is not the right way to pass the context, but there is no other way without breaking event interface
 	ctx context.Context
@@ -196,7 +197,7 @@ func (r *recorder) Warningf(reason, messageFmt string, args ...interface{}) {
 
 // Event emits the normal type event.
 func (r *recorder) Event(reason, message string) {
-	event := makeEvent(r.involvedObjectRef, r.sourceComponent, corev1.EventTypeNormal, reason, message)
+	event := makeEvent(r.clock, r.involvedObjectRef, r.sourceComponent, corev1.EventTypeNormal, reason, message)
 	ctx := context.Background()
 	if r.ctx != nil {
 		ctx = r.ctx
@@ -208,7 +209,7 @@ func (r *recorder) Event(reason, message string) {
 
 // Warning emits the warning type event.
 func (r *recorder) Warning(reason, message string) {
-	event := makeEvent(r.involvedObjectRef, r.sourceComponent, corev1.EventTypeWarning, reason, message)
+	event := makeEvent(r.clock, r.involvedObjectRef, r.sourceComponent, corev1.EventTypeWarning, reason, message)
 	ctx := context.Background()
 	if r.ctx != nil {
 		ctx = r.ctx
@@ -218,10 +219,11 @@ func (r *recorder) Warning(reason, message string) {
 	}
 }
 
-func makeEvent(involvedObjRef *corev1.ObjectReference, sourceComponent string, eventType, reason, message string) *corev1.Event {
-	currentTime := metav1.Time{Time: time.Now()}
+func makeEvent(clock clock.PassiveClock, involvedObjRef *corev1.ObjectReference, sourceComponent string, eventType, reason, message string) *corev1.Event {
+	currentTime := metav1.Time{Time: clock.Now()}
 	event := &corev1.Event{
 		ObjectMeta: metav1.ObjectMeta{
+			// TODO this is always used to create a unique event.  Perhaps we should hash the message to be unique enough for apply-configuration
 			Name:      fmt.Sprintf("%v.%x", involvedObjRef.Name, currentTime.UnixNano()),
 			Namespace: involvedObjRef.Namespace,
 		},

pkg/operator/events/recorder_in_memory.go

@@ -3,6 +3,7 @@ package events
 import (
 	"context"
 	"fmt"
+	"k8s.io/utils/clock"
 	"sync"
 
 	corev1 "k8s.io/api/core/v1"
@@ -12,6 +13,7 @@ import (
 type inMemoryEventRecorder struct {
 	events []*corev1.Event
 	source string
+	clock  clock.PassiveClock
 	ctx    context.Context
 	sync.Mutex
 }
@@ -31,8 +33,12 @@ type InMemoryRecorder interface {
 
 // NewInMemoryRecorder provides event recorder that stores all events recorded in memory and allow to replay them using the Events() method.
 // This recorder should be only used in unit tests.
-func NewInMemoryRecorder(sourceComponent string) InMemoryRecorder {
-	return &inMemoryEventRecorder{events: []*corev1.Event{}, source: sourceComponent}
+func NewInMemoryRecorder(sourceComponent string, clock clock.PassiveClock) InMemoryRecorder {
+	return &inMemoryEventRecorder{
+		events: []*corev1.Event{},
+		source: sourceComponent,
+		clock:  clock,
+	}
 }
 
 func (r *inMemoryEventRecorder) ComponentName() string {
@@ -65,7 +71,7 @@ func (r *inMemoryEventRecorder) Events() []*corev1.Event {
 func (r *inMemoryEventRecorder) Event(reason, message string) {
 	r.Lock()
 	defer r.Unlock()
-	event := makeEvent(&inMemoryDummyObjectReference, r.source, corev1.EventTypeNormal, reason, message)
+	event := makeEvent(r.clock, &inMemoryDummyObjectReference, r.source, corev1.EventTypeNormal, reason, message)
 	r.events = append(r.events, event)
 }
 
@@ -76,7 +82,7 @@ func (r *inMemoryEventRecorder) Eventf(reason, messageFmt string, args ...interf
 func (r *inMemoryEventRecorder) Warning(reason, message string) {
 	r.Lock()
 	defer r.Unlock()
-	event := makeEvent(&inMemoryDummyObjectReference, r.source, corev1.EventTypeWarning, reason, message)
+	event := makeEvent(r.clock, &inMemoryDummyObjectReference, r.source, corev1.EventTypeWarning, reason, message)
 	klog.Info(event.String())
 	r.events = append(r.events, event)
 }

pkg/operator/events/recorder_logging.go

@@ -3,13 +3,15 @@ package events
 import (
 	"context"
 	"fmt"
+	"k8s.io/utils/clock"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/klog/v2"
 )
 
 type LoggingEventRecorder struct {
 	component string
+	clock     clock.PassiveClock
 	ctx       context.Context
 }
 
@@ -19,8 +21,11 @@ func (r *LoggingEventRecorder) WithContext(ctx context.Context) Recorder {
 }
 
 // NewLoggingEventRecorder provides event recorder that will log all recorded events via klog.
-func NewLoggingEventRecorder(component string) Recorder {
-	return &LoggingEventRecorder{component: component}
+func NewLoggingEventRecorder(component string, clock clock.PassiveClock) Recorder {
+	return &LoggingEventRecorder{
+		component: component,
+		clock:     clock,
+	}
 }
 
 func (r *LoggingEventRecorder) ComponentName() string {
@@ -40,7 +45,7 @@ func (r *LoggingEventRecorder) WithComponentSuffix(suffix string) Recorder {
 }
 
 func (r *LoggingEventRecorder) Event(reason, message string) {
-	event := makeEvent(&inMemoryDummyObjectReference, "", corev1.EventTypeNormal, reason, message)
+	event := makeEvent(r.clock, &inMemoryDummyObjectReference, "", corev1.EventTypeNormal, reason, message)
 	klog.Info(event.String())
 }
 
@@ -49,7 +54,7 @@ func (r *LoggingEventRecorder) Eventf(reason, messageFmt string, args ...interfa
 }
 
 func (r *LoggingEventRecorder) Warning(reason, message string) {
-	event := makeEvent(&inMemoryDummyObjectReference, "", corev1.EventTypeWarning, reason, message)
+	event := makeEvent(r.clock, &inMemoryDummyObjectReference, "", corev1.EventTypeWarning, reason, message)
 	klog.Warning(event.String())
 }
 

pkg/operator/events/recorder_upstream.go

@@ -3,6 +3,7 @@ package events
 import (
 	"context"
 	"fmt"
+	"k8s.io/utils/clock"
 	"strings"
 	"sync"
 
@@ -16,19 +17,19 @@ import (
 )
 
 // NewKubeRecorder returns new event recorder with tweaked correlator options.
-func NewKubeRecorderWithOptions(client corev1client.EventInterface, options record.CorrelatorOptions, sourceComponentName string, involvedObjectRef *corev1.ObjectReference) Recorder {
+func NewKubeRecorderWithOptions(client corev1client.EventInterface, options record.CorrelatorOptions, sourceComponentName string, involvedObjectRef *corev1.ObjectReference, clock clock.PassiveClock) Recorder {
 	return (&upstreamRecorder{
 		client:            client,
 		component:         sourceComponentName,
 		involvedObjectRef: involvedObjectRef,
 		options:           options,
-		fallbackRecorder:  NewRecorder(client, sourceComponentName, involvedObjectRef),
+		fallbackRecorder:  NewRecorder(client, sourceComponentName, involvedObjectRef, clock),
 	}).ForComponent(sourceComponentName)
 }
 
 // NewKubeRecorder returns new event recorder with default correlator options.
-func NewKubeRecorder(client corev1client.EventInterface, sourceComponentName string, involvedObjectRef *corev1.ObjectReference) Recorder {
-	return NewKubeRecorderWithOptions(client, record.CorrelatorOptions{}, sourceComponentName, involvedObjectRef)
+func NewKubeRecorder(client corev1client.EventInterface, sourceComponentName string, involvedObjectRef *corev1.ObjectReference, clock clock.PassiveClock) Recorder {
+	return NewKubeRecorderWithOptions(client, record.CorrelatorOptions{}, sourceComponentName, involvedObjectRef, clock)
 }
 
 // upstreamRecorder is an implementation of Recorder interface.

pkg/operator/staticpod/certsyncpod/certsync_cmd.go

@@ -2,6 +2,7 @@ package certsyncpod
 
 import (
 	"context"
+	"k8s.io/utils/clock"
 	"os"
 	"time"
 
@@ -23,6 +24,7 @@ type CertSyncControllerOptions struct {
 	KubeConfigFile string
 	Namespace      string
 	DestinationDir string
+	Clock          clock.PassiveClock
 
 	configMaps []installer.UnrevisionedResource
 	secrets    []installer.UnrevisionedResource
@@ -33,6 +35,7 @@ type CertSyncControllerOptions struct {
 
 func NewCertSyncControllerCommand(configmaps, secrets []installer.UnrevisionedResource) *cobra.Command {
 	o := &CertSyncControllerOptions{
+		Clock:      clock.RealClock{},
 		configMaps: configmaps,
 		secrets:    secrets,
 	}
@@ -87,7 +90,7 @@ func (o *CertSyncControllerOptions) Run() error {
 			Kind:       "Pod",
 			Namespace:  os.Getenv("POD_NAMESPACE"),
 			Name:       os.Getenv("POD_NAME"),
-		})
+		}, o.Clock)
 
 	controller := NewCertSyncController(
 		o.DestinationDir,

pkg/operator/staticpod/controllers.go

@@ -2,6 +2,7 @@ package staticpod
 
 import (
 	"fmt"
+	"k8s.io/utils/clock"
 	"time"
 
 	operatorv1 "github.com/openshift/api/operator/v1"
@@ -39,6 +40,7 @@ type staticPodOperatorControllerBuilder struct {
 	kubeClient              kubernetes.Interface
 	kubeInformers           v1helpers.KubeInformersForNamespaces
 	configInformers         externalversions.SharedInformerFactory
+	clock                   clock.Clock
 	eventRecorder           events.Recorder
 
 	// resource information
@@ -84,12 +86,14 @@ func NewBuilder(
 	kubeClient kubernetes.Interface,
 	kubeInformers v1helpers.KubeInformersForNamespaces,
 	configInformers externalversions.SharedInformerFactory,
+	clock clock.Clock,
 ) Builder {
 	return &staticPodOperatorControllerBuilder{
 		staticPodOperatorClient: staticPodOperatorClient,
 		kubeClient:              kubeClient,
 		kubeInformers:           kubeInformers,
 		configInformers:         configInformers,
+		clock:                   clock,
 	}
 }
 
@@ -209,7 +213,7 @@ func (b *staticPodOperatorControllerBuilder) ToControllers() (manager.Controller
 
 	eventRecorder := b.eventRecorder
 	if eventRecorder == nil {
-		eventRecorder = events.NewLoggingEventRecorder("static-pod-operator-controller")
+		eventRecorder = events.NewLoggingEventRecorder("static-pod-operator-controller", b.clock)
 	}
 	versionRecorder := b.versionRecorder
 	if versionRecorder == nil {