Commit 9df41301 authored by derekwaynecarr's avatar derekwaynecarr

Expose last sync resource version

parent c4cbc9bc
...@@ -51,6 +51,10 @@ type Reflector struct { ...@@ -51,6 +51,10 @@ type Reflector struct {
// the beginning of the next one. // the beginning of the next one.
period time.Duration period time.Duration
resyncPeriod time.Duration resyncPeriod time.Duration
// lastSyncResourceVersion is the resource version token last
// observed when doing a sync with the underlying store
// it is not thread safe as it is not synchronized with access to the store
lastSyncResourceVersion string
} }
// NewNamespaceKeyedIndexerAndReflector creates an Indexer and a Reflector // NewNamespaceKeyedIndexerAndReflector creates an Indexer and a Reflector
...@@ -130,6 +134,7 @@ func (r *Reflector) listAndWatch() { ...@@ -130,6 +134,7 @@ func (r *Reflector) listAndWatch() {
glog.Errorf("Unable to sync list result: %v", err) glog.Errorf("Unable to sync list result: %v", err)
return return
} }
r.lastSyncResourceVersion = resourceVersion
for { for {
w, err := r.listerWatcher.Watch(resourceVersion) w, err := r.listerWatcher.Watch(resourceVersion)
...@@ -203,6 +208,7 @@ loop: ...@@ -203,6 +208,7 @@ loop:
glog.Errorf("unable to understand watch event %#v", event) glog.Errorf("unable to understand watch event %#v", event)
} }
*resourceVersion = meta.ResourceVersion() *resourceVersion = meta.ResourceVersion()
r.lastSyncResourceVersion = *resourceVersion
eventCount++ eventCount++
} }
} }
...@@ -215,3 +221,9 @@ loop: ...@@ -215,3 +221,9 @@ loop:
glog.V(4).Infof("Watch close - %v total %v items received", r.expectedType, eventCount) glog.V(4).Infof("Watch close - %v total %v items received", r.expectedType, eventCount)
return nil return nil
} }
// LastSyncResourceVersion is the resource version observed when last sync with the underlying store
// The value returned is not synchronized with access to the underlying store and is not thread-safe
func (r *Reflector) LastSyncResourceVersion() string {
return r.lastSyncResourceVersion
}
...@@ -112,6 +112,11 @@ func TestReflector_watchHandler(t *testing.T) { ...@@ -112,6 +112,11 @@ func TestReflector_watchHandler(t *testing.T) {
if e, a := "32", resumeRV; e != a { if e, a := "32", resumeRV; e != a {
t.Errorf("expected %v, got %v", e, a) t.Errorf("expected %v, got %v", e, a)
} }
// last sync resource version should be the last version synced with store
if e, a := "32", g.LastSyncResourceVersion(); e != a {
t.Errorf("expected %v, got %v", e, a)
}
} }
func TestReflector_watchHandlerTimeout(t *testing.T) { func TestReflector_watchHandlerTimeout(t *testing.T) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment