github/client-go
1
0
Fork 0
mirror of https://github.com/kubernetes/client-go.git synced 2025-09-14 06:13:19 +00:00
Code Issues Projects Releases Wiki Activity

Switch event recorder to use events.k8s.io/v1

Browse Source 
Kubernetes-commit: ade242288346d968ab97f42ffb30e0855e235afc
This commit is contained in:
Chelsey Chen
2020-06-16 18:25:18 -04:00
committed by Kubernetes Publisher
parent de1f58acbf
commit 67fdd63515
4 changed files with 100 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
tools/events/event_broadcaster.go
View File

@@ -17,27 +17,29 @@ limitations under the License.
package events package events
import ( import (
"context"
"fmt"
"os" "os"
"sync" "sync"
"time" "time"
corev1 "k8s.io/api/core/v1" corev1 "k8s.io/api/core/v1"
eventsv1 "k8s.io/api/events/v1"
"k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/clock" "k8s.io/apimachinery/pkg/util/clock"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/watch"
restclient "k8s.io/client-go/rest"
"k8s.io/api/events/v1beta1"
"k8s.io/apimachinery/pkg/util/json" "k8s.io/apimachinery/pkg/util/json"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/strategicpatch" "k8s.io/apimachinery/pkg/util/strategicpatch"
"k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/util/wait"
"k8s.io/apimachinery/pkg/watch"
clientset "k8s.io/client-go/kubernetes" clientset "k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/kubernetes/scheme"
typedv1core "k8s.io/client-go/kubernetes/typed/core/v1" typedv1core "k8s.io/client-go/kubernetes/typed/core/v1"
typedv1beta1 "k8s.io/client-go/kubernetes/typed/events/v1beta1" typedeventsv1 "k8s.io/client-go/kubernetes/typed/events/v1"
restclient "k8s.io/client-go/rest"
"k8s.io/client-go/tools/record" "k8s.io/client-go/tools/record"
"k8s.io/client-go/tools/record/util" "k8s.io/client-go/tools/record/util"
"k8s.io/klog/v2" "k8s.io/klog/v2"
@@ -64,40 +66,49 @@ type eventKey struct {
type eventBroadcasterImpl struct { type eventBroadcasterImpl struct {
*watch.Broadcaster *watch.Broadcaster
mu sync.Mutex mu sync.Mutex
eventCache map[eventKey]*v1beta1.Event eventCache map[eventKey]*eventsv1.Event
sleepDuration time.Duration sleepDuration time.Duration
sink EventSink sink EventSink
} }
// EventSinkImpl wraps EventInterface to implement EventSink. // EventSinkImpl wraps EventsV1Interface to implement EventSink.
// TODO: this makes it easier for testing purpose and masks the logic of performing API calls. // TODO: this makes it easier for testing purpose and masks the logic of performing API calls.
// Note that rollbacking to raw clientset should also be transparent. // Note that rollbacking to raw clientset should also be transparent.
type EventSinkImpl struct { type EventSinkImpl struct {
Interface typedv1beta1.EventInterface Interface typedeventsv1.EventsV1Interface
} }
// Create is the same as CreateWithEventNamespace of the EventExpansion // Create takes the representation of a event and creates it. Returns the server's representation of the event, and an error, if there is any.
func (e *EventSinkImpl) Create(event *v1beta1.Event) (*v1beta1.Event, error) { func (e *EventSinkImpl) Create(event *eventsv1.Event) (*eventsv1.Event, error) {
return e.Interface.CreateWithEventNamespace(event) if event.Namespace == "" {
return nil, fmt.Errorf("can't create an event with empty namespace")
}
return e.Interface.Events(event.Namespace).Create(context.TODO(), event, metav1.CreateOptions{})
} }
// Update is the same as UpdateithEventNamespace of the EventExpansion // Update takes the representation of a event and updates it. Returns the server's representation of the event, and an error, if there is any.
func (e *EventSinkImpl) Update(event *v1beta1.Event) (*v1beta1.Event, error) { func (e *EventSinkImpl) Update(event *eventsv1.Event) (*eventsv1.Event, error) {
return e.Interface.UpdateWithEventNamespace(event) if event.Namespace == "" {
return nil, fmt.Errorf("can't update an event with empty namespace")
}
return e.Interface.Events(event.Namespace).Update(context.TODO(), event, metav1.UpdateOptions{})
} }
// Patch is the same as PatchWithEventNamespace of the EventExpansion // Patch applies the patch and returns the patched event, and an error, if there is any.
func (e *EventSinkImpl) Patch(event *v1beta1.Event, data []byte) (*v1beta1.Event, error) { func (e *EventSinkImpl) Patch(event *eventsv1.Event, data []byte) (*eventsv1.Event, error) {
return e.Interface.PatchWithEventNamespace(event, data) if event.Namespace == "" {
return nil, fmt.Errorf("can't patch an event with empty namespace")
}
return e.Interface.Events(event.Namespace).Patch(context.TODO(), event.Name, types.StrategicMergePatchType, data, metav1.PatchOptions{})
} }
// NewBroadcaster Creates a new event broadcaster. // NewBroadcaster Creates a new event broadcaster.
func NewBroadcaster(sink EventSink) EventBroadcaster { func NewBroadcaster(sink EventSink) EventBroadcaster {
return newBroadcaster(sink, defaultSleepDuration, map[eventKey]*v1beta1.Event{}) return newBroadcaster(sink, defaultSleepDuration, map[eventKey]*eventsv1.Event{})
} }
// NewBroadcasterForTest Creates a new event broadcaster for test purposes. // NewBroadcasterForTest Creates a new event broadcaster for test purposes.
func newBroadcaster(sink EventSink, sleepDuration time.Duration, eventCache map[eventKey]*v1beta1.Event) EventBroadcaster { func newBroadcaster(sink EventSink, sleepDuration time.Duration, eventCache map[eventKey]*eventsv1.Event) EventBroadcaster {
return &eventBroadcasterImpl{ return &eventBroadcasterImpl{
Broadcaster: watch.NewBroadcaster(maxQueuedEvents, watch.DropIfChannelFull), Broadcaster: watch.NewBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
eventCache: eventCache, eventCache: eventCache,
@@ -154,11 +165,11 @@ func (e *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, reportingCont
return &recorderImpl{scheme, reportingController, reportingInstance, e.Broadcaster, clock.RealClock{}} return &recorderImpl{scheme, reportingController, reportingInstance, e.Broadcaster, clock.RealClock{}}
} }
func (e *eventBroadcasterImpl) recordToSink(event *v1beta1.Event, clock clock.Clock) { func (e *eventBroadcasterImpl) recordToSink(event *eventsv1.Event, clock clock.Clock) {
// Make a copy before modification, because there could be multiple listeners. // Make a copy before modification, because there could be multiple listeners.
eventCopy := event.DeepCopy() eventCopy := event.DeepCopy()
go func() { go func() {
evToRecord := func() *v1beta1.Event { evToRecord := func() *eventsv1.Event {
e.mu.Lock() e.mu.Lock()
defer e.mu.Unlock() defer e.mu.Unlock()
eventKey := getKey(eventCopy) eventKey := getKey(eventCopy)
@@ -169,7 +180,7 @@ func (e *eventBroadcasterImpl) recordToSink(event *v1beta1.Event, clock clock.Cl
isomorphicEvent.Series.LastObservedTime = metav1.MicroTime{Time: clock.Now()} isomorphicEvent.Series.LastObservedTime = metav1.MicroTime{Time: clock.Now()}
return nil return nil
} }
isomorphicEvent.Series = &v1beta1.EventSeries{ isomorphicEvent.Series = &eventsv1.EventSeries{
Count: 1, Count: 1,
LastObservedTime: metav1.MicroTime{Time: clock.Now()}, LastObservedTime: metav1.MicroTime{Time: clock.Now()},
} }
@@ -190,7 +201,7 @@ func (e *eventBroadcasterImpl) recordToSink(event *v1beta1.Event, clock clock.Cl
}() }()
} }
func (e *eventBroadcasterImpl) attemptRecording(event *v1beta1.Event) *v1beta1.Event { func (e *eventBroadcasterImpl) attemptRecording(event *eventsv1.Event) *eventsv1.Event {
tries := 0 tries := 0
for { for {
if recordedEvent, retry := recordEvent(e.sink, event); !retry { if recordedEvent, retry := recordEvent(e.sink, event); !retry {
@@ -207,8 +218,8 @@ func (e *eventBroadcasterImpl) attemptRecording(event *v1beta1.Event) *v1beta1.E
} }
} }
func recordEvent(sink EventSink, event *v1beta1.Event) (*v1beta1.Event, bool) { func recordEvent(sink EventSink, event *eventsv1.Event) (*eventsv1.Event, bool) {
var newEvent *v1beta1.Event var newEvent *eventsv1.Event
var err error var err error
isEventSeries := event.Series != nil isEventSeries := event.Series != nil
if isEventSeries { if isEventSeries {
@@ -252,7 +263,7 @@ func recordEvent(sink EventSink, event *v1beta1.Event) (*v1beta1.Event, bool) {
return nil, true return nil, true
} }
func createPatchBytesForSeries(event *v1beta1.Event) ([]byte, error) { func createPatchBytesForSeries(event *eventsv1.Event) ([]byte, error) {
oldEvent := event.DeepCopy() oldEvent := event.DeepCopy()
oldEvent.Series = nil oldEvent.Series = nil
oldData, err := json.Marshal(oldEvent) oldData, err := json.Marshal(oldEvent)
@@ -263,10 +274,10 @@ func createPatchBytesForSeries(event *v1beta1.Event) ([]byte, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
return strategicpatch.CreateTwoWayMergePatch(oldData, newData, v1beta1.Event{}) return strategicpatch.CreateTwoWayMergePatch(oldData, newData, eventsv1.Event{})
} }
func getKey(event *v1beta1.Event) eventKey { func getKey(event *eventsv1.Event) eventKey {
key := eventKey{ key := eventKey{
action: event.Action, action: event.Action,
reason: event.Reason, reason: event.Reason,
@@ -305,9 +316,9 @@ func (e *eventBroadcasterImpl) StartRecordingToSink(stopCh <-chan struct{}) {
e.finishSeries() e.finishSeries()
}, finishTime, stopCh) }, finishTime, stopCh)
eventHandler := func(obj runtime.Object) { eventHandler := func(obj runtime.Object) {
event, ok := obj.(*v1beta1.Event) event, ok := obj.(*eventsv1.Event)
if !ok { if !ok {
klog.Errorf("unexpected type, expected v1beta1.Event") klog.Errorf("unexpected type, expected eventsv1.Event")
return return
} }
e.recordToSink(event, clock.RealClock{}) e.recordToSink(event, clock.RealClock{})
@@ -320,19 +331,19 @@ func (e *eventBroadcasterImpl) StartRecordingToSink(stopCh <-chan struct{}) {
} }
type eventBroadcasterAdapterImpl struct { type eventBroadcasterAdapterImpl struct {
coreClient typedv1core.EventsGetter coreClient typedv1core.EventsGetter
coreBroadcaster record.EventBroadcaster coreBroadcaster record.EventBroadcaster
v1beta1Client typedv1beta1.EventsGetter eventsv1Client typedeventsv1.EventsV1Interface
v1beta1Broadcaster EventBroadcaster eventsv1Broadcaster EventBroadcaster
} }
// NewEventBroadcasterAdapter creates a wrapper around new and legacy broadcasters to simplify // NewEventBroadcasterAdapter creates a wrapper around new and legacy broadcasters to simplify
// migration of individual components to the new Event API. // migration of individual components to the new Event API.
func NewEventBroadcasterAdapter(client clientset.Interface) EventBroadcasterAdapter { func NewEventBroadcasterAdapter(client clientset.Interface) EventBroadcasterAdapter {
eventClient := &eventBroadcasterAdapterImpl{} eventClient := &eventBroadcasterAdapterImpl{}
if _, err := client.Discovery().ServerResourcesForGroupVersion(v1beta1.SchemeGroupVersion.String()); err == nil { if _, err := client.Discovery().ServerResourcesForGroupVersion(eventsv1.SchemeGroupVersion.String()); err == nil {
eventClient.v1beta1Client = client.EventsV1beta1() eventClient.eventsv1Client = client.EventsV1()
eventClient.v1beta1Broadcaster = NewBroadcaster(&EventSinkImpl{Interface: eventClient.v1beta1Client.Events("")}) eventClient.eventsv1Broadcaster = NewBroadcaster(&EventSinkImpl{Interface: eventClient.eventsv1Client})
} }
// Even though there can soon exist cases when coreBroadcaster won't really be needed, // Even though there can soon exist cases when coreBroadcaster won't really be needed,
// we create it unconditionally because its overhead is minor and will simplify using usage // we create it unconditionally because its overhead is minor and will simplify using usage
@@ -344,8 +355,8 @@ func NewEventBroadcasterAdapter(client clientset.Interface) EventBroadcasterAdap
// StartRecordingToSink starts sending events received from the specified eventBroadcaster to the given sink. // StartRecordingToSink starts sending events received from the specified eventBroadcaster to the given sink.
func (e *eventBroadcasterAdapterImpl) StartRecordingToSink(stopCh <-chan struct{}) { func (e *eventBroadcasterAdapterImpl) StartRecordingToSink(stopCh <-chan struct{}) {
if e.v1beta1Broadcaster != nil && e.v1beta1Client != nil { if e.eventsv1Broadcaster != nil && e.eventsv1Client != nil {
e.v1beta1Broadcaster.StartRecordingToSink(stopCh) e.eventsv1Broadcaster.StartRecordingToSink(stopCh)
} }
if e.coreBroadcaster != nil && e.coreClient != nil { if e.coreBroadcaster != nil && e.coreClient != nil {
e.coreBroadcaster.StartRecordingToSink(&typedv1core.EventSinkImpl{Interface: e.coreClient.Events("")}) e.coreBroadcaster.StartRecordingToSink(&typedv1core.EventSinkImpl{Interface: e.coreClient.Events("")})
@@ -353,8 +364,8 @@ func (e *eventBroadcasterAdapterImpl) StartRecordingToSink(stopCh <-chan struct{
} }
func (e *eventBroadcasterAdapterImpl) NewRecorder(name string) EventRecorder { func (e *eventBroadcasterAdapterImpl) NewRecorder(name string) EventRecorder {
if e.v1beta1Broadcaster != nil && e.v1beta1Client != nil { if e.eventsv1Broadcaster != nil && e.eventsv1Client != nil {
return e.v1beta1Broadcaster.NewRecorder(scheme.Scheme, name) return e.eventsv1Broadcaster.NewRecorder(scheme.Scheme, name)
} }
return record.NewEventRecorderAdapter(e.DeprecatedNewLegacyRecorder(name)) return record.NewEventRecorderAdapter(e.DeprecatedNewLegacyRecorder(name))
} }
@@ -367,7 +378,7 @@ func (e *eventBroadcasterAdapterImpl) Shutdown() {
if e.coreBroadcaster != nil { if e.coreBroadcaster != nil {
e.coreBroadcaster.Shutdown() e.coreBroadcaster.Shutdown()
} }
if e.v1beta1Broadcaster != nil { if e.eventsv1Broadcaster != nil {
e.v1beta1Broadcaster.Shutdown() e.eventsv1Broadcaster.Shutdown()
} }
} }

9
tools/events/event_recorder.go
View File

@@ -21,15 +21,14 @@ import (
"time" "time"
v1 "k8s.io/api/core/v1" v1 "k8s.io/api/core/v1"
eventsv1 "k8s.io/api/events/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/util/clock" "k8s.io/apimachinery/pkg/util/clock"
utilruntime "k8s.io/apimachinery/pkg/util/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/watch" "k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/tools/reference"
"k8s.io/api/events/v1beta1"
"k8s.io/client-go/tools/record/util" "k8s.io/client-go/tools/record/util"
"k8s.io/client-go/tools/reference"
"k8s.io/klog/v2" "k8s.io/klog/v2"
) )
@@ -64,13 +63,13 @@ func (recorder *recorderImpl) Eventf(regarding runtime.Object, related runtime.O
}() }()
} }
func (recorder *recorderImpl) makeEvent(refRegarding *v1.ObjectReference, refRelated *v1.ObjectReference, timestamp metav1.MicroTime, eventtype, reason, message string, reportingController string, reportingInstance string, action string) *v1beta1.Event { func (recorder *recorderImpl) makeEvent(refRegarding *v1.ObjectReference, refRelated *v1.ObjectReference, timestamp metav1.MicroTime, eventtype, reason, message string, reportingController string, reportingInstance string, action string) *eventsv1.Event {
t := metav1.Time{Time: recorder.clock.Now()} t := metav1.Time{Time: recorder.clock.Now()}
namespace := refRegarding.Namespace namespace := refRegarding.Namespace
if namespace == "" { if namespace == "" {
namespace = metav1.NamespaceDefault namespace = metav1.NamespaceDefault
} }
return &v1beta1.Event{ return &eventsv1.Event{
ObjectMeta: metav1.ObjectMeta{ ObjectMeta: metav1.ObjectMeta{
Name: fmt.Sprintf("%v.%x", refRegarding.Name, t.UnixNano()), Name: fmt.Sprintf("%v.%x", refRegarding.Name, t.UnixNano()),
Namespace: namespace, Namespace: namespace,

74
tools/events/eventseries_test.go
View File

@@ -25,7 +25,7 @@ import (
"strings" "strings"
v1 "k8s.io/api/core/v1" v1 "k8s.io/api/core/v1"
"k8s.io/api/events/v1beta1" eventsv1 "k8s.io/api/events/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
k8sruntime "k8s.io/apimachinery/pkg/runtime" k8sruntime "k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/util/wait"
@@ -35,13 +35,13 @@ import (
) )
type testEventSeriesSink struct { type testEventSeriesSink struct {
OnCreate func(e *v1beta1.Event) (*v1beta1.Event, error) OnCreate func(e *eventsv1.Event) (*eventsv1.Event, error)
OnUpdate func(e *v1beta1.Event) (*v1beta1.Event, error) OnUpdate func(e *eventsv1.Event) (*eventsv1.Event, error)
OnPatch func(e *v1beta1.Event, p []byte) (*v1beta1.Event, error) OnPatch func(e *eventsv1.Event, p []byte) (*eventsv1.Event, error)
} }
// Create records the event for testing. // Create records the event for testing.
func (t *testEventSeriesSink) Create(e *v1beta1.Event) (*v1beta1.Event, error) { func (t *testEventSeriesSink) Create(e *eventsv1.Event) (*eventsv1.Event, error) {
if t.OnCreate != nil { if t.OnCreate != nil {
return t.OnCreate(e) return t.OnCreate(e)
} }
@@ -49,7 +49,7 @@ func (t *testEventSeriesSink) Create(e *v1beta1.Event) (*v1beta1.Event, error) {
} }
// Update records the event for testing. // Update records the event for testing.
func (t *testEventSeriesSink) Update(e *v1beta1.Event) (*v1beta1.Event, error) { func (t *testEventSeriesSink) Update(e *eventsv1.Event) (*eventsv1.Event, error) {
if t.OnUpdate != nil { if t.OnUpdate != nil {
return t.OnUpdate(e) return t.OnUpdate(e)
} }
@@ -57,7 +57,7 @@ func (t *testEventSeriesSink) Update(e *v1beta1.Event) (*v1beta1.Event, error) {
} }
// Patch records the event for testing. // Patch records the event for testing.
func (t *testEventSeriesSink) Patch(e *v1beta1.Event, p []byte) (*v1beta1.Event, error) { func (t *testEventSeriesSink) Patch(e *eventsv1.Event, p []byte) (*eventsv1.Event, error) {
if t.OnPatch != nil { if t.OnPatch != nil {
return t.OnPatch(e, p) return t.OnPatch(e, p)
} }
@@ -86,7 +86,7 @@ func TestEventSeriesf(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
expectedEvent := &v1beta1.Event{ expectedEvent := &eventsv1.Event{
ObjectMeta: metav1.ObjectMeta{ ObjectMeta: metav1.ObjectMeta{
Name: "foo", Name: "foo",
Namespace: "baz", Namespace: "baz",
@@ -107,13 +107,13 @@ func TestEventSeriesf(t *testing.T) {
nonIsomorphicEvent := expectedEvent.DeepCopy() nonIsomorphicEvent := expectedEvent.DeepCopy()
nonIsomorphicEvent.Action = "stopped" nonIsomorphicEvent.Action = "stopped"
expectedEvent.Series = &v1beta1.EventSeries{Count: 1} expectedEvent.Series = &eventsv1.EventSeries{Count: 1}
table := []struct { table := []struct {
regarding k8sruntime.Object regarding k8sruntime.Object
related k8sruntime.Object related k8sruntime.Object
actual *v1beta1.Event actual *eventsv1.Event
elements []interface{} elements []interface{}
expect *v1beta1.Event expect *eventsv1.Event
expectUpdate bool expectUpdate bool
}{ }{
{ {
@@ -136,27 +136,27 @@ func TestEventSeriesf(t *testing.T) {
stopCh := make(chan struct{}) stopCh := make(chan struct{})
createEvent := make(chan *v1beta1.Event) createEvent := make(chan *eventsv1.Event)
updateEvent := make(chan *v1beta1.Event) updateEvent := make(chan *eventsv1.Event)
patchEvent := make(chan *v1beta1.Event) patchEvent := make(chan *eventsv1.Event)
testEvents := testEventSeriesSink{ testEvents := testEventSeriesSink{
OnCreate: func(event *v1beta1.Event) (*v1beta1.Event, error) { OnCreate: func(event *eventsv1.Event) (*eventsv1.Event, error) {
createEvent <- event createEvent <- event
return event, nil return event, nil
}, },
OnUpdate: func(event *v1beta1.Event) (*v1beta1.Event, error) { OnUpdate: func(event *eventsv1.Event) (*eventsv1.Event, error) {
updateEvent <- event updateEvent <- event
return event, nil return event, nil
}, },
OnPatch: func(event *v1beta1.Event, patch []byte) (*v1beta1.Event, error) { OnPatch: func(event *eventsv1.Event, patch []byte) (*eventsv1.Event, error) {
// event we receive is already patched, usually the sink uses it only to retrieve the name and namespace, here // event we receive is already patched, usually the sink uses it only to retrieve the name and namespace, here
// we'll use it directly // we'll use it directly
patchEvent <- event patchEvent <- event
return event, nil return event, nil
}, },
} }
eventBroadcaster := newBroadcaster(&testEvents, 0, map[eventKey]*v1beta1.Event{}) eventBroadcaster := newBroadcaster(&testEvents, 0, map[eventKey]*eventsv1.Event{})
recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "eventTest") recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "eventTest")
eventBroadcaster.StartRecordingToSink(stopCh) eventBroadcaster.StartRecordingToSink(stopCh)
recorder.Eventf(regarding, related, isomorphicEvent.Type, isomorphicEvent.Reason, isomorphicEvent.Action, isomorphicEvent.Note, []interface{}{1}) recorder.Eventf(regarding, related, isomorphicEvent.Type, isomorphicEvent.Reason, isomorphicEvent.Action, isomorphicEvent.Note, []interface{}{1})
@@ -179,7 +179,7 @@ func TestEventSeriesf(t *testing.T) {
close(stopCh) close(stopCh)
} }
func validateEvent(messagePrefix string, expectedUpdate bool, actualEvent *v1beta1.Event, expectedEvent *v1beta1.Event, t *testing.T) { func validateEvent(messagePrefix string, expectedUpdate bool, actualEvent *eventsv1.Event, expectedEvent *eventsv1.Event, t *testing.T) {
recvEvent := *actualEvent recvEvent := *actualEvent
// Just check that the timestamp was set. // Just check that the timestamp was set.
@@ -229,32 +229,32 @@ func TestFinishSeries(t *testing.T) {
} }
LastObservedTime := metav1.MicroTime{Time: time.Now().Add(-9 * time.Minute)} LastObservedTime := metav1.MicroTime{Time: time.Now().Add(-9 * time.Minute)}
createEvent := make(chan *v1beta1.Event, 10) createEvent := make(chan *eventsv1.Event, 10)
updateEvent := make(chan *v1beta1.Event, 10) updateEvent := make(chan *eventsv1.Event, 10)
patchEvent := make(chan *v1beta1.Event, 10) patchEvent := make(chan *eventsv1.Event, 10)
testEvents := testEventSeriesSink{ testEvents := testEventSeriesSink{
OnCreate: func(event *v1beta1.Event) (*v1beta1.Event, error) { OnCreate: func(event *eventsv1.Event) (*eventsv1.Event, error) {
createEvent <- event createEvent <- event
return event, nil return event, nil
}, },
OnUpdate: func(event *v1beta1.Event) (*v1beta1.Event, error) { OnUpdate: func(event *eventsv1.Event) (*eventsv1.Event, error) {
updateEvent <- event updateEvent <- event
return event, nil return event, nil
}, },
OnPatch: func(event *v1beta1.Event, patch []byte) (*v1beta1.Event, error) { OnPatch: func(event *eventsv1.Event, patch []byte) (*eventsv1.Event, error) {
// event we receive is already patched, usually the sink uses it // event we receive is already patched, usually the sink uses it
// only to retrieve the name and namespace, here we'll use it directly // only to retrieve the name and namespace, here we'll use it directly
patchEvent <- event patchEvent <- event
return event, nil return event, nil
}, },
} }
cache := map[eventKey]*v1beta1.Event{} cache := map[eventKey]*eventsv1.Event{}
eventBroadcaster := newBroadcaster(&testEvents, 0, cache).(*eventBroadcasterImpl) eventBroadcaster := newBroadcaster(&testEvents, 0, cache).(*eventBroadcasterImpl)
recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "k8s.io/kube-foo").(*recorderImpl) recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "k8s.io/kube-foo").(*recorderImpl)
cachedEvent := recorder.makeEvent(regarding, related, metav1.MicroTime{time.Now()}, v1.EventTypeNormal, "test", "some verbose message: 1", "eventTest", "eventTest-"+hostname, "started") cachedEvent := recorder.makeEvent(regarding, related, metav1.MicroTime{time.Now()}, v1.EventTypeNormal, "test", "some verbose message: 1", "eventTest", "eventTest-"+hostname, "started")
nonFinishedEvent := cachedEvent.DeepCopy() nonFinishedEvent := cachedEvent.DeepCopy()
nonFinishedEvent.ReportingController = "nonFinished-controller" nonFinishedEvent.ReportingController = "nonFinished-controller"
cachedEvent.Series = &v1beta1.EventSeries{ cachedEvent.Series = &eventsv1.EventSeries{
Count: 10, Count: 10,
LastObservedTime: LastObservedTime, LastObservedTime: LastObservedTime,
} }
@@ -300,15 +300,15 @@ func TestRefreshExistingEventSeries(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
LastObservedTime := metav1.MicroTime{Time: time.Now().Add(-9 * time.Minute)} LastObservedTime := metav1.MicroTime{Time: time.Now().Add(-9 * time.Minute)}
createEvent := make(chan *v1beta1.Event, 10) createEvent := make(chan *eventsv1.Event, 10)
updateEvent := make(chan *v1beta1.Event, 10) updateEvent := make(chan *eventsv1.Event, 10)
patchEvent := make(chan *v1beta1.Event, 10) patchEvent := make(chan *eventsv1.Event, 10)
table := []struct { table := []struct {
patchFunc func(event *v1beta1.Event, patch []byte) (*v1beta1.Event, error) patchFunc func(event *eventsv1.Event, patch []byte) (*eventsv1.Event, error)
}{ }{
{ {
patchFunc: func(event *v1beta1.Event, patch []byte) (*v1beta1.Event, error) { patchFunc: func(event *eventsv1.Event, patch []byte) (*eventsv1.Event, error) {
// event we receive is already patched, usually the sink uses it // event we receive is already patched, usually the sink uses it
//only to retrieve the name and namespace, here we'll use it directly. //only to retrieve the name and namespace, here we'll use it directly.
patchEvent <- event patchEvent <- event
@@ -316,7 +316,7 @@ func TestRefreshExistingEventSeries(t *testing.T) {
}, },
}, },
{ {
patchFunc: func(event *v1beta1.Event, patch []byte) (*v1beta1.Event, error) { patchFunc: func(event *eventsv1.Event, patch []byte) (*eventsv1.Event, error) {
// we simulate an apiserver error here // we simulate an apiserver error here
patchEvent <- nil patchEvent <- nil
return nil, &restclient.RequestConstructionError{} return nil, &restclient.RequestConstructionError{}
@@ -325,21 +325,21 @@ func TestRefreshExistingEventSeries(t *testing.T) {
} }
for _, item := range table { for _, item := range table {
testEvents := testEventSeriesSink{ testEvents := testEventSeriesSink{
OnCreate: func(event *v1beta1.Event) (*v1beta1.Event, error) { OnCreate: func(event *eventsv1.Event) (*eventsv1.Event, error) {
createEvent <- event createEvent <- event
return event, nil return event, nil
}, },
OnUpdate: func(event *v1beta1.Event) (*v1beta1.Event, error) { OnUpdate: func(event *eventsv1.Event) (*eventsv1.Event, error) {
updateEvent <- event updateEvent <- event
return event, nil return event, nil
}, },
OnPatch: item.patchFunc, OnPatch: item.patchFunc,
} }
cache := map[eventKey]*v1beta1.Event{} cache := map[eventKey]*eventsv1.Event{}
eventBroadcaster := newBroadcaster(&testEvents, 0, cache).(*eventBroadcasterImpl) eventBroadcaster := newBroadcaster(&testEvents, 0, cache).(*eventBroadcasterImpl)
recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "k8s.io/kube-foo").(*recorderImpl) recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "k8s.io/kube-foo").(*recorderImpl)
cachedEvent := recorder.makeEvent(regarding, related, metav1.MicroTime{time.Now()}, v1.EventTypeNormal, "test", "some verbose message: 1", "eventTest", "eventTest-"+hostname, "started") cachedEvent := recorder.makeEvent(regarding, related, metav1.MicroTime{time.Now()}, v1.EventTypeNormal, "test", "some verbose message: 1", "eventTest", "eventTest-"+hostname, "started")
cachedEvent.Series = &v1beta1.EventSeries{ cachedEvent.Series = &eventsv1.EventSeries{
Count: 10, Count: 10,
LastObservedTime: LastObservedTime, LastObservedTime: LastObservedTime,
} }

8
tools/events/interfaces.go
View File

@@ -17,7 +17,7 @@ limitations under the License.
package events package events
import ( import (
"k8s.io/api/events/v1beta1" eventsv1 "k8s.io/api/events/v1"
"k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/tools/record" "k8s.io/client-go/tools/record"
) )
@@ -65,9 +65,9 @@ type EventBroadcaster interface {
// It is assumed that EventSink will return the same sorts of errors as // It is assumed that EventSink will return the same sorts of errors as
// client-go's REST client. // client-go's REST client.
type EventSink interface { type EventSink interface {
Create(event *v1beta1.Event) (*v1beta1.Event, error) Create(event *eventsv1.Event) (*eventsv1.Event, error)
Update(event *v1beta1.Event) (*v1beta1.Event, error) Update(event *eventsv1.Event) (*eventsv1.Event, error)
Patch(oldEvent *v1beta1.Event, data []byte) (*v1beta1.Event, error) Patch(oldEvent *eventsv1.Event, data []byte) (*eventsv1.Event, error)
} }
// EventBroadcasterAdapter is a auxiliary interface to simplify migration to // EventBroadcasterAdapter is a auxiliary interface to simplify migration to