Update dependencies to v0.15.8-beta.0 tag

Automated cherry pick of #83911: Remove check causing informers to miss notifications

Kubernetes-commit: 0f71449be978a5e9efca3e7431941cc90211b339

Godeps/Godeps.json

@@ -136,7 +136,7 @@
 		},
 		{
 			"ImportPath": "golang.org/x/net",
-			"Rev": "65e2d4e15006"
+			"Rev": "cdfb69ac37fc"
 		},
 		{
 			"ImportPath": "golang.org/x/oauth2",
@@ -184,15 +184,15 @@
 		},
 		{
 			"ImportPath": "gopkg.in/yaml.v2",
-			"Rev": "v2.2.1"
+			"Rev": "v2.2.4"
 		},
 		{
 			"ImportPath": "k8s.io/api",
-			"Rev": "576cfa93bfcf"
+			"Rev": "v0.15.8-beta.0"
 		},
 		{
 			"ImportPath": "k8s.io/apimachinery",
-			"Rev": "587ae2000094"
+			"Rev": "v0.15.8-beta.0"
 		},
 		{
 			"ImportPath": "k8s.io/klog",

go.mod

@@ -22,12 +22,12 @@ require (
 	github.com/spf13/pflag v1.0.1
 	github.com/stretchr/testify v1.2.2
 	golang.org/x/crypto v0.0.0-20181025213731-e84da0312774
-	golang.org/x/net v0.0.0-20190206173232-65e2d4e15006
+	golang.org/x/net v0.0.0-20190812203447-cdfb69ac37fc
 	golang.org/x/oauth2 v0.0.0-20190402181905-9f3314589c9a
 	golang.org/x/time v0.0.0-20161028155119-f51c12702a4d
 	google.golang.org/appengine v1.5.0 // indirect
-	k8s.io/api v0.0.0-20190528154556-576cfa93bfcf
-	k8s.io/apimachinery v0.0.0-20190528154421-587ae2000094
+	k8s.io/api v0.15.8-beta.0
+	k8s.io/apimachinery v0.15.8-beta.0
 	k8s.io/klog v0.3.1
 	k8s.io/utils v0.0.0-20190221042446-c2654d5206da
 	sigs.k8s.io/yaml v1.1.0
@@ -37,6 +37,6 @@ replace (
 	golang.org/x/sync => golang.org/x/sync v0.0.0-20181108010431-42b317875d0f
 	golang.org/x/sys => golang.org/x/sys v0.0.0-20190209173611-3b5209105503
 	golang.org/x/tools => golang.org/x/tools v0.0.0-20190313210603-aa82965741a9
-	k8s.io/api => k8s.io/api v0.0.0-20190528154556-576cfa93bfcf
-	k8s.io/apimachinery => k8s.io/apimachinery v0.0.0-20190528154421-587ae2000094
+	k8s.io/api => k8s.io/api v0.15.8-beta.0
+	k8s.io/apimachinery => k8s.io/apimachinery v0.15.8-beta.0
 )

go.sum

@@ -64,8 +64,8 @@ golang.org/x/crypto v0.0.0-20181025213731-e84da0312774/go.mod h1:6SG95UA2DQfeDnf
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
-golang.org/x/net v0.0.0-20190206173232-65e2d4e15006 h1:bfLnR+k0tq5Lqt6dflRLcZiz6UaXCMt3vhYJ1l4FQ80=
-golang.org/x/net v0.0.0-20190206173232-65e2d4e15006/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190812203447-cdfb69ac37fc h1:gkKoSkUmnU6bpS/VhkuO27bzQeSA51uaEfbOW5dNb68=
+golang.org/x/net v0.0.0-20190812203447-cdfb69ac37fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/oauth2 v0.0.0-20190402181905-9f3314589c9a h1:tImsplftrFpALCYumobsd0K86vlAs/eXGFms2txfJfA=
 golang.org/x/oauth2 v0.0.0-20190402181905-9f3314589c9a/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f h1:Bl/8QSvNqXvPGPGXa2z5xUTmV7VDcZyvRZ+QQXkXTZQ=
@@ -91,8 +91,10 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkep
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.2.1 h1:mUhvW9EsL+naU5Q3cakzfE91YhliOondGd6ZrsDBHQE=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-k8s.io/api v0.0.0-20190528154556-576cfa93bfcf/go.mod h1:kSN4o7sxlIvlMOEe8JbQDVmC4xvMyyK0eQpOLHALC8U=
-k8s.io/apimachinery v0.0.0-20190528154421-587ae2000094/go.mod h1:I4A+glKBHiTgiEjQiCCQfCAIcIMFGt291SmsvcrFzJA=
+gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I=
+gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+k8s.io/api v0.15.8-beta.0/go.mod h1:GwNIL+0thT8hUXk1bLWqinbqwwInzMiJkVcYBdTZw5c=
+k8s.io/apimachinery v0.15.8-beta.0/go.mod h1:Xc10RHc1U+F/e9GCloJ8QAeCGevSVP5xhOhqlE+e1kM=
 k8s.io/klog v0.3.1 h1:RVgyDHY/kFKtLqh67NvEWIgkMneNoIrdkN0CxDSQc68=
 k8s.io/klog v0.3.1/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk=
 k8s.io/kube-openapi v0.0.0-20190228160746-b3a7cee44a30 h1:TRb4wNWoBVrH9plmkp2q86FIDppkbrEXdXlxU3a3BMI=

kubernetes/typed/events/v1beta1/event_expansion.go

@@ -0,0 +1,98 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1beta1
+
+import (
+	"fmt"
+
+	"k8s.io/api/events/v1beta1"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// The EventExpansion interface allows manually adding extra methods to the EventInterface.
+// TODO: Add querying functions to the event expansion
+type EventExpansion interface {
+	// CreateWithEventNamespace is the same as a Create
+	// except that it sends the request to the event.Namespace.
+	CreateWithEventNamespace(event *v1beta1.Event) (*v1beta1.Event, error)
+	// UpdateWithEventNamespace is the same as a Update
+	// except that it sends the request to the event.Namespace.
+	UpdateWithEventNamespace(event *v1beta1.Event) (*v1beta1.Event, error)
+	// PatchWithEventNamespace is the same as an Update
+	// except that it sends the request to the event.Namespace.
+	PatchWithEventNamespace(event *v1beta1.Event, data []byte) (*v1beta1.Event, error)
+}
+
+// CreateWithEventNamespace makes a new event.
+// Returns the copy of the event the server returns, or an error.
+// The namespace to create the event within is deduced from the event.
+// it must either match this event client's namespace, or this event client must
+// have been created with the "" namespace.
+func (e *events) CreateWithEventNamespace(event *v1beta1.Event) (*v1beta1.Event, error) {
+	if e.ns != "" && event.Namespace != e.ns {
+		return nil, fmt.Errorf("can't create an event with namespace '%v' in namespace '%v'", event.Namespace, e.ns)
+	}
+	result := &v1beta1.Event{}
+	err := e.client.Post().
+		NamespaceIfScoped(event.Namespace, len(event.Namespace) > 0).
+		Resource("events").
+		Body(event).
+		Do().
+		Into(result)
+	return result, err
+}
+
+// UpdateWithEventNamespace modifies an existing event.
+// It returns the copy of the event that the server returns, or an error.
+// The namespace and key to update the event within is deduced from the event.
+// The namespace must either match this event client's namespace, or this event client must have been
+// created with the "" namespace.
+// Update also requires the ResourceVersion to be set in the event object.
+func (e *events) UpdateWithEventNamespace(event *v1beta1.Event) (*v1beta1.Event, error) {
+	if e.ns != "" && event.Namespace != e.ns {
+		return nil, fmt.Errorf("can't update an event with namespace '%v' in namespace '%v'", event.Namespace, e.ns)
+	}
+	result := &v1beta1.Event{}
+	err := e.client.Put().
+		NamespaceIfScoped(event.Namespace, len(event.Namespace) > 0).
+		Resource("events").
+		Name(event.Name).
+		Body(event).
+		Do().
+		Into(result)
+	return result, err
+}
+
+// PatchWithEventNamespace modifies an existing event.
+// It returns the copy of the event that the server returns, or an error.
+// The namespace and name of the target event is deduced from the event.
+// The namespace must either match this event client's namespace, or this event client must
+//  have been created with the "" namespace.
+func (e *events) PatchWithEventNamespace(event *v1beta1.Event, data []byte) (*v1beta1.Event, error) {
+	if e.ns != "" && event.Namespace != e.ns {
+		return nil, fmt.Errorf("can't patch an event with namespace '%v' in namespace '%v'", event.Namespace, e.ns)
+	}
+	result := &v1beta1.Event{}
+	err := e.client.Patch(types.StrategicMergePatchType).
+		NamespaceIfScoped(event.Namespace, len(event.Namespace) > 0).
+		Resource("events").
+		Name(event.Name).
+		Body(data).
+		Do().
+		Into(result)
+	return result, err
+}

kubernetes/typed/events/v1beta1/fake/fake_event_expansion.go

@@ -0,0 +1,66 @@
+/*
+Copyright 2014 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package fake
+
+import (
+	v1beta1 "k8s.io/api/events/v1beta1"
+	types "k8s.io/apimachinery/pkg/types"
+	core "k8s.io/client-go/testing"
+)
+
+// CreateWithEventNamespace creats a new event. Returns the copy of the event the server returns, or an error.
+func (c *FakeEvents) CreateWithEventNamespace(event *v1beta1.Event) (*v1beta1.Event, error) {
+	action := core.NewRootCreateAction(eventsResource, event)
+	if c.ns != "" {
+		action = core.NewCreateAction(eventsResource, c.ns, event)
+	}
+	obj, err := c.Fake.Invokes(action, event)
+	if obj == nil {
+		return nil, err
+	}
+
+	return obj.(*v1beta1.Event), err
+}
+
+// UpdateWithEventNamespace replaces an existing event. Returns the copy of the event the server returns, or an error.
+func (c *FakeEvents) UpdateWithEventNamespace(event *v1beta1.Event) (*v1beta1.Event, error) {
+	action := core.NewRootUpdateAction(eventsResource, event)
+	if c.ns != "" {
+		action = core.NewUpdateAction(eventsResource, c.ns, event)
+	}
+	obj, err := c.Fake.Invokes(action, event)
+	if obj == nil {
+		return nil, err
+	}
+
+	return obj.(*v1beta1.Event), err
+}
+
+// PatchWithEventNamespace patches an existing event. Returns the copy of the event the server returns, or an error.
+func (c *FakeEvents) PatchWithEventNamespace(event *v1beta1.Event, data []byte) (*v1beta1.Event, error) {
+	pt := types.StrategicMergePatchType
+	action := core.NewRootPatchAction(eventsResource, event.Name, pt, data)
+	if c.ns != "" {
+		action = core.NewPatchAction(eventsResource, c.ns, event.Name, pt, data)
+	}
+	obj, err := c.Fake.Invokes(action, event)
+	if obj == nil {
+		return nil, err
+	}
+
+	return obj.(*v1beta1.Event), err
+}

kubernetes/typed/events/v1beta1/generated_expansion.go

@@ -17,5 +17,3 @@ limitations under the License.
 // Code generated by client-gen. DO NOT EDIT.
 
 package v1beta1
-
-type EventExpansion interface{}

tools/cache/delta_fifo.go

@@ -295,13 +295,6 @@ func isDeletionDup(a, b *Delta) *Delta {
 	return b
 }
 
-// willObjectBeDeletedLocked returns true only if the last delta for the
-// given object is Delete. Caller must lock first.
-func (f *DeltaFIFO) willObjectBeDeletedLocked(id string) bool {
-	deltas := f.items[id]
-	return len(deltas) > 0 && deltas[len(deltas)-1].Type == Deleted
-}
-
 // queueActionLocked appends to the delta list for the object.
 // Caller must lock first.
 func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
@@ -310,13 +303,6 @@ func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) err
 		return KeyError{obj, err}
 	}
 
-	// If object is supposed to be deleted (last event is Deleted),
-	// then we should ignore Sync events, because it would result in
-	// recreation of this object.
-	if actionType == Sync && f.willObjectBeDeletedLocked(id) {
-		return nil
-	}
-
 	newDeltas := append(f.items[id], Delta{actionType, obj})
 	newDeltas = dedupDeltas(newDeltas)
 

tools/cache/delta_fifo_test.go

@@ -85,6 +85,33 @@ func TestDeltaFIFO_basic(t *testing.T) {
 	}
 }
 
+// TestDeltaFIFO_replaceWithDeleteDeltaIn tests that a `Sync` delta for an
+// object `O` with ID `X` is added when .Replace is called and `O` is among the
+// replacement objects even if the DeltaFIFO already stores in terminal position
+// a delta of type `Delete` for ID `X`. Not adding the `Sync` delta causes
+// SharedIndexInformers to miss `O`'s create notification, see https://github.com/kubernetes/kubernetes/issues/83810
+// for more details.
+func TestDeltaFIFO_replaceWithDeleteDeltaIn(t *testing.T) {
+	oldObj := mkFifoObj("foo", 1)
+	newObj := mkFifoObj("foo", 2)
+
+	f := NewDeltaFIFO(testFifoObjectKeyFunc, keyLookupFunc(func() []testFifoObject {
+		return []testFifoObject{oldObj}
+	}))
+
+	f.Delete(oldObj)
+	f.Replace([]interface{}{newObj}, "")
+
+	actualDeltas := Pop(f)
+	expectedDeltas := Deltas{
+		Delta{Type: Deleted, Object: oldObj},
+		Delta{Type: Sync, Object: newObj},
+	}
+	if !reflect.DeepEqual(expectedDeltas, actualDeltas) {
+		t.Errorf("expected %#v, got %#v", expectedDeltas, actualDeltas)
+	}
+}
+
 func TestDeltaFIFO_requeueOnPop(t *testing.T) {
 	f := NewDeltaFIFO(testFifoObjectKeyFunc, nil)
 

tools/events/event_broadcaster.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/json"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	"k8s.io/apimachinery/pkg/util/wait"
+	typedv1beta1 "k8s.io/client-go/kubernetes/typed/events/v1beta1"
 	"k8s.io/client-go/tools/record/util"
 	"k8s.io/klog"
 )
@@ -65,22 +66,44 @@ type eventBroadcasterImpl struct {
 	sink          EventSink
 }
 
+// EventSinkImpl wraps EventInterface to implement EventSink.
+// 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.
+type EventSinkImpl struct {
+	Interface typedv1beta1.EventInterface
+}
+
+// Create is the same as CreateWithEventNamespace of the EventExpansion
+func (e *EventSinkImpl) Create(event *v1beta1.Event) (*v1beta1.Event, error) {
+	return e.Interface.CreateWithEventNamespace(event)
+}
+
+// Update is the same as UpdateithEventNamespace of the EventExpansion
+func (e *EventSinkImpl) Update(event *v1beta1.Event) (*v1beta1.Event, error) {
+	return e.Interface.UpdateWithEventNamespace(event)
+}
+
+// Patch is the same as PatchWithEventNamespace of the EventExpansion
+func (e *EventSinkImpl) Patch(event *v1beta1.Event, data []byte) (*v1beta1.Event, error) {
+	return e.Interface.PatchWithEventNamespace(event, data)
+}
+
 // NewBroadcaster Creates a new event broadcaster.
 func NewBroadcaster(sink EventSink) EventBroadcaster {
-	return newBroadcaster(sink, defaultSleepDuration)
+	return newBroadcaster(sink, defaultSleepDuration, map[eventKey]*v1beta1.Event{})
 }
 
 // NewBroadcasterForTest Creates a new event broadcaster for test purposes.
-func newBroadcaster(sink EventSink, sleepDuration time.Duration) EventBroadcaster {
+func newBroadcaster(sink EventSink, sleepDuration time.Duration, eventCache map[eventKey]*v1beta1.Event) EventBroadcaster {
 	return &eventBroadcasterImpl{
 		Broadcaster:   watch.NewBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
-		eventCache:    map[eventKey]*v1beta1.Event{},
+		eventCache:    eventCache,
 		sleepDuration: sleepDuration,
 		sink:          sink,
 	}
 }
 
-// TODO: add test for refreshExistingEventSeries
+// refreshExistingEventSeries refresh events TTL
 func (e *eventBroadcasterImpl) refreshExistingEventSeries() {
 	// TODO: Investigate whether lock contention won't be a problem
 	e.mu.Lock()
@@ -94,19 +117,23 @@ func (e *eventBroadcasterImpl) refreshExistingEventSeries() {
 	}
 }
 
-// TODO: add test for finishSeries
+// finishSeries checks if a series has ended and either:
+// - write final count to the apiserver
+// - delete a singleton event (i.e. series field is nil) from the cache
 func (e *eventBroadcasterImpl) finishSeries() {
 	// TODO: Investigate whether lock contention won't be a problem
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	for isomorphicKey, event := range e.eventCache {
-		eventSeries := event.Series
-		if eventSeries != nil {
-			if eventSeries.LastObservedTime.Time.Add(finishTime).Before(time.Now()) {
+		eventSerie := event.Series
+		if eventSerie != nil {
+			if eventSerie.LastObservedTime.Time.Before(time.Now().Add(-finishTime)) {
 				if _, retry := recordEvent(e.sink, event); !retry {
 					delete(e.eventCache, isomorphicKey)
 				}
 			}
+		} else if event.EventTime.Time.Before(time.Now().Add(-finishTime)) {
+			delete(e.eventCache, isomorphicKey)
 		}
 	}
 }

tools/events/event_recorder.go

@@ -85,5 +85,8 @@ func (recorder *recorderImpl) makeEvent(refRegarding *v1.ObjectReference, refRel
 		Related:             refRelated,
 		Note:                message,
 		Type:                eventtype,
+		// TODO: remove this when we change conversion to convert eventSource
+		// to reportingController
+		DeprecatedSource: v1.EventSource{Component: reportingController},
 	}
 }

tools/events/eventseries_test.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/api/events/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8sruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes/scheme"
 	ref "k8s.io/client-go/tools/reference"
 )
@@ -154,7 +155,7 @@ func TestEventSeriesf(t *testing.T) {
 			return event, nil
 		},
 	}
-	eventBroadcaster := newBroadcaster(&testEvents, 0)
+	eventBroadcaster := newBroadcaster(&testEvents, 0, map[eventKey]*v1beta1.Event{})
 	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "eventTest")
 	eventBroadcaster.StartRecordingToSink(stopCh)
 	recorder.Eventf(regarding, related, isomorphicEvent.Type, isomorphicEvent.Reason, isomorphicEvent.Action, isomorphicEvent.Note, []interface{}{1})
@@ -206,3 +207,143 @@ func validateEventSerie(messagePrefix string, expectedUpdate bool, actualEvent *
 	}
 
 }
+
+func TestFinishSeries(t *testing.T) {
+	hostname, _ := os.Hostname()
+	testPod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			SelfLink:  "/api/version/pods/foo",
+			Name:      "foo",
+			Namespace: "baz",
+			UID:       "bar",
+		},
+	}
+	regarding, err := ref.GetPartialReference(scheme.Scheme, testPod, ".spec.containers[1]")
+	if err != nil {
+		t.Fatal(err)
+	}
+	related, err := ref.GetPartialReference(scheme.Scheme, testPod, ".spec.containers[0]")
+	if err != nil {
+		t.Fatal(err)
+	}
+	LastObservedTime := metav1.MicroTime{Time: time.Now().Add(-9 * time.Minute)}
+
+	createEvent := make(chan *v1beta1.Event, 10)
+	updateEvent := make(chan *v1beta1.Event, 10)
+	patchEvent := make(chan *v1beta1.Event, 10)
+	testEvents := testEventSeriesSink{
+		OnCreate: func(event *v1beta1.Event) (*v1beta1.Event, error) {
+			createEvent <- event
+			return event, nil
+		},
+		OnUpdate: func(event *v1beta1.Event) (*v1beta1.Event, error) {
+			updateEvent <- event
+			return event, nil
+		},
+		OnPatch: func(event *v1beta1.Event, patch []byte) (*v1beta1.Event, error) {
+			// event we receive is already patched, usually the sink uses it
+			// only to retrieve the name and namespace, here we'll use it directly
+			patchEvent <- event
+			return event, nil
+		},
+	}
+	cache := map[eventKey]*v1beta1.Event{}
+	eventBroadcaster := newBroadcaster(&testEvents, 0, cache).(*eventBroadcasterImpl)
+	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")
+	nonFinishedEvent := cachedEvent.DeepCopy()
+	nonFinishedEvent.ReportingController = "nonFinished-controller"
+	cachedEvent.Series = &v1beta1.EventSeries{
+		Count:            10,
+		LastObservedTime: LastObservedTime,
+	}
+	cache[getKey(cachedEvent)] = cachedEvent
+	cache[getKey(nonFinishedEvent)] = nonFinishedEvent
+	eventBroadcaster.finishSeries()
+	select {
+	case actualEvent := <-patchEvent:
+		t.Logf("validating event affected by patch request")
+		eventBroadcaster.mu.Lock()
+		defer eventBroadcaster.mu.Unlock()
+		if len(cache) != 1 {
+			t.Errorf("cache should be empty, but instead got a size of %v", len(cache))
+		}
+		if !actualEvent.Series.LastObservedTime.Equal(&cachedEvent.Series.LastObservedTime) {
+			t.Errorf("series was expected be seen with LastObservedTime %v, but instead got %v ", cachedEvent.Series.LastObservedTime, actualEvent.Series.LastObservedTime)
+		}
+		// check that we emitted only one event
+		if len(patchEvent) != 0 || len(createEvent) != 0 || len(updateEvent) != 0 {
+			t.Errorf("exactly one event should be emitted, but got %v", len(patchEvent))
+		}
+	case <-time.After(wait.ForeverTestTimeout):
+		t.Fatalf("timeout after %v", wait.ForeverTestTimeout)
+	}
+}
+
+func TestRefreshExistingEventSeries(t *testing.T) {
+	hostname, _ := os.Hostname()
+	testPod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			SelfLink:  "/api/version/pods/foo",
+			Name:      "foo",
+			Namespace: "baz",
+			UID:       "bar",
+		},
+	}
+	regarding, err := ref.GetPartialReference(scheme.Scheme, testPod, ".spec.containers[1]")
+	if err != nil {
+		t.Fatal(err)
+	}
+	related, err := ref.GetPartialReference(scheme.Scheme, testPod, ".spec.containers[0]")
+	if err != nil {
+		t.Fatal(err)
+	}
+	LastObservedTime := metav1.MicroTime{Time: time.Now().Add(-9 * time.Minute)}
+
+	createEvent := make(chan *v1beta1.Event, 10)
+	updateEvent := make(chan *v1beta1.Event, 10)
+	patchEvent := make(chan *v1beta1.Event, 10)
+	testEvents := testEventSeriesSink{
+		OnCreate: func(event *v1beta1.Event) (*v1beta1.Event, error) {
+			createEvent <- event
+			return event, nil
+		},
+		OnUpdate: func(event *v1beta1.Event) (*v1beta1.Event, error) {
+			updateEvent <- event
+			return event, nil
+		},
+		OnPatch: func(event *v1beta1.Event, patch []byte) (*v1beta1.Event, error) {
+			// event we receive is already patched, usually the sink uses it
+			//only to retrieve the name and namespace, here we'll use it directly.
+			patchEvent <- event
+			return event, nil
+		},
+	}
+	cache := map[eventKey]*v1beta1.Event{}
+	eventBroadcaster := newBroadcaster(&testEvents, 0, cache).(*eventBroadcasterImpl)
+	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.Series = &v1beta1.EventSeries{
+		Count:            10,
+		LastObservedTime: LastObservedTime,
+	}
+	cacheKey := getKey(cachedEvent)
+	cache[cacheKey] = cachedEvent
+
+	eventBroadcaster.refreshExistingEventSeries()
+	select {
+	case <-patchEvent:
+		t.Logf("validating event affected by patch request")
+		eventBroadcaster.mu.Lock()
+		defer eventBroadcaster.mu.Unlock()
+		if len(cache) != 1 {
+			t.Errorf("cache should be with same size, but instead got a size of %v", len(cache))
+		}
+		// check that we emitted only one event
+		if len(patchEvent) != 0 || len(createEvent) != 0 || len(updateEvent) != 0 {
+			t.Errorf("exactly one event should be emitted, but got %v", len(patchEvent))
+		}
+	case <-time.After(wait.ForeverTestTimeout):
+		t.Fatalf("timeout after %v", wait.ForeverTestTimeout)
+	}
+}

tools/record/event_test.go

@@ -153,7 +153,7 @@ func TestEventf(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "bar",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[2]",
 				},
 				Reason:  "Started",
@@ -162,7 +162,7 @@ func TestEventf(t *testing.T) {
 				Count:   1,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
 			expectUpdate: false,
 		},
 		{
@@ -181,7 +181,7 @@ func TestEventf(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "bar",
-					APIVersion: "version",
+					APIVersion: "v1",
 				},
 				Reason:  "Killed",
 				Message: "some other verbose message: 1",
@@ -189,7 +189,7 @@ func TestEventf(t *testing.T) {
 				Count:   1,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"version", ResourceVersion:"", FieldPath:""}): type: 'Normal' reason: 'Killed' some other verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"v1", ResourceVersion:"", FieldPath:""}): type: 'Normal' reason: 'Killed' some other verbose message: 1`,
 			expectUpdate: false,
 		},
 		{
@@ -208,7 +208,7 @@ func TestEventf(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "bar",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[2]",
 				},
 				Reason:  "Started",
@@ -217,7 +217,7 @@ func TestEventf(t *testing.T) {
 				Count:   2,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
 			expectUpdate: true,
 		},
 		{
@@ -236,7 +236,7 @@ func TestEventf(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "differentUid",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[3]",
 				},
 				Reason:  "Started",
@@ -245,7 +245,7 @@ func TestEventf(t *testing.T) {
 				Count:   1,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
 			expectUpdate: false,
 		},
 		{
@@ -264,7 +264,7 @@ func TestEventf(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "bar",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[2]",
 				},
 				Reason:  "Started",
@@ -273,7 +273,7 @@ func TestEventf(t *testing.T) {
 				Count:   3,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
 			expectUpdate: true,
 		},
 		{
@@ -292,7 +292,7 @@ func TestEventf(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "differentUid",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[3]",
 				},
 				Reason:  "Stopped",
@@ -301,7 +301,7 @@ func TestEventf(t *testing.T) {
 				Count:   1,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Stopped' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Stopped' some verbose message: 1`,
 			expectUpdate: false,
 		},
 		{
@@ -320,7 +320,7 @@ func TestEventf(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "differentUid",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[3]",
 				},
 				Reason:  "Stopped",
@@ -329,7 +329,7 @@ func TestEventf(t *testing.T) {
 				Count:   2,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Stopped' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Stopped' some verbose message: 1`,
 			expectUpdate: true,
 		},
 	}
@@ -509,7 +509,7 @@ func TestLotsOfEvents(t *testing.T) {
 			Name:       fmt.Sprintf("foo-%v", i),
 			Namespace:  "baz",
 			UID:        "bar",
-			APIVersion: "version",
+			APIVersion: "v1",
 		}
 		// we need to vary the reason to prevent aggregation
 		go recorder.Eventf(ref, v1.EventTypeNormal, "Reason-"+string(i), strconv.Itoa(i))
@@ -567,7 +567,7 @@ func TestEventfNoNamespace(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "",
 					UID:        "bar",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[2]",
 				},
 				Reason:  "Started",
@@ -576,7 +576,7 @@ func TestEventfNoNamespace(t *testing.T) {
 				Count:   1,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"", Name:"foo", UID:"bar", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"", Name:"foo", UID:"bar", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
 			expectUpdate: false,
 		},
 	}
@@ -677,7 +677,7 @@ func TestMultiSinkCache(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "bar",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[2]",
 				},
 				Reason:  "Started",
@@ -686,7 +686,7 @@ func TestMultiSinkCache(t *testing.T) {
 				Count:   1,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
 			expectUpdate: false,
 		},
 		{
@@ -705,7 +705,7 @@ func TestMultiSinkCache(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "bar",
-					APIVersion: "version",
+					APIVersion: "v1",
 				},
 				Reason:  "Killed",
 				Message: "some other verbose message: 1",
@@ -713,7 +713,7 @@ func TestMultiSinkCache(t *testing.T) {
 				Count:   1,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"version", ResourceVersion:"", FieldPath:""}): type: 'Normal' reason: 'Killed' some other verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"v1", ResourceVersion:"", FieldPath:""}): type: 'Normal' reason: 'Killed' some other verbose message: 1`,
 			expectUpdate: false,
 		},
 		{
@@ -732,7 +732,7 @@ func TestMultiSinkCache(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "bar",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[2]",
 				},
 				Reason:  "Started",
@@ -741,7 +741,7 @@ func TestMultiSinkCache(t *testing.T) {
 				Count:   2,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
 			expectUpdate: true,
 		},
 		{
@@ -760,7 +760,7 @@ func TestMultiSinkCache(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "differentUid",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[3]",
 				},
 				Reason:  "Started",
@@ -769,7 +769,7 @@ func TestMultiSinkCache(t *testing.T) {
 				Count:   1,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
 			expectUpdate: false,
 		},
 		{
@@ -788,7 +788,7 @@ func TestMultiSinkCache(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "bar",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[2]",
 				},
 				Reason:  "Started",
@@ -797,7 +797,7 @@ func TestMultiSinkCache(t *testing.T) {
 				Count:   3,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"bar", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[2]"}): type: 'Normal' reason: 'Started' some verbose message: 1`,
 			expectUpdate: true,
 		},
 		{
@@ -816,7 +816,7 @@ func TestMultiSinkCache(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "differentUid",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[3]",
 				},
 				Reason:  "Stopped",
@@ -825,7 +825,7 @@ func TestMultiSinkCache(t *testing.T) {
 				Count:   1,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Stopped' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Stopped' some verbose message: 1`,
 			expectUpdate: false,
 		},
 		{
@@ -844,7 +844,7 @@ func TestMultiSinkCache(t *testing.T) {
 					Name:       "foo",
 					Namespace:  "baz",
 					UID:        "differentUid",
-					APIVersion: "version",
+					APIVersion: "v1",
 					FieldPath:  "spec.containers[3]",
 				},
 				Reason:  "Stopped",
@@ -853,7 +853,7 @@ func TestMultiSinkCache(t *testing.T) {
 				Count:   2,
 				Type:    v1.EventTypeNormal,
 			},
-			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"version", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Stopped' some verbose message: 1`,
+			expectLog:    `Event(v1.ObjectReference{Kind:"Pod", Namespace:"baz", Name:"foo", UID:"differentUid", APIVersion:"v1", ResourceVersion:"", FieldPath:"spec.containers[3]"}): type: 'Normal' reason: 'Stopped' some verbose message: 1`,
 			expectUpdate: true,
 		},
 	}

tools/reference/ref.go

@@ -19,8 +19,6 @@ package reference
 import (
 	"errors"
 	"fmt"
-	"net/url"
-	"strings"
 
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -30,8 +28,7 @@ import (
 
 var (
 	// Errors that could be returned by GetReference.
-	ErrNilObject  = errors.New("can't reference a nil object")
-	ErrNoSelfLink = errors.New("selfLink was empty, can't make reference")
+	ErrNilObject = errors.New("can't reference a nil object")
 )
 
 // GetReference returns an ObjectReference which refers to the given
@@ -47,20 +44,6 @@ func GetReference(scheme *runtime.Scheme, obj runtime.Object) (*v1.ObjectReferen
 		return ref, nil
 	}
 
-	gvk := obj.GetObjectKind().GroupVersionKind()
-
-	// if the object referenced is actually persisted, we can just get kind from meta
-	// if we are building an object reference to something not yet persisted, we should fallback to scheme
-	kind := gvk.Kind
-	if len(kind) == 0 {
-		// TODO: this is wrong
-		gvks, _, err := scheme.ObjectKinds(obj)
-		if err != nil {
-			return nil, err
-		}
-		kind = gvks[0].Kind
-	}
-
 	// An object that implements only List has enough metadata to build a reference
 	var listMeta metav1.Common
 	objectMeta, err := meta.Accessor(obj)
@@ -73,29 +56,29 @@ func GetReference(scheme *runtime.Scheme, obj runtime.Object) (*v1.ObjectReferen
 		listMeta = objectMeta
 	}
 
-	// if the object referenced is actually persisted, we can also get version from meta
-	version := gvk.GroupVersion().String()
-	if len(version) == 0 {
-		selfLink := listMeta.GetSelfLink()
-		if len(selfLink) == 0 {
-			return nil, ErrNoSelfLink
-		}
-		selfLinkUrl, err := url.Parse(selfLink)
+	gvk := obj.GetObjectKind().GroupVersionKind()
+
+	// If object meta doesn't contain data about kind and/or version,
+	// we are falling back to scheme.
+	//
+	// TODO: This doesn't work for CRDs, which are not registered in scheme.
+	if gvk.Empty() {
+		gvks, _, err := scheme.ObjectKinds(obj)
 		if err != nil {
 			return nil, err
 		}
-		// example paths: /<prefix>/<version>/*
-		parts := strings.Split(selfLinkUrl.Path, "/")
-		if len(parts) < 4 {
-			return nil, fmt.Errorf("unexpected self link format: '%v'; got version '%v'", selfLink, version)
-		}
-		if parts[1] == "api" {
-			version = parts[2]
-		} else {
-			version = parts[2] + "/" + parts[3]
+		if len(gvks) == 0 || gvks[0].Empty() {
+			return nil, fmt.Errorf("unexpected gvks registered for object %T: %v", obj, gvks)
 		}
+		// TODO: The same object can be registered for multiple group versions
+		// (although in practise this doesn't seem to be used).
+		// In such case, the version set may not be correct.
+		gvk = gvks[0]
 	}
 
+	kind := gvk.Kind
+	version := gvk.GroupVersion().String()
+
 	// only has list metadata
 	if objectMeta == nil {
 		return &v1.ObjectReference{

tools/reference/ref_test.go

@@ -37,29 +37,31 @@ func TestGetReferenceRefVersion(t *testing.T) {
 	tests := []struct {
 		name               string
 		input              *TestRuntimeObj
+		groupVersion       schema.GroupVersion
 		expectedRefVersion string
 	}{
 		{
-			name: "api from selflink",
+			name: "v1 GV from scheme",
 			input: &TestRuntimeObj{
-				ObjectMeta: metav1.ObjectMeta{SelfLink: "/api/v1/namespaces"},
+				ObjectMeta: metav1.ObjectMeta{SelfLink: "/bad-selflink/unused"},
 			},
+			groupVersion:       schema.GroupVersion{Group: "", Version: "v1"},
 			expectedRefVersion: "v1",
 		},
 		{
-			name: "foo.group/v3 from selflink",
+			name: "foo.group/v3 GV from scheme",
 			input: &TestRuntimeObj{
-				ObjectMeta: metav1.ObjectMeta{SelfLink: "/apis/foo.group/v3/namespaces"},
+				ObjectMeta: metav1.ObjectMeta{SelfLink: "/bad-selflink/unused"},
 			},
+			groupVersion:       schema.GroupVersion{Group: "foo.group", Version: "v3"},
 			expectedRefVersion: "foo.group/v3",
 		},
 	}
 
-	scheme := runtime.NewScheme()
-	scheme.AddKnownTypes(schema.GroupVersion{Group: "this", Version: "is ignored"}, &TestRuntimeObj{})
-
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
+			scheme := runtime.NewScheme()
+			scheme.AddKnownTypes(test.groupVersion, &TestRuntimeObj{})
 			ref, err := GetReference(scheme, test.input)
 			if err != nil {
 				t.Fatal(err)

tools/watch/informerwatcher.go

@@ -18,42 +18,86 @@ package watch
 
 import (
 	"sync"
-	"sync/atomic"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/tools/cache"
 )
 
-func newTicketer() *ticketer {
-	return &ticketer{
+func newEventProcessor(out chan<- watch.Event) *eventProcessor {
+	return &eventProcessor{
+		out:  out,
 		cond: sync.NewCond(&sync.Mutex{}),
+		done: make(chan struct{}),
 	}
 }
 
-type ticketer struct {
-	counter uint64
+// eventProcessor buffers events and writes them to an out chan when a reader
+// is waiting. Because of the requirement to buffer events, it synchronizes
+// input with a condition, and synchronizes output with a channels. It needs to
+// be able to yield while both waiting on an input condition and while blocked
+// on writing to the output channel.
+type eventProcessor struct {
+	out chan<- watch.Event
 
-	cond    *sync.Cond
-	current uint64
+	cond *sync.Cond
+	buff []watch.Event
+
+	done chan struct{}
 }
 
-func (t *ticketer) GetTicket() uint64 {
-	// -1 to start from 0
-	return atomic.AddUint64(&t.counter, 1) - 1
+func (e *eventProcessor) run() {
+	for {
+		batch := e.takeBatch()
+		e.writeBatch(batch)
+		if e.stopped() {
+			return
+		}
+	}
 }
 
-func (t *ticketer) WaitForTicket(ticket uint64, f func()) {
-	t.cond.L.Lock()
-	defer t.cond.L.Unlock()
-	for ticket != t.current {
-		t.cond.Wait()
+func (e *eventProcessor) takeBatch() []watch.Event {
+	e.cond.L.Lock()
+	defer e.cond.L.Unlock()
+
+	for len(e.buff) == 0 && !e.stopped() {
+		e.cond.Wait()
 	}
 
-	f()
+	batch := e.buff
+	e.buff = nil
+	return batch
+}
 
-	t.current++
-	t.cond.Broadcast()
+func (e *eventProcessor) writeBatch(events []watch.Event) {
+	for _, event := range events {
+		select {
+		case e.out <- event:
+		case <-e.done:
+			return
+		}
+	}
+}
+
+func (e *eventProcessor) push(event watch.Event) {
+	e.cond.L.Lock()
+	defer e.cond.L.Unlock()
+	defer e.cond.Signal()
+	e.buff = append(e.buff, event)
+}
+
+func (e *eventProcessor) stopped() bool {
+	select {
+	case <-e.done:
+		return true
+	default:
+		return false
+	}
+}
+
+func (e *eventProcessor) stop() {
+	close(e.done)
+	e.cond.Signal()
 }
 
 // NewIndexerInformerWatcher will create an IndexerInformer and wrap it into watch.Interface
@@ -61,55 +105,44 @@ func (t *ticketer) WaitForTicket(ticket uint64, f func()) {
 // it also returns a channel you can use to wait for the informers to fully shutdown.
 func NewIndexerInformerWatcher(lw cache.ListerWatcher, objType runtime.Object) (cache.Indexer, cache.Controller, watch.Interface, <-chan struct{}) {
 	ch := make(chan watch.Event)
-	doneCh := make(chan struct{})
 	w := watch.NewProxyWatcher(ch)
-	t := newTicketer()
+	e := newEventProcessor(ch)
 
 	indexer, informer := cache.NewIndexerInformer(lw, objType, 0, cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
-			go t.WaitForTicket(t.GetTicket(), func() {
-				select {
-				case ch <- watch.Event{
-					Type:   watch.Added,
-					Object: obj.(runtime.Object),
-				}:
-				case <-w.StopChan():
-				}
+			e.push(watch.Event{
+				Type:   watch.Added,
+				Object: obj.(runtime.Object),
 			})
 		},
 		UpdateFunc: func(old, new interface{}) {
-			go t.WaitForTicket(t.GetTicket(), func() {
-				select {
-				case ch <- watch.Event{
-					Type:   watch.Modified,
-					Object: new.(runtime.Object),
-				}:
-				case <-w.StopChan():
-				}
+			e.push(watch.Event{
+				Type:   watch.Modified,
+				Object: new.(runtime.Object),
 			})
 		},
 		DeleteFunc: func(obj interface{}) {
-			go t.WaitForTicket(t.GetTicket(), func() {
-				staleObj, stale := obj.(cache.DeletedFinalStateUnknown)
-				if stale {
-					// We have no means of passing the additional information down using watch API based on watch.Event
-					// but the caller can filter such objects by checking if metadata.deletionTimestamp is set
-					obj = staleObj
-				}
+			staleObj, stale := obj.(cache.DeletedFinalStateUnknown)
+			if stale {
+				// We have no means of passing the additional information down using
+				// watch API based on watch.Event but the caller can filter such
+				// objects by checking if metadata.deletionTimestamp is set
+				obj = staleObj
+			}
 
-				select {
-				case ch <- watch.Event{
-					Type:   watch.Deleted,
-					Object: obj.(runtime.Object),
-				}:
-				case <-w.StopChan():
-				}
+			e.push(watch.Event{
+				Type:   watch.Deleted,
+				Object: obj.(runtime.Object),
 			})
 		},
 	}, cache.Indexers{})
 
+	go e.run()
+
+	doneCh := make(chan struct{})
 	go func() {
 		defer close(doneCh)
+		defer e.stop()
 		informer.Run(w.StopChan())
 	}()
 

tools/watch/informerwatcher_test.go

@@ -17,8 +17,9 @@ limitations under the License.
 package watch
 
 import (
-	"math/rand"
+	"context"
 	"reflect"
+	goruntime "runtime"
 	"sort"
 	"testing"
 	"time"
@@ -28,6 +29,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/diff"
 	"k8s.io/apimachinery/pkg/watch"
 	fakeclientset "k8s.io/client-go/kubernetes/fake"
@@ -35,6 +37,86 @@ import (
 	"k8s.io/client-go/tools/cache"
 )
 
+// TestEventProcessorExit is expected to timeout if the event processor fails
+// to exit when stopped.
+func TestEventProcessorExit(t *testing.T) {
+	event := watch.Event{}
+
+	tests := []struct {
+		name  string
+		write func(e *eventProcessor)
+	}{
+		{
+			name: "exit on blocked read",
+			write: func(e *eventProcessor) {
+				e.push(event)
+			},
+		},
+		{
+			name: "exit on blocked write",
+			write: func(e *eventProcessor) {
+				e.push(event)
+				e.push(event)
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			out := make(chan watch.Event)
+			e := newEventProcessor(out)
+
+			test.write(e)
+
+			exited := make(chan struct{})
+			go func() {
+				e.run()
+				close(exited)
+			}()
+
+			<-out
+			e.stop()
+			goruntime.Gosched()
+			<-exited
+		})
+	}
+}
+
+type apiInt int
+
+func (apiInt) GetObjectKind() schema.ObjectKind { return nil }
+func (apiInt) DeepCopyObject() runtime.Object   { return nil }
+
+func TestEventProcessorOrdersEvents(t *testing.T) {
+	out := make(chan watch.Event)
+	e := newEventProcessor(out)
+	go e.run()
+
+	numProcessed := 0
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	go func() {
+		for i := 0; i < 1000; i++ {
+			e := <-out
+			if got, want := int(e.Object.(apiInt)), i; got != want {
+				t.Errorf("unexpected event: got=%d, want=%d", got, want)
+			}
+			numProcessed++
+		}
+		cancel()
+	}()
+
+	for i := 0; i < 1000; i++ {
+		e.push(watch.Event{Object: apiInt(i)})
+	}
+
+	<-ctx.Done()
+	e.stop()
+
+	if numProcessed != 1000 {
+		t.Errorf("unexpected number of events processed: %d", numProcessed)
+	}
+
+}
+
 type byEventTypeAndName []watch.Event
 
 func (a byEventTypeAndName) Len() int      { return len(a) }
@@ -51,44 +133,6 @@ func (a byEventTypeAndName) Less(i, j int) bool {
 	return a[i].Object.(*corev1.Secret).Name < a[j].Object.(*corev1.Secret).Name
 }
 
-func TestTicketer(t *testing.T) {
-	tg := newTicketer()
-
-	const numTickets = 100 // current golang limit for race detector is 8192 simultaneously alive goroutines
-	var tickets []uint64
-	for i := 0; i < numTickets; i++ {
-		ticket := tg.GetTicket()
-		tickets = append(tickets, ticket)
-
-		exp, got := uint64(i), ticket
-		if got != exp {
-			t.Fatalf("expected ticket %d, got %d", exp, got)
-		}
-	}
-
-	// shuffle tickets
-	rand.Shuffle(len(tickets), func(i, j int) {
-		tickets[i], tickets[j] = tickets[j], tickets[i]
-	})
-
-	res := make(chan uint64, len(tickets))
-	for _, ticket := range tickets {
-		go func(ticket uint64) {
-			time.Sleep(time.Duration(rand.Intn(50)) * time.Millisecond)
-			tg.WaitForTicket(ticket, func() {
-				res <- ticket
-			})
-		}(ticket)
-	}
-
-	for i := 0; i < numTickets; i++ {
-		exp, got := uint64(i), <-res
-		if got != exp {
-			t.Fatalf("expected ticket %d, got %d", exp, got)
-		}
-	}
-}
-
 func TestNewInformerWatcher(t *testing.T) {
 	// Make sure there are no 2 same types of events on a secret with the same name or that might be flaky.
 	tt := []struct {