var emptyFunc … var coreNamespaceResource … const storageWatchListPageSize … const defaultBookmarkFrequency … type Config … type watchersMap … func (wm watchersMap) addWatcher(w *cacheWatcher, number int) { … } func (wm watchersMap) deleteWatcher(number int) { … } func (wm watchersMap) terminateAll(done func(*cacheWatcher)) { … } type indexedWatchers … func (i *indexedWatchers) addWatcher(w *cacheWatcher, number int, scope namespacedName, value string, supported bool) { … } func (i *indexedWatchers) deleteWatcher(number int, scope namespacedName, value string, supported bool) { … } func (i *indexedWatchers) terminateAll(groupResource schema.GroupResource, done func(*cacheWatcher)) { … } type watcherBookmarkTimeBuckets … func newTimeBucketWatchers(clock clock.Clock, bookmarkFrequency time.Duration) *watcherBookmarkTimeBuckets { … } // adds a watcher to the bucket, if the deadline is before the start, it will be // added to the first one. func (t *watcherBookmarkTimeBuckets) addWatcherThreadUnsafe(w *cacheWatcher) bool { … } func (t *watcherBookmarkTimeBuckets) popExpiredWatchersThreadUnsafe() [][]*cacheWatcher { … } type filterWithAttrsFunc … type indexedTriggerFunc … type Cacher … func (c *Cacher) RequestWatchProgress(ctx context.Context) error { … } // NewCacherFromConfig creates a new Cacher responsible for servicing WATCH and LIST requests from // its internal cache and updating its cache in the background based on the // given configuration. func NewCacherFromConfig(config Config) (*Cacher, error) { … } func (c *Cacher) startCaching(stopChannel <-chan struct{ … } // Versioner implements storage.Interface. func (c *Cacher) Versioner() storage.Versioner { … } // Create implements storage.Interface. func (c *Cacher) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error { … } // Delete implements storage.Interface. func (c *Cacher) Delete( ctx context.Context, key string, out runtime.Object, preconditions *storage.Preconditions, validateDeletion storage.ValidateObjectFunc, _ runtime.Object, opts storage.DeleteOptions) error { … } type namespacedName … // Watch implements storage.Interface. func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) { … } // Get implements storage.Interface. func (c *Cacher) Get(ctx context.Context, key string, opts storage.GetOptions, objPtr runtime.Object) error { … } // NOTICE: Keep in sync with shouldListFromStorage function in // // staging/src/k8s.io/apiserver/pkg/util/flowcontrol/request/list_work_estimator.go func shouldDelegateList(opts storage.ListOptions) bool { … } // computeListLimit determines whether the cacher should // apply a limit to an incoming LIST request and returns its value. // // note that this function doesn't check RVM nor the Continuation token. // these parameters are validated by the shouldDelegateList function. // // as of today, the limit is ignored for requests that set RV == 0 func computeListLimit(opts storage.ListOptions) int64 { … } func shouldDelegateListOnNotReadyCache(opts storage.ListOptions) bool { … } func (c *Cacher) listItems(ctx context.Context, listRV uint64, key string, pred storage.SelectionPredicate, recursive bool) ([]interface{ … } // GetList implements storage.Interface func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error { … } // GuaranteedUpdate implements storage.Interface. func (c *Cacher) GuaranteedUpdate( ctx context.Context, key string, destination runtime.Object, ignoreNotFound bool, preconditions *storage.Preconditions, tryUpdate storage.UpdateFunc, _ runtime.Object) error { … } // Count implements storage.Interface. func (c *Cacher) Count(pathPrefix string) (int64, error) { … } // ReadinessCheck implements storage.Interface. func (c *Cacher) ReadinessCheck() error { … } // baseObjectThreadUnsafe omits locking for cachingObject. func baseObjectThreadUnsafe(object runtime.Object) runtime.Object { … } func (c *Cacher) triggerValuesThreadUnsafe(event *watchCacheEvent) ([]string, bool) { … } func (c *Cacher) processEvent(event *watchCacheEvent) { … } func (c *Cacher) dispatchEvents() { … } func setCachingObjects(event *watchCacheEvent, versioner storage.Versioner) { … } func (c *Cacher) dispatchEvent(event *watchCacheEvent) { … } func (c *Cacher) startDispatchingBookmarkEventsLocked() { … } // startDispatching chooses watchers potentially interested in a given event // a marks dispatching as true. func (c *Cacher) startDispatching(event *watchCacheEvent) { … } // finishDispatching stops all the watchers that were supposed to be // stopped in the meantime, but it was deferred to avoid closing input // channels of watchers, as add() may still have writing to it. // It also marks dispatching as false. func (c *Cacher) finishDispatching() { … } func (c *Cacher) terminateAllWatchers() { … } func (c *Cacher) stopWatcherLocked(watcher *cacheWatcher) { … } func (c *Cacher) isStopped() bool { … } // Stop implements the graceful termination. func (c *Cacher) Stop() { … } func forgetWatcher(c *Cacher, w *cacheWatcher, index int, scope namespacedName, triggerValue string, triggerSupported bool) func(bool) { … } func filterWithAttrsAndPrefixFunction(key string, p storage.SelectionPredicate) filterWithAttrsFunc { … } // LastSyncResourceVersion returns resource version to which the underlying cache is synced. func (c *Cacher) LastSyncResourceVersion() (uint64, error) { … } // getBookmarkAfterResourceVersionLockedFunc returns a function that // spits a ResourceVersion after which the bookmark event will be delivered. // // The returned function must be called under the watchCache lock. func (c *Cacher) getBookmarkAfterResourceVersionLockedFunc(parsedResourceVersion, requiredResourceVersion uint64, opts storage.ListOptions) (func() uint64, error) { … } func isListWatchRequest(opts storage.ListOptions) bool { … } // getWatchCacheResourceVersion returns a ResourceVersion to which the watch cache must be synchronized to // // Depending on the input parameters, the semantics of the returned ResourceVersion are: // - must be at Exact RV (when parsedWatchResourceVersion > 0) // - can be at Any RV (when parsedWatchResourceVersion = 0) // - must be at Most Recent RV (return an RV from etcd) // // note that the above semantic is enforced by the API validation (defined elsewhere): // // if SendInitiaEvents != nil => ResourceVersionMatch = NotOlderThan // if ResourceVersionmatch != nil => ResourceVersionMatch = NotOlderThan & SendInitialEvents != nil func (c *Cacher) getWatchCacheResourceVersion(ctx context.Context, parsedWatchResourceVersion uint64, opts storage.ListOptions) (uint64, error) { … } // waitUntilWatchCacheFreshAndForceAllEvents waits until cache is at least // as fresh as given requestedWatchRV if sendInitialEvents was requested. // otherwise, we allow for establishing the connection because the clients // can wait for events without unnecessary blocking. func (c *Cacher) waitUntilWatchCacheFreshAndForceAllEvents(ctx context.Context, requestedWatchRV uint64, opts storage.ListOptions) error { … } // Wait blocks until the cacher is Ready or Stopped, it returns an error if Stopped. func (c *Cacher) Wait(ctx context.Context) error { … } // setInitialEventsEndBookmarkIfRequested sets initialEventsEndBookmark field in watchCacheInterval for watchlist request func (c *Cacher) setInitialEventsEndBookmarkIfRequested(cacheInterval *watchCacheInterval, opts storage.ListOptions, currentResourceVersion uint64) { … } type errWatcher … func newErrWatcher(err error) *errWatcher { … } // Implements watch.Interface. func (c *errWatcher) ResultChan() <-chan watch.Event { … } // Implements watch.Interface. func (c *errWatcher) Stop() { … } type immediateCloseWatcher … func newImmediateCloseWatcher() *immediateCloseWatcher { … } // Implements watch.Interface. func (c *immediateCloseWatcher) ResultChan() <-chan watch.Event { … } // Implements watch.Interface. func (c *immediateCloseWatcher) Stop() { … }
Codebase Browser
kubernetes