Update k8s.io dependencies to master to get selflink fix in client-go

This commit is contained in:
Michelle Au
2019-08-12 13:48:24 -07:00
parent 7e09056156
commit d38cdc2f76
315 changed files with 86399 additions and 37559 deletions

View File

@@ -79,6 +79,7 @@ type controller struct {
clock clock.Clock
}
// Controller is a generic controller framework.
type Controller interface {
Run(stopCh <-chan struct{})
HasSynced() bool
@@ -149,7 +150,7 @@ func (c *controller) processLoop() {
for {
obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
if err != nil {
if err == FIFOClosedError {
if err == ErrFIFOClosed {
return
}
if c.config.RetryOnError {

View File

@@ -160,7 +160,7 @@ func (f *DeltaFIFO) KeyOf(obj interface{}) (string, error) {
return f.keyFunc(obj)
}
// Return true if an Add/Update/Delete/AddIfNotPresent are called first,
// HasSynced returns true if an Add/Update/Delete/AddIfNotPresent are called first,
// or an Update called first but the first batch of items inserted by Replace() has been popped
func (f *DeltaFIFO) HasSynced() bool {
f.lock.Lock()
@@ -389,7 +389,7 @@ func (f *DeltaFIFO) GetByKey(key string) (item interface{}, exists bool, err err
return d, exists, nil
}
// Checks if the queue is closed
// IsClosed checks if the queue is closed
func (f *DeltaFIFO) IsClosed() bool {
f.closedLock.Lock()
defer f.closedLock.Unlock()
@@ -417,7 +417,7 @@ func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
// When Close() is called, the f.closed is set and the condition is broadcasted.
// Which causes this loop to continue and return from the Pop().
if f.IsClosed() {
return nil, FIFOClosedError
return nil, ErrFIFOClosed
}
f.cond.Wait()
@@ -593,6 +593,7 @@ type KeyGetter interface {
// DeltaType is the type of a change (addition, deletion, etc)
type DeltaType string
// Change type definition
const (
Added DeltaType = "Added"
Updated DeltaType = "Updated"

View File

@@ -48,14 +48,14 @@ type ExpirationCache struct {
// ExpirationPolicy dictates when an object expires. Currently only abstracted out
// so unittests don't rely on the system clock.
type ExpirationPolicy interface {
IsExpired(obj *timestampedEntry) bool
IsExpired(obj *TimestampedEntry) bool
}
// TTLPolicy implements a ttl based ExpirationPolicy.
type TTLPolicy struct {
// >0: Expire entries with an age > ttl
// <=0: Don't expire any entry
Ttl time.Duration
TTL time.Duration
// Clock used to calculate ttl expiration
Clock clock.Clock
@@ -63,26 +63,30 @@ type TTLPolicy struct {
// IsExpired returns true if the given object is older than the ttl, or it can't
// determine its age.
func (p *TTLPolicy) IsExpired(obj *timestampedEntry) bool {
return p.Ttl > 0 && p.Clock.Since(obj.timestamp) > p.Ttl
func (p *TTLPolicy) IsExpired(obj *TimestampedEntry) bool {
return p.TTL > 0 && p.Clock.Since(obj.Timestamp) > p.TTL
}
// timestampedEntry is the only type allowed in a ExpirationCache.
type timestampedEntry struct {
obj interface{}
timestamp time.Time
// TimestampedEntry is the only type allowed in a ExpirationCache.
// Keep in mind that it is not safe to share timestamps between computers.
// Behavior may be inconsistent if you get a timestamp from the API Server and
// use it on the client machine as part of your ExpirationCache.
type TimestampedEntry struct {
Obj interface{}
Timestamp time.Time
key string
}
// getTimestampedEntry returns the timestampedEntry stored under the given key.
func (c *ExpirationCache) getTimestampedEntry(key string) (*timestampedEntry, bool) {
// getTimestampedEntry returns the TimestampedEntry stored under the given key.
func (c *ExpirationCache) getTimestampedEntry(key string) (*TimestampedEntry, bool) {
item, _ := c.cacheStorage.Get(key)
if tsEntry, ok := item.(*timestampedEntry); ok {
if tsEntry, ok := item.(*TimestampedEntry); ok {
return tsEntry, true
}
return nil, false
}
// getOrExpire retrieves the object from the timestampedEntry if and only if it hasn't
// getOrExpire retrieves the object from the TimestampedEntry if and only if it hasn't
// already expired. It holds a write lock across deletion.
func (c *ExpirationCache) getOrExpire(key string) (interface{}, bool) {
// Prevent all inserts from the time we deem an item as "expired" to when we
@@ -95,11 +99,11 @@ func (c *ExpirationCache) getOrExpire(key string) (interface{}, bool) {
return nil, false
}
if c.expirationPolicy.IsExpired(timestampedItem) {
klog.V(4).Infof("Entry %v: %+v has expired", key, timestampedItem.obj)
klog.V(4).Infof("Entry %v: %+v has expired", key, timestampedItem.Obj)
c.cacheStorage.Delete(key)
return nil, false
}
return timestampedItem.obj, true
return timestampedItem.Obj, true
}
// GetByKey returns the item stored under the key, or sets exists=false.
@@ -126,10 +130,8 @@ func (c *ExpirationCache) List() []interface{} {
list := make([]interface{}, 0, len(items))
for _, item := range items {
obj := item.(*timestampedEntry).obj
if key, err := c.keyFunc(obj); err != nil {
list = append(list, obj)
} else if obj, exists := c.getOrExpire(key); exists {
key := item.(*TimestampedEntry).key
if obj, exists := c.getOrExpire(key); exists {
list = append(list, obj)
}
}
@@ -151,7 +153,7 @@ func (c *ExpirationCache) Add(obj interface{}) error {
c.expirationLock.Lock()
defer c.expirationLock.Unlock()
c.cacheStorage.Add(key, &timestampedEntry{obj, c.clock.Now()})
c.cacheStorage.Add(key, &TimestampedEntry{obj, c.clock.Now(), key})
return nil
}
@@ -184,7 +186,7 @@ func (c *ExpirationCache) Replace(list []interface{}, resourceVersion string) er
if err != nil {
return KeyError{item, err}
}
items[key] = &timestampedEntry{item, ts}
items[key] = &TimestampedEntry{item, ts, key}
}
c.expirationLock.Lock()
defer c.expirationLock.Unlock()
@@ -199,10 +201,15 @@ func (c *ExpirationCache) Resync() error {
// NewTTLStore creates and returns a ExpirationCache with a TTLPolicy
func NewTTLStore(keyFunc KeyFunc, ttl time.Duration) Store {
return NewExpirationStore(keyFunc, &TTLPolicy{ttl, clock.RealClock{}})
}
// NewExpirationStore creates and returns a ExpirationCache for a given policy
func NewExpirationStore(keyFunc KeyFunc, expirationPolicy ExpirationPolicy) Store {
return &ExpirationCache{
cacheStorage: NewThreadSafeStore(Indexers{}, Indices{}),
keyFunc: keyFunc,
clock: clock.RealClock{},
expirationPolicy: &TTLPolicy{ttl, clock.RealClock{}},
expirationPolicy: expirationPolicy,
}
}

View File

@@ -33,16 +33,19 @@ func (c *fakeThreadSafeMap) Delete(key string) {
}
}
// FakeExpirationPolicy keeps the list for keys which never expires.
type FakeExpirationPolicy struct {
NeverExpire sets.String
RetrieveKeyFunc KeyFunc
}
func (p *FakeExpirationPolicy) IsExpired(obj *timestampedEntry) bool {
// IsExpired used to check if object is expired.
func (p *FakeExpirationPolicy) IsExpired(obj *TimestampedEntry) bool {
key, _ := p.RetrieveKeyFunc(obj)
return !p.NeverExpire.Has(key)
}
// NewFakeExpirationStore creates a new instance for the ExpirationCache.
func NewFakeExpirationStore(keyFunc KeyFunc, deletedKeys chan<- string, expirationPolicy ExpirationPolicy, cacheClock clock.Clock) Store {
cacheStorage := NewThreadSafeStore(Indexers{}, Indices{})
return &ExpirationCache{

View File

@@ -16,7 +16,7 @@ limitations under the License.
package cache
// FakeStore lets you define custom functions for store operations
// FakeCustomStore lets you define custom functions for store operations.
type FakeCustomStore struct {
AddFunc func(obj interface{}) error
UpdateFunc func(obj interface{}) error
@@ -25,7 +25,7 @@ type FakeCustomStore struct {
ListKeysFunc func() []string
GetFunc func(obj interface{}) (item interface{}, exists bool, err error)
GetByKeyFunc func(key string) (item interface{}, exists bool, err error)
ReplaceFunc func(list []interface{}, resourceVerion string) error
ReplaceFunc func(list []interface{}, resourceVersion string) error
ResyncFunc func() error
}

View File

@@ -34,7 +34,8 @@ type ErrRequeue struct {
Err error
}
var FIFOClosedError error = errors.New("DeltaFIFO: manipulating with closed queue")
// ErrFIFOClosed used when FIFO is closed
var ErrFIFOClosed = errors.New("DeltaFIFO: manipulating with closed queue")
func (e ErrRequeue) Error() string {
if e.Err == nil {
@@ -66,7 +67,7 @@ type Queue interface {
Close()
}
// Helper function for popping from Queue.
// Pop is helper function for popping from Queue.
// WARNING: Do NOT use this function in non-test code to avoid races
// unless you really really really really know what you are doing.
func Pop(queue Queue) interface{} {
@@ -126,7 +127,7 @@ func (f *FIFO) Close() {
f.cond.Broadcast()
}
// Return true if an Add/Update/Delete/AddIfNotPresent are called first,
// HasSynced returns true if an Add/Update/Delete/AddIfNotPresent are called first,
// or an Update called first but the first batch of items inserted by Replace() has been popped
func (f *FIFO) HasSynced() bool {
f.lock.Lock()
@@ -242,7 +243,7 @@ func (f *FIFO) GetByKey(key string) (item interface{}, exists bool, err error) {
return item, exists, nil
}
// Checks if the queue is closed
// IsClosed checks if the queue is closed
func (f *FIFO) IsClosed() bool {
f.closedLock.Lock()
defer f.closedLock.Unlock()
@@ -267,7 +268,7 @@ func (f *FIFO) Pop(process PopProcessFunc) (interface{}, error) {
// When Close() is called, the f.closed is set and the condition is broadcasted.
// Which causes this loop to continue and return from the Pop().
if f.IsClosed() {
return nil, FIFOClosedError
return nil, ErrFIFOClosed
}
f.cond.Wait()

View File

@@ -28,7 +28,9 @@ const (
closedMsg = "heap is closed"
)
// LessFunc is used to compare two objects in the heap.
type LessFunc func(interface{}, interface{}) bool
type heapItem struct {
obj interface{} // The object which is stored in the heap.
index int // The index of the object's key in the Heap.queue.
@@ -158,7 +160,7 @@ func (h *Heap) Add(obj interface{}) error {
return nil
}
// Adds all the items in the list to the queue and then signals the condition
// BulkAdd adds all the items in the list to the queue and then signals the condition
// variable. It is useful when the caller would like to add all of the items
// to the queue before consumer starts processing them.
func (h *Heap) BulkAdd(list []interface{}) error {
@@ -249,11 +251,11 @@ func (h *Heap) Pop() (interface{}, error) {
h.cond.Wait()
}
obj := heap.Pop(h.data)
if obj != nil {
return obj, nil
} else {
if obj == nil {
return nil, fmt.Errorf("object was removed from heap data")
}
return obj, nil
}
// List returns a list of all the items.

View File

@@ -23,17 +23,27 @@ import (
"k8s.io/apimachinery/pkg/util/sets"
)
// Indexer is a storage interface that lets you list objects using multiple indexing functions
// Indexer is a storage interface that lets you list objects using multiple indexing functions.
// There are three kinds of strings here.
// One is a storage key, as defined in the Store interface.
// Another kind is a name of an index.
// The third kind of string is an "indexed value", which is produced by an
// IndexFunc and can be a field value or any other string computed from the object.
type Indexer interface {
Store
// Retrieve list of objects that match on the named indexing function
// Index returns the stored objects whose set of indexed values
// intersects the set of indexed values of the given object, for
// the named index
Index(indexName string, obj interface{}) ([]interface{}, error)
// IndexKeys returns the set of keys that match on the named indexing function.
IndexKeys(indexName, indexKey string) ([]string, error)
// ListIndexFuncValues returns the list of generated values of an Index func
// IndexKeys returns the storage keys of the stored objects whose
// set of indexed values for the named index includes the given
// indexed value
IndexKeys(indexName, indexedValue string) ([]string, error)
// ListIndexFuncValues returns all the indexed values of the given index
ListIndexFuncValues(indexName string) []string
// ByIndex lists object that match on the named indexing function with the exact key
ByIndex(indexName, indexKey string) ([]interface{}, error)
// ByIndex returns the stored objects whose set of indexed values
// for the named index includes the given indexed value
ByIndex(indexName, indexedValue string) ([]interface{}, error)
// GetIndexer return the indexers
GetIndexers() Indexers
@@ -42,7 +52,7 @@ type Indexer interface {
AddIndexers(newIndexers Indexers) error
}
// IndexFunc knows how to provide an indexed value for an object.
// IndexFunc knows how to compute the set of indexed values for an object.
type IndexFunc func(obj interface{}) ([]string, error)
// IndexFuncToKeyFuncAdapter adapts an indexFunc to a keyFunc. This is only useful if your index function returns
@@ -65,6 +75,7 @@ func IndexFuncToKeyFuncAdapter(indexFunc IndexFunc) KeyFunc {
}
const (
// NamespaceIndex is the lookup name for the most comment index function, which is to index by the namespace field.
NamespaceIndex string = "namespace"
)

View File

@@ -30,6 +30,7 @@ import (
// AppendFunc is used to add a matching item to whatever list the caller is using
type AppendFunc func(interface{})
// ListAll calls appendFn with each value retrieved from store which matches the selector.
func ListAll(store Store, selector labels.Selector, appendFn AppendFunc) error {
selectAll := selector.Empty()
for _, m := range store.List() {
@@ -50,6 +51,7 @@ func ListAll(store Store, selector labels.Selector, appendFn AppendFunc) error {
return nil
}
// ListAllByNamespace used to list items belongs to namespace from Indexer.
func ListAllByNamespace(indexer Indexer, namespace string, selector labels.Selector, appendFn AppendFunc) error {
selectAll := selector.Empty()
if namespace == metav1.NamespaceAll {
@@ -124,6 +126,7 @@ type GenericNamespaceLister interface {
Get(name string) (runtime.Object, error)
}
// NewGenericLister creates a new instance for the genericLister.
func NewGenericLister(indexer Indexer, resource schema.GroupResource) GenericLister {
return &genericLister{indexer: indexer, resource: resource}
}

View File

@@ -42,6 +42,7 @@ type MutationCache interface {
Mutation(interface{})
}
// ResourceVersionComparator is able to compare object versions.
type ResourceVersionComparator interface {
CompareResourceVersion(lhs, rhs runtime.Object) int
}

View File

@@ -36,12 +36,14 @@ func init() {
mutationDetectionEnabled, _ = strconv.ParseBool(os.Getenv("KUBE_CACHE_MUTATION_DETECTOR"))
}
type CacheMutationDetector interface {
// MutationDetector is able to monitor if the object be modified outside.
type MutationDetector interface {
AddObject(obj interface{})
Run(stopCh <-chan struct{})
}
func NewCacheMutationDetector(name string) CacheMutationDetector {
// NewCacheMutationDetector creates a new instance for the defaultCacheMutationDetector.
func NewCacheMutationDetector(name string) MutationDetector {
if !mutationDetectionEnabled {
return dummyMutationDetector{}
}
@@ -114,7 +116,7 @@ func (d *defaultCacheMutationDetector) CompareObjects() {
altered := false
for i, obj := range d.cachedObjs {
if !reflect.DeepEqual(obj.cached, obj.copied) {
fmt.Printf("CACHE %s[%d] ALTERED!\n%v\n", d.name, i, diff.ObjectDiff(obj.cached, obj.copied))
fmt.Printf("CACHE %s[%d] ALTERED!\n%v\n", d.name, i, diff.ObjectGoPrintSideBySide(obj.cached, obj.copied))
altered = true
}
}

View File

@@ -17,6 +17,7 @@ limitations under the License.
package cache
import (
"context"
"errors"
"fmt"
"io"
@@ -24,7 +25,6 @@ import (
"net"
"net/url"
"reflect"
"strings"
"sync"
"syscall"
"time"
@@ -38,6 +38,7 @@ import (
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/tools/pager"
"k8s.io/klog"
"k8s.io/utils/trace"
)
@@ -68,6 +69,9 @@ type Reflector struct {
lastSyncResourceVersion string
// lastSyncResourceVersionMutex guards read/write access to lastSyncResourceVersion
lastSyncResourceVersionMutex sync.RWMutex
// WatchListPageSize is the requested chunk size of initial and resync watch lists.
// Defaults to pager.PageSize.
WatchListPageSize int64
}
var (
@@ -79,7 +83,7 @@ var (
// NewNamespaceKeyedIndexerAndReflector creates an Indexer and a Reflector
// The indexer is configured to key on namespace
func NewNamespaceKeyedIndexerAndReflector(lw ListerWatcher, expectedType interface{}, resyncPeriod time.Duration) (indexer Indexer, reflector *Reflector) {
indexer = NewIndexer(MetaNamespaceKeyFunc, Indexers{"namespace": MetaNamespaceIndexFunc})
indexer = NewIndexer(MetaNamespaceKeyFunc, Indexers{NamespaceIndex: MetaNamespaceIndexFunc})
reflector = NewReflector(lw, expectedType, indexer, resyncPeriod)
return indexer, reflector
}
@@ -108,11 +112,6 @@ func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{},
return r
}
func makeValidPrometheusMetricLabel(in string) string {
// this isn't perfect, but it removes our common characters
return strings.NewReplacer("/", "_", ".", "_", "-", "_", ":", "_").Replace(in)
}
// internalPackages are packages that ignored when creating a default reflector name. These packages are in the common
// call chains to NewReflector, so they'd be low entropy names for reflectors
var internalPackages = []string{"client-go/tools/cache/"}
@@ -167,7 +166,7 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
options := metav1.ListOptions{ResourceVersion: "0"}
if err := func() error {
initTrace := trace.New("Reflector " + r.name + " ListAndWatch")
initTrace := trace.New("Reflector ListAndWatch", trace.Field{"name", r.name})
defer initTrace.LogIfLong(10 * time.Second)
var list runtime.Object
var err error
@@ -179,7 +178,16 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
panicCh <- r
}
}()
list, err = r.listerWatcher.List(options)
// Attempt to gather list in chunks, if supported by listerWatcher, if not, the first
// list request will return the full response.
pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
return r.listerWatcher.List(opts)
}))
if r.WatchListPageSize != 0 {
pager.PageSize = r.WatchListPageSize
}
// Pager falls back to full list if paginated list calls fail due to an "Expired" error.
list, err = pager.List(context.Background(), options)
close(listCh)
}()
select {
@@ -257,6 +265,10 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
// We want to avoid situations of hanging watchers. Stop any wachers that do not
// receive any events within the timeout window.
TimeoutSeconds: &timeoutSeconds,
// To reduce load on kube-apiserver on watch restarts, you may enable watch bookmarks.
// Reflector doesn't assume bookmarks are returned at all (if the server do not support
// watch bookmarks, it will ignore this field).
AllowWatchBookmarks: true,
}
w, err := r.listerWatcher.Watch(options)
@@ -286,7 +298,12 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
if err := r.watchHandler(w, &resourceVersion, resyncerrc, stopCh); err != nil {
if err != errorStopRequested {
klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedType, err)
switch {
case apierrs.IsResourceExpired(err):
klog.V(4).Infof("%s: watch of %v ended with: %v", r.name, r.expectedType, err)
default:
klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedType, err)
}
}
return nil
}
@@ -354,6 +371,8 @@ loop:
if err != nil {
utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", r.name, event.Object, err))
}
case watch.Bookmark:
// A `Bookmark` means watch has synced here, just update the resourceVersion
default:
utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
}
@@ -363,7 +382,7 @@ loop:
}
}
watchDuration := r.clock.Now().Sub(start)
watchDuration := r.clock.Since(start)
if watchDuration < 1*time.Second && eventCount == 0 {
return fmt.Errorf("very short watch: %s: Unexpected watch close - watch lasted less than a second and no items received", r.name)
}

View File

@@ -94,23 +94,6 @@ var metricsFactory = struct {
metricsProvider: noopMetricsProvider{},
}
func newReflectorMetrics(name string) *reflectorMetrics {
var ret *reflectorMetrics
if len(name) == 0 {
return ret
}
return &reflectorMetrics{
numberOfLists: metricsFactory.metricsProvider.NewListsMetric(name),
listDuration: metricsFactory.metricsProvider.NewListDurationMetric(name),
numberOfItemsInList: metricsFactory.metricsProvider.NewItemsInListMetric(name),
numberOfWatches: metricsFactory.metricsProvider.NewWatchesMetric(name),
numberOfShortWatches: metricsFactory.metricsProvider.NewShortWatchesMetric(name),
watchDuration: metricsFactory.metricsProvider.NewWatchDurationMetric(name),
numberOfItemsInWatch: metricsFactory.metricsProvider.NewItemsInWatchMetric(name),
lastResourceVersion: metricsFactory.metricsProvider.NewLastResourceVersionMetric(name),
}
}
// SetReflectorMetricsProvider sets the metrics provider
func SetReflectorMetricsProvider(metricsProvider MetricsProvider) {
metricsFactory.setProviders.Do(func() {

View File

@@ -31,31 +31,84 @@ import (
"k8s.io/klog"
)
// SharedInformer has a shared data cache and is capable of distributing notifications for changes
// to the cache to multiple listeners who registered via AddEventHandler. If you use this, there is
// one behavior change compared to a standard Informer. When you receive a notification, the cache
// will be AT LEAST as fresh as the notification, but it MAY be more fresh. You should NOT depend
// on the contents of the cache exactly matching the notification you've received in handler
// functions. If there was a create, followed by a delete, the cache may NOT have your item. This
// has advantages over the broadcaster since it allows us to share a common cache across many
// controllers. Extending the broadcaster would have required us keep duplicate caches for each
// watch.
// SharedInformer provides eventually consistent linkage of its
// clients to the authoritative state of a given collection of
// objects. An object is identified by its API group, kind/resource,
// namespace, and name. One SharedInformer provides linkage to objects
// of a particular API group and kind/resource. The linked object
// collection of a SharedInformer may be further restricted to one
// namespace and/or by label selector and/or field selector.
//
// The authoritative state of an object is what apiservers provide
// access to, and an object goes through a strict sequence of states.
// A state is either "absent" or present with a ResourceVersion and
// other appropriate content.
//
// A SharedInformer maintains a local cache, exposed by GetStore(), of
// the state of each relevant object. This cache is eventually
// consistent with the authoritative state. This means that, unless
// prevented by persistent communication problems, if ever a
// particular object ID X is authoritatively associated with a state S
// then for every SharedInformer I whose collection includes (X, S)
// eventually either (1) I's cache associates X with S or a later
// state of X, (2) I is stopped, or (3) the authoritative state
// service for X terminates. To be formally complete, we say that the
// absent state meets any restriction by label selector or field
// selector.
//
// As a simple example, if a collection of objects is henceforeth
// unchanging and a SharedInformer is created that links to that
// collection then that SharedInformer's cache eventually holds an
// exact copy of that collection (unless it is stopped too soon, the
// authoritative state service ends, or communication problems between
// the two persistently thwart achievement).
//
// As another simple example, if the local cache ever holds a
// non-absent state for some object ID and the object is eventually
// removed from the authoritative state then eventually the object is
// removed from the local cache (unless the SharedInformer is stopped
// too soon, the authoritative state service ends, or communication
// problems persistently thwart the desired result).
//
// The keys in GetStore() are of the form namespace/name for namespaced
// objects, and are simply the name for non-namespaced objects.
//
// A client is identified here by a ResourceEventHandler. For every
// update to the SharedInformer's local cache and for every client,
// eventually either the SharedInformer is stopped or the client is
// notified of the update. These notifications happen after the
// corresponding cache update and, in the case of a
// SharedIndexInformer, after the corresponding index updates. It is
// possible that additional cache and index updates happen before such
// a prescribed notification. For a given SharedInformer and client,
// all notifications are delivered sequentially. For a given
// SharedInformer, client, and object ID, the notifications are
// delivered in order.
//
// A delete notification exposes the last locally known non-absent
// state, except that its ResourceVersion is replaced with a
// ResourceVersion in which the object is actually absent.
type SharedInformer interface {
// AddEventHandler adds an event handler to the shared informer using the shared informer's resync
// period. Events to a single handler are delivered sequentially, but there is no coordination
// between different handlers.
AddEventHandler(handler ResourceEventHandler)
// AddEventHandlerWithResyncPeriod adds an event handler to the shared informer using the
// specified resync period. Events to a single handler are delivered sequentially, but there is
// no coordination between different handlers.
// AddEventHandlerWithResyncPeriod adds an event handler to the
// shared informer using the specified resync period. The resync
// operation consists of delivering to the handler a create
// notification for every object in the informer's local cache; it
// does not add any interactions with the authoritative storage.
AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration)
// GetStore returns the Store.
// GetStore returns the informer's local cache as a Store.
GetStore() Store
// GetController gives back a synthetic interface that "votes" to start the informer
GetController() Controller
// Run starts the shared informer, which will be stopped when stopCh is closed.
// Run starts and runs the shared informer, returning after it stops.
// The informer will be stopped when stopCh is closed.
Run(stopCh <-chan struct{})
// HasSynced returns true if the shared informer's store has synced.
// HasSynced returns true if the shared informer's store has been
// informed by at least one full LIST of the authoritative state
// of the informer's object collection. This is unrelated to "resync".
HasSynced() bool
// LastSyncResourceVersion is the resource version observed when last synced with the underlying
// store. The value returned is not synchronized with access to the underlying store and is not
@@ -63,6 +116,7 @@ type SharedInformer interface {
LastSyncResourceVersion() string
}
// SharedIndexInformer provides add and get Indexers ability based on SharedInformer.
type SharedIndexInformer interface {
SharedInformer
// AddIndexers add indexers to the informer before it starts.
@@ -129,7 +183,7 @@ type sharedIndexInformer struct {
controller Controller
processor *sharedProcessor
cacheMutationDetector CacheMutationDetector
cacheMutationDetector MutationDetector
// This block is tracked to handle late initialization of the controller
listerWatcher ListerWatcher
@@ -169,7 +223,7 @@ func (v *dummyController) HasSynced() bool {
return v.informer.HasSynced()
}
func (c *dummyController) LastSyncResourceVersion() string {
func (v *dummyController) LastSyncResourceVersion() string {
return ""
}
@@ -555,7 +609,7 @@ func (p *processorListener) run() {
case deleteNotification:
p.handler.OnDelete(notification.oldObj)
default:
utilruntime.HandleError(fmt.Errorf("unrecognized notification: %#v", next))
utilruntime.HandleError(fmt.Errorf("unrecognized notification: %T", next))
}
}
// the only way to get here is if the p.nextCh is empty and closed

0
vendor/k8s.io/client-go/tools/cache/store.go generated vendored Executable file → Normal file
View File

View File

@@ -185,7 +185,7 @@ func (c *threadSafeMap) ByIndex(indexName, indexKey string) ([]interface{}, erro
set := index[indexKey]
list := make([]interface{}, 0, set.Len())
for _, key := range set.List() {
for key := range set {
list = append(list, c.items[key])
}
@@ -302,6 +302,7 @@ func (c *threadSafeMap) Resync() error {
return nil
}
// NewThreadSafeStore creates a new instance of ThreadSafeStore.
func NewThreadSafeStore(indexers Indexers, indices Indices) ThreadSafeStore {
return &threadSafeMap{
items: map[string]interface{}{},

View File

@@ -31,6 +31,7 @@ type UndeltaStore struct {
// Assert that it implements the Store interface.
var _ Store = &UndeltaStore{}
// Add inserts an object into the store and sends complete state by calling PushFunc.
// Note about thread safety. The Store implementation (cache.cache) uses a lock for all methods.
// In the functions below, the lock gets released and reacquired betweend the {Add,Delete,etc}
// and the List. So, the following can happen, resulting in two identical calls to PushFunc.
@@ -41,7 +42,6 @@ var _ Store = &UndeltaStore{}
// 3 Store.Add(b)
// 4 Store.List() -> [a,b]
// 5 Store.List() -> [a,b]
func (u *UndeltaStore) Add(obj interface{}) error {
if err := u.Store.Add(obj); err != nil {
return err
@@ -50,6 +50,7 @@ func (u *UndeltaStore) Add(obj interface{}) error {
return nil
}
// Update sets an item in the cache to its updated state and sends complete state by calling PushFunc.
func (u *UndeltaStore) Update(obj interface{}) error {
if err := u.Store.Update(obj); err != nil {
return err
@@ -58,6 +59,7 @@ func (u *UndeltaStore) Update(obj interface{}) error {
return nil
}
// Delete removes an item from the cache and sends complete state by calling PushFunc.
func (u *UndeltaStore) Delete(obj interface{}) error {
if err := u.Store.Delete(obj); err != nil {
return err
@@ -66,6 +68,10 @@ func (u *UndeltaStore) Delete(obj interface{}) error {
return nil
}
// Replace will delete the contents of current store, using instead the given list.
// 'u' takes ownership of the list, you should not reference the list again
// after calling this function.
// The new contents complete state will be sent by calling PushFunc after replacement.
func (u *UndeltaStore) Replace(list []interface{}, resourceVersion string) error {
if err := u.Store.Replace(list, resourceVersion); err != nil {
return err