Support per-handler resync periods

staging/src/k8s.io/client-go/tools/cache/controller.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/util/clock"
 )
 
 // Config contains all the settings for a Controller.
@@ -50,6 +51,11 @@ type Config struct {
 	// queue.
 	FullResyncPeriod time.Duration
 
+	// ShouldResync, if specified, is invoked when the controller's reflector determines the next
+	// periodic sync should occur. If this returns true, it means the reflector should proceed with
+	// the resync.
+	ShouldResync ShouldResyncFunc
+
 	// If true, when Process() returns an error, re-enqueue the object.
 	// TODO: add interface to let you inject a delay/backoff or drop
 	//       the object completely if desired. Pass the object in
@@ -57,6 +63,11 @@ type Config struct {
 	RetryOnError bool
 }
 
+// ShouldResyncFunc is a type of function that indicates if a reflector should perform a
+// resync or not. It can be used by a shared informer to support multiple event handlers with custom
+// resync periods.
+type ShouldResyncFunc func() bool
+
 // ProcessFunc processes a single object.
 type ProcessFunc func(obj interface{}) error
 
@@ -65,6 +76,7 @@ type controller struct {
 	config         Config
 	reflector      *Reflector
 	reflectorMutex sync.RWMutex
+	clock          clock.Clock
 }
 
 type Controller interface {
@@ -77,6 +89,7 @@ type Controller interface {
 func New(c *Config) Controller {
 	ctlr := &controller{
 		config: *c,
+		clock:  &clock.RealClock{},
 	}
 	return ctlr
 }
@@ -92,6 +105,8 @@ func (c *controller) Run(stopCh <-chan struct{}) {
 		c.config.Queue,
 		c.config.FullResyncPeriod,
 	)
+	r.ShouldResync = c.config.ShouldResync
+	r.clock = c.clock
 
 	c.reflectorMutex.Lock()
 	c.reflector = r

staging/src/k8s.io/client-go/tools/cache/processor_listener_test.go

@@ -26,7 +26,7 @@ import (
 // TestPopReleaseLock tests that when processor listener blocks on chan,
 // it should release the lock for pendingNotifications.
 func TestPopReleaseLock(t *testing.T) {
-	pl := newProcessListener(nil)
+	pl := newProcessListener(nil, 0, 0, time.Now())
 	stopCh := make(chan struct{})
 	defer close(stopCh)
 	// make pop() block on nextCh: waiting for receiver to get notification.

staging/src/k8s.io/client-go/tools/cache/reflector.go

@@ -41,6 +41,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/util/clock"
 )
 
 // Reflector watches a specified resource and causes all changes to be reflected in the given store.
@@ -58,8 +59,9 @@ type Reflector struct {
 	// the beginning of the next one.
 	period       time.Duration
 	resyncPeriod time.Duration
-	// now() returns current time - exposed for testing purposes
-	now func() time.Time
+	ShouldResync func() bool
+	// clock allows tests to manipulate time
+	clock clock.Clock
 	// lastSyncResourceVersion is the resource version token last
 	// observed when doing a sync with the underlying store
 	// it is thread safe, but not synchronized with the underlying store
@@ -103,7 +105,7 @@ func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{},
 		expectedType:  reflect.TypeOf(expectedType),
 		period:        time.Second,
 		resyncPeriod:  resyncPeriod,
-		now:           time.Now,
+		clock:         &clock.RealClock{},
 	}
 	return r
 }
@@ -223,8 +225,8 @@ func (r *Reflector) resyncChan() (<-chan time.Time, func() bool) {
 	// always fail so we end up listing frequently. Then, if we don't
 	// manually stop the timer, we could end up with many timers active
 	// concurrently.
-	t := time.NewTimer(r.resyncPeriod)
-	return t.C, t.Stop
+	t := r.clock.NewTimer(r.resyncPeriod)
+	return t.C(), t.Stop
 }
 
 // ListAndWatch first lists all items and get the resource version at the moment of call,
@@ -270,10 +272,12 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
 			case <-cancelCh:
 				return
 			}
-			glog.V(4).Infof("%s: forcing resync", r.name)
-			if err := r.store.Resync(); err != nil {
-				resyncerrc <- err
-				return
+			if r.ShouldResync == nil || r.ShouldResync() {
+				glog.V(4).Infof("%s: forcing resync", r.name)
+				if err := r.store.Resync(); err != nil {
+					resyncerrc <- err
+					return
+				}
 			}
 			cleanup()
 			resyncCh, cleanup = r.resyncChan()
@@ -334,7 +338,7 @@ func (r *Reflector) syncWith(items []runtime.Object, resourceVersion string) err
 
 // watchHandler watches w and keeps *resourceVersion up to date.
 func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error {
-	start := time.Now()
+	start := r.clock.Now()
 	eventCount := 0
 
 	// Stopping the watcher should be idempotent and if we return from this function there's no way
@@ -393,7 +397,7 @@ loop:
 		}
 	}
 
-	watchDuration := time.Now().Sub(start)
+	watchDuration := r.clock.Now().Sub(start)
 	if watchDuration < 1*time.Second && eventCount == 0 {
 		glog.V(4).Infof("%s: Unexpected watch close - watch lasted less than a second and no items received", r.name)
 		return errors.New("very short watch")

staging/src/k8s.io/client-go/tools/cache/shared_informer_test.go

+/*
+Copyright 2017 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 cache
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/pkg/api"
+	fcache "k8s.io/client-go/tools/cache/testing"
+	"k8s.io/client-go/util/clock"
+)
+
+type testListener struct {
+	lock              sync.RWMutex
+	resyncPeriod      time.Duration
+	expectedItemNames sets.String
+	receivedItemNames []string
+	name              string
+}
+
+func newTestListener(name string, resyncPeriod time.Duration, expected ...string) *testListener {
+	l := &testListener{
+		resyncPeriod:      resyncPeriod,
+		expectedItemNames: sets.NewString(expected...),
+		name:              name,
+	}
+	return l
+}
+
+func (l *testListener) OnAdd(obj interface{}) {
+	l.handle(obj)
+}
+
+func (l *testListener) OnUpdate(old, new interface{}) {
+	l.handle(new)
+}
+
+func (l *testListener) OnDelete(obj interface{}) {
+}
+
+func (l *testListener) handle(obj interface{}) {
+	key, _ := MetaNamespaceKeyFunc(obj)
+	fmt.Printf("%s: handle: %v\n", l.name, key)
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	objectMeta, _ := meta.Accessor(obj)
+	l.receivedItemNames = append(l.receivedItemNames, objectMeta.GetName())
+}
+
+func (l *testListener) ok() bool {
+	fmt.Println("polling")
+	err := wait.PollImmediate(100*time.Millisecond, 2*time.Second, func() (bool, error) {
+		if l.satisfiedExpectations() {
+			return true, nil
+		}
+		return false, nil
+	})
+	if err != nil {
+		return false
+	}
+
+	// wait just a bit to allow any unexpected stragglers to come in
+	fmt.Println("sleeping")
+	time.Sleep(1 * time.Second)
+	fmt.Println("final check")
+	return l.satisfiedExpectations()
+}
+
+func (l *testListener) satisfiedExpectations() bool {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+
+	return len(l.receivedItemNames) == l.expectedItemNames.Len() && sets.NewString(l.receivedItemNames...).Equal(l.expectedItemNames)
+}
+
+func TestListenerResyncPeriods(t *testing.T) {
+	// source simulates an apiserver object endpoint.
+	source := fcache.NewFakeControllerSource()
+	source.Add(&api.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}})
+	source.Add(&api.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod2"}})
+
+	// create the shared informer and resync every 1s
+	informer := NewSharedInformer(source, &api.Pod{}, 1*time.Second).(*sharedIndexInformer)
+
+	clock := clock.NewFakeClock(time.Now())
+	informer.clock = clock
+	informer.processor.clock = clock
+
+	// listener 1, never resync
+	listener1 := newTestListener("listener1", 0, "pod1", "pod2")
+	informer.AddEventHandlerWithResyncPeriod(listener1, listener1.resyncPeriod)
+
+	// listener 2, resync every 2s
+	listener2 := newTestListener("listener2", 2*time.Second, "pod1", "pod2")
+	informer.AddEventHandlerWithResyncPeriod(listener2, listener2.resyncPeriod)
+
+	// listener 3, resync every 3s
+	listener3 := newTestListener("listener3", 3*time.Second, "pod1", "pod2")
+	informer.AddEventHandlerWithResyncPeriod(listener3, listener3.resyncPeriod)
+	listeners := []*testListener{listener1, listener2, listener3}
+
+	stop := make(chan struct{})
+	defer close(stop)
+
+	go informer.Run(stop)
+
+	// ensure all listeners got the initial List
+	for _, listener := range listeners {
+		if !listener.ok() {
+			t.Errorf("%s: expected %v, got %v", listener.name, listener.expectedItemNames, listener.receivedItemNames)
+		}
+	}
+
+	// reset
+	for _, listener := range listeners {
+		listener.receivedItemNames = []string{}
+	}
+
+	// advance so listener2 gets a resync
+	clock.Step(2 * time.Second)
+
+	// make sure listener2 got the resync
+	if !listener2.ok() {
+		t.Errorf("%s: expected %v, got %v", listener2.name, listener2.expectedItemNames, listener2.receivedItemNames)
+	}
+
+	// wait a bit to give errant items a chance to go to 1 and 3
+	time.Sleep(1 * time.Second)
+
+	// make sure listeners 1 and 3 got nothing
+	if len(listener1.receivedItemNames) != 0 {
+		t.Errorf("listener1: should not have resynced (got %d)", len(listener1.receivedItemNames))
+	}
+	if len(listener3.receivedItemNames) != 0 {
+		t.Errorf("listener3: should not have resynced (got %d)", len(listener3.receivedItemNames))
+	}
+
+	// reset
+	for _, listener := range listeners {
+		listener.receivedItemNames = []string{}
+	}
+
+	// advance so listener3 gets a resync
+	clock.Step(1 * time.Second)
+
+	// make sure listener3 got the resync
+	if !listener3.ok() {
+		t.Errorf("%s: expected %v, got %v", listener3.name, listener3.expectedItemNames, listener3.receivedItemNames)
+	}
+
+	// wait a bit to give errant items a chance to go to 1 and 2
+	time.Sleep(1 * time.Second)
+
+	// make sure listeners 1 and 2 got nothing
+	if len(listener1.receivedItemNames) != 0 {
+		t.Errorf("listener1: should not have resynced (got %d)", len(listener1.receivedItemNames))
+	}
+	if len(listener2.receivedItemNames) != 0 {
+		t.Errorf("listener2: should not have resynced (got %d)", len(listener2.receivedItemNames))
+	}
+}
+
+func TestResyncCheckPeriod(t *testing.T) {
+	// source simulates an apiserver object endpoint.
+	source := fcache.NewFakeControllerSource()
+
+	// create the shared informer and resync every 12 hours
+	informer := NewSharedInformer(source, &api.Pod{}, 12*time.Hour).(*sharedIndexInformer)
+
+	clock := clock.NewFakeClock(time.Now())
+	informer.clock = clock
+	informer.processor.clock = clock
+
+	// listener 1, never resync
+	listener1 := newTestListener("listener1", 0)
+	informer.AddEventHandlerWithResyncPeriod(listener1, listener1.resyncPeriod)
+	if e, a := 12*time.Hour, informer.resyncCheckPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+	if e, a := time.Duration(0), informer.processor.listeners[0].resyncPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+
+	// listener 2, resync every minute
+	listener2 := newTestListener("listener2", 1*time.Minute)
+	informer.AddEventHandlerWithResyncPeriod(listener2, listener2.resyncPeriod)
+	if e, a := 1*time.Minute, informer.resyncCheckPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+	if e, a := time.Duration(0), informer.processor.listeners[0].resyncPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+	if e, a := 1*time.Minute, informer.processor.listeners[1].resyncPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+
+	// listener 3, resync every 55 seconds
+	listener3 := newTestListener("listener3", 55*time.Second)
+	informer.AddEventHandlerWithResyncPeriod(listener3, listener3.resyncPeriod)
+	if e, a := 55*time.Second, informer.resyncCheckPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+	if e, a := time.Duration(0), informer.processor.listeners[0].resyncPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+	if e, a := 1*time.Minute, informer.processor.listeners[1].resyncPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+	if e, a := 55*time.Second, informer.processor.listeners[2].resyncPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+
+	// listener 4, resync every 5 seconds
+	listener4 := newTestListener("listener4", 5*time.Second)
+	informer.AddEventHandlerWithResyncPeriod(listener4, listener4.resyncPeriod)
+	if e, a := 5*time.Second, informer.resyncCheckPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+	if e, a := time.Duration(0), informer.processor.listeners[0].resyncPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+	if e, a := 1*time.Minute, informer.processor.listeners[1].resyncPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+	if e, a := 55*time.Second, informer.processor.listeners[2].resyncPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+	if e, a := 5*time.Second, informer.processor.listeners[3].resyncPeriod; e != a {
+		t.Errorf("expected %d, got %d", e, a)
+	}
+}

vendor/BUILD

@@ -12403,6 +12403,7 @@ go_test(
         "k8s.io/client-go/tools/cache/mutation_detector_test.go",
         "k8s.io/client-go/tools/cache/processor_listener_test.go",
         "k8s.io/client-go/tools/cache/reflector_test.go",
+        "k8s.io/client-go/tools/cache/shared_informer_test.go",
         "k8s.io/client-go/tools/cache/store_test.go",
         "k8s.io/client-go/tools/cache/undelta_store_test.go",
     ],
@@ -12410,6 +12411,7 @@ go_test(
     tags = ["automanaged"],
     deps = [
         "//vendor:github.com/google/gofuzz",
+        "//vendor:k8s.io/apimachinery/pkg/api/meta",
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
         "//vendor:k8s.io/apimachinery/pkg/runtime",
         "//vendor:k8s.io/apimachinery/pkg/util/sets",