-
Notifications
You must be signed in to change notification settings - Fork 38.6k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Avoid computing keys multiple times in Cacher #35029
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -125,6 +125,8 @@ func (i *indexedWatchers) terminateAll(objectType reflect.Type) { | |
} | ||
} | ||
|
||
type filterObjectFunc func(string, runtime.Object) bool | ||
|
||
// Cacher is responsible for serving WATCH and LIST requests for a given | ||
// resource from its internal cache and updating its cache in the background | ||
// based on the underlying storage contents. | ||
|
@@ -161,9 +163,6 @@ type Cacher struct { | |
// Versioner is used to handle resource versions. | ||
versioner Versioner | ||
|
||
// keyFunc is used to get a key in the underyling storage for a given object. | ||
keyFunc func(runtime.Object) (string, error) | ||
|
||
// triggerFunc is used for optimizing amount of watchers that needs to process | ||
// an incoming event. | ||
triggerFunc TriggerPublisherFunc | ||
|
@@ -183,7 +182,7 @@ type Cacher struct { | |
// internal cache and updating its cache in the background based on the given | ||
// configuration. | ||
func NewCacherFromConfig(config CacherConfig) *Cacher { | ||
watchCache := newWatchCache(config.CacheCapacity) | ||
watchCache := newWatchCache(config.CacheCapacity, config.KeyFunc) | ||
listerWatcher := newCacherListerWatcher(config.Storage, config.ResourcePrefix, config.NewListFunc) | ||
|
||
// Give this error when it is constructed rather than when you get the | ||
|
@@ -201,7 +200,6 @@ func NewCacherFromConfig(config CacherConfig) *Cacher { | |
watchCache: watchCache, | ||
reflector: cache.NewReflector(listerWatcher, config.Type, watchCache, 0), | ||
versioner: config.Versioner, | ||
keyFunc: config.KeyFunc, | ||
triggerFunc: config.TriggerPublisherFunc, | ||
watcherIdx: 0, | ||
watchers: indexedWatchers{ | ||
|
@@ -328,7 +326,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, resourceVersion string, | |
c.Lock() | ||
defer c.Unlock() | ||
forget := forgetWatcher(c, c.watcherIdx, triggerValue, triggerSupported) | ||
watcher := newCacheWatcher(watchRV, chanSize, initEvents, filterFunction(key, c.keyFunc, pred), forget) | ||
watcher := newCacheWatcher(watchRV, chanSize, initEvents, filterFunction(key, pred), forget) | ||
|
||
c.watchers.addWatcher(watcher, c.watcherIdx, triggerValue, triggerSupported) | ||
c.watcherIdx++ | ||
|
@@ -382,20 +380,20 @@ func (c *Cacher) List(ctx context.Context, key string, resourceVersion string, p | |
if err != nil || listVal.Kind() != reflect.Slice { | ||
return fmt.Errorf("need a pointer to slice, got %v", listVal.Kind()) | ||
} | ||
filter := filterFunction(key, c.keyFunc, pred) | ||
filter := filterFunction(key, pred) | ||
|
||
objs, readResourceVersion, err := c.watchCache.WaitUntilFreshAndList(listRV, trace) | ||
if err != nil { | ||
return fmt.Errorf("failed to wait for fresh list: %v", err) | ||
} | ||
trace.Step(fmt.Sprintf("Listed %d items from cache", len(objs))) | ||
for _, obj := range objs { | ||
object, ok := obj.(runtime.Object) | ||
elem, ok := obj.(*storeElement) | ||
if !ok { | ||
return fmt.Errorf("non runtime.Object returned from storage: %v", obj) | ||
return fmt.Errorf("non *storeElement returned from storage: %v", obj) | ||
} | ||
if filter(object) { | ||
listVal.Set(reflect.Append(listVal, reflect.ValueOf(object).Elem())) | ||
if filter(elem.Key, elem.Object) { | ||
listVal.Set(reflect.Append(listVal, reflect.ValueOf(elem.Object).Elem())) | ||
} | ||
} | ||
trace.Step(fmt.Sprintf("Filtered %d items", listVal.Len())) | ||
|
@@ -524,14 +522,9 @@ func forgetWatcher(c *Cacher, index int, triggerValue string, triggerSupported b | |
} | ||
} | ||
|
||
func filterFunction(key string, keyFunc func(runtime.Object) (string, error), p SelectionPredicate) FilterFunc { | ||
func filterFunction(key string, p SelectionPredicate) filterObjectFunc { | ||
f := SimpleFilter(p) | ||
filterFunc := func(obj runtime.Object) bool { | ||
objKey, err := keyFunc(obj) | ||
if err != nil { | ||
glog.Errorf("invalid object for filter. Obj: %v. Err: %v", obj, err) | ||
return false | ||
} | ||
filterFunc := func(objKey string, obj runtime.Object) bool { | ||
if !hasPathPrefix(objKey, key) { | ||
return false | ||
} | ||
|
@@ -626,12 +619,12 @@ type cacheWatcher struct { | |
sync.Mutex | ||
input chan watchCacheEvent | ||
result chan watch.Event | ||
filter FilterFunc | ||
filter filterObjectFunc | ||
stopped bool | ||
forget func(bool) | ||
} | ||
|
||
func newCacheWatcher(resourceVersion uint64, chanSize int, initEvents []watchCacheEvent, filter FilterFunc, forget func(bool)) *cacheWatcher { | ||
func newCacheWatcher(resourceVersion uint64, chanSize int, initEvents []watchCacheEvent, filter filterObjectFunc, forget func(bool)) *cacheWatcher { | ||
watcher := &cacheWatcher{ | ||
input: make(chan watchCacheEvent, chanSize), | ||
result: make(chan watch.Event, chanSize), | ||
|
@@ -707,11 +700,12 @@ func (c *cacheWatcher) add(event *watchCacheEvent) { | |
} | ||
} | ||
|
||
func (c *cacheWatcher) sendWatchCacheEvent(event watchCacheEvent) { | ||
curObjPasses := event.Type != watch.Deleted && c.filter(event.Object) | ||
// NOTE: sendWatchCacheEvent is assumed to not modify <event> !!! | ||
func (c *cacheWatcher) sendWatchCacheEvent(event *watchCacheEvent) { | ||
curObjPasses := event.Type != watch.Deleted && c.filter(event.Key, event.Object) | ||
oldObjPasses := false | ||
if event.PrevObject != nil { | ||
oldObjPasses = c.filter(event.PrevObject) | ||
oldObjPasses = c.filter(event.Key, event.PrevObject) | ||
} | ||
if !curObjPasses && !oldObjPasses { | ||
// Watcher is not interested in that object. | ||
|
@@ -752,7 +746,7 @@ func (c *cacheWatcher) process(initEvents []watchCacheEvent, resourceVersion uin | |
const initProcessThreshold = 500 * time.Millisecond | ||
startTime := time.Now() | ||
for _, event := range initEvents { | ||
c.sendWatchCacheEvent(event) | ||
c.sendWatchCacheEvent(&event) | ||
} | ||
processingTime := time.Since(startTime) | ||
if processingTime > initProcessThreshold { | ||
|
@@ -772,7 +766,7 @@ func (c *cacheWatcher) process(initEvents []watchCacheEvent, resourceVersion uin | |
} | ||
// only send events newer than resourceVersion | ||
if event.ResourceVersion > resourceVersion { | ||
c.sendWatchCacheEvent(event) | ||
c.sendWatchCacheEvent(&event) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. same question here, what happens if a watcher mutates an event (or object in an event)? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. As above - added comment to "sendWatchCacheEvent" method. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ok, just realized sendWatchCacheEvent copies the object and constructs a watch.Event per watcher, so that's fine |
||
} | ||
} | ||
} | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
does this depend on all watchers treating the incoming event as immutable?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes (and this is true assumption today). I added explicit comment for that method.