const blockTimeout … const resourceVersionTooHighRetrySeconds … const eventFreshDuration … const defaultLowerBoundCapacity … const defaultUpperBoundCapacity … type watchCacheEvent … type watchCache … func newWatchCache( keyFunc func(runtime.Object) (string, error), eventHandler func(*watchCacheEvent), getAttrsFunc func(runtime.Object) (labels.Set, fields.Set, error), versioner storage.Versioner, indexers *cache.Indexers, clock clock.WithTicker, groupResource schema.GroupResource, progressRequester *conditionalProgressRequester) *watchCache { … } // Add takes runtime.Object as an argument. func (w *watchCache) Add(obj interface{ … } // Update takes runtime.Object as an argument. func (w *watchCache) Update(obj interface{ … } // Delete takes runtime.Object as an argument. func (w *watchCache) Delete(obj interface{ … } func (w *watchCache) objectToVersionedRuntimeObject(obj interface{ … } // processEvent is safe as long as there is at most one call to it in flight // at any point in time. func (w *watchCache) processEvent(event watch.Event, resourceVersion uint64, updateFunc func(*storeElement) error) error { … } // Assumes that lock is already held for write. func (w *watchCache) updateCache(event *watchCacheEvent) { … } // resizeCacheLocked resizes the cache if necessary: // - increases capacity by 2x if cache is full and all cached events occurred within last eventFreshDuration. // - decreases capacity by 2x when recent quarter of events occurred outside of eventFreshDuration(protect watchCache from flapping). func (w *watchCache) resizeCacheLocked(eventTime time.Time) { … } // isCacheFullLocked used to judge whether watchCacheEvent is full. // Assumes that lock is already held for write. func (w *watchCache) isCacheFullLocked() bool { … } // doCacheResizeLocked resize watchCache's event array with different capacity. // Assumes that lock is already held for write. func (w *watchCache) doCacheResizeLocked(capacity int) { … } func (w *watchCache) UpdateResourceVersion(resourceVersion string) { … } // List returns list of pointers to <storeElement> objects. func (w *watchCache) List() []interface{ … } // waitUntilFreshAndBlock waits until cache is at least as fresh as given <resourceVersion>. // NOTE: This function acquired lock and doesn't release it. // You HAVE TO explicitly call w.RUnlock() after this function. func (w *watchCache) waitUntilFreshAndBlock(ctx context.Context, resourceVersion uint64) error { … } type sortableStoreElements … func (s sortableStoreElements) Len() int { … } func (s sortableStoreElements) Less(i, j int) bool { … } func (s sortableStoreElements) Swap(i, j int) { … } // WaitUntilFreshAndList returns list of pointers to `storeElement` objects along // with their ResourceVersion and the name of the index, if any, that was used. func (w *watchCache) WaitUntilFreshAndList(ctx context.Context, resourceVersion uint64, key string, matchValues []storage.MatchValue) ([]interface{ … } func (w *watchCache) waitUntilFreshAndListItems(ctx context.Context, resourceVersion uint64, key string, matchValues []storage.MatchValue) (result []interface{ … } func filterPrefix(prefix string, items []interface{ … } func (w *watchCache) notFresh(resourceVersion uint64) bool { … } // WaitUntilFreshAndGet returns a pointers to <storeElement> object. func (w *watchCache) WaitUntilFreshAndGet(ctx context.Context, resourceVersion uint64, key string) (interface{ … } func (w *watchCache) ListKeys() []string { … } // Get takes runtime.Object as a parameter. However, it returns // pointer to <storeElement>. func (w *watchCache) Get(obj interface{ … } // GetByKey returns pointer to <storeElement>. func (w *watchCache) GetByKey(key string) (interface{ … } // Replace takes slice of runtime.Object as a parameter. func (w *watchCache) Replace(objs []interface{ … } func (w *watchCache) SetOnReplace(onReplace func()) { … } func (w *watchCache) Resync() error { … } func (w *watchCache) getListResourceVersion() uint64 { … } func (w *watchCache) currentCapacity() int { … } const minWatchChanSize … const maxWatchChanSizeWithIndexAndTrigger … const maxWatchChanSizeWithIndexWithoutTrigger … const maxWatchChanSizeWithoutIndex … func (w *watchCache) suggestedWatchChannelSize(indexExists, triggerUsed bool) int { … } // isIndexValidLocked checks if a given index is still valid. // This assumes that the lock is held. func (w *watchCache) isIndexValidLocked(index int) bool { … } // getAllEventsSinceLocked returns a watchCacheInterval that can be used to // retrieve events since a certain resourceVersion. This function assumes to // be called under the watchCache lock. func (w *watchCache) getAllEventsSinceLocked(resourceVersion uint64, key string, opts storage.ListOptions) (*watchCacheInterval, error) { … } // getIntervalFromStoreLocked returns a watchCacheInterval // that covers the entire storage state. // This function assumes to be called under the watchCache lock. func (w *watchCache) getIntervalFromStoreLocked(key string, matchesSingle bool) (*watchCacheInterval, error) { … }
Codebase Browser
kubernetes