Fix dependencies to point to kubernetes-1.15.0-beta.2 tag

Kubernetes-commit: 3c5152b9a54ed2273710900be24e5d0828bff79b

Godeps/Godeps.json

@@ -188,15 +188,15 @@
 		},
 		{
 			"ImportPath": "k8s.io/api",
-			"Rev": "576cfa93bfcf"
+			"Rev": "641787b3be95"
 		},
 		{
 			"ImportPath": "k8s.io/apimachinery",
-			"Rev": "587ae2000094"
+			"Rev": "000fc10c9799"
 		},
 		{
 			"ImportPath": "k8s.io/klog",
-			"Rev": "v0.3.1"
+			"Rev": "v0.3.2"
 		},
 		{
 			"ImportPath": "k8s.io/kube-openapi",

go.mod

@@ -26,9 +26,9 @@ require (
 	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/klog v0.3.1
+	k8s.io/api v0.0.0-20190620085001-641787b3be95
+	k8s.io/apimachinery v0.0.0-20190604020035-000fc10c9799
+	k8s.io/klog v0.3.2
 	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.0.0-20190620085001-641787b3be95
+	k8s.io/apimachinery => k8s.io/apimachinery v0.0.0-20190604020035-000fc10c9799
 )

go.sum

@@ -91,10 +91,11 @@ 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=
-k8s.io/klog v0.3.1 h1:RVgyDHY/kFKtLqh67NvEWIgkMneNoIrdkN0CxDSQc68=
-k8s.io/klog v0.3.1/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk=
+k8s.io/api v0.0.0-20190620085001-641787b3be95/go.mod h1:7eKIQT9wWvrNprZDctj9nplM6gv7p9en+A6U3604pHw=
+k8s.io/apimachinery v0.0.0-20190604020035-000fc10c9799 h1:1RlD6VTLZGu2+UZdiyiY5hjvw7Gr99IAScyfu1Tu5Wo=
+k8s.io/apimachinery v0.0.0-20190604020035-000fc10c9799/go.mod h1:u/2VL7tgEMV0FFTV9q0JO+7cnTsV44LP8Pmx41R4AQ4=
+k8s.io/klog v0.3.2 h1:qvP/U6CcZ6qyi/qSHlJKdlAboCzo3mT0DAm0XAarpz4=
+k8s.io/klog v0.3.2/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk=
 k8s.io/kube-openapi v0.0.0-20190228160746-b3a7cee44a30 h1:TRb4wNWoBVrH9plmkp2q86FIDppkbrEXdXlxU3a3BMI=
 k8s.io/kube-openapi v0.0.0-20190228160746-b3a7cee44a30/go.mod h1:BXM9ceUBTj2QnfH2MK1odQs778ajze1RxcmP6S8RVVc=
 k8s.io/utils v0.0.0-20190221042446-c2654d5206da h1:ElyM7RPonbKnQqOcw7dG2IK5uvQQn3b/WPHqD5mBvP4=

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/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/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 {