watchcache

Documentation

Index

• func GetKeyFunc(gvr schema.GroupVersionResource, isNamespaced bool) keyFunc
• type CacheWatcher
  • func NewCacheWatcher(chanSize int) *CacheWatcher
  • func (c *CacheWatcher) Add(event *watch.Event, timer *time.Timer) bool
  • func (c *CacheWatcher) NonblockingAdd(event *watch.Event) bool
  • func (c *CacheWatcher) Process(ctx context.Context, initEvents []*watch.Event)
  • func (c *CacheWatcher) ResultChan() <-chan watch.Event
  • func (c *CacheWatcher) Stop()
  • func (c *CacheWatcher) StopThreadUnsafe()
• type ClusterResourceVersion
  • func GetClusterResourceVersionFromEvent(event *watch.Event) (*ClusterResourceVersion, error)
  • func NewClusterResourceVersion(cluster string) *ClusterResourceVersion
  • func NewClusterResourceVersionFromString(rv string) (*ClusterResourceVersion, error)
  • func (crv *ClusterResourceVersion) GetClusterResourceVersion() string
  • func (crv *ClusterResourceVersion) IsEmpty() bool
  • func (crv *ClusterResourceVersion) IsEqual(another *ClusterResourceVersion) bool
• type ClusterResourceVersionSynchro
  • func NewClusterResourceVersionSynchro(cluster string) *ClusterResourceVersionSynchro
  • func (crvs *ClusterResourceVersionSynchro) RemoveCluster(clusterName string)
  • func (crvs *ClusterResourceVersionSynchro) SetClusterResourceVersion(clusterName string, resourceVersion string)
  • func (crvs *ClusterResourceVersionSynchro) UpdateClusterResourceVersion(obj runtime.Object, cluster string) (*ClusterResourceVersion, error)
• type StoreElement
• type WatchCache
  • func NewWatchCache(capacity int, gvr schema.GroupVersionResource, isNamespaced bool) *WatchCache
  • func (w *WatchCache) Add(obj runtime.Object, clusterName string, ...) error
  • func (w *WatchCache) AddIndexer(clusterName string, indexers *cache.Indexers)
  • func (w *WatchCache) CleanCluster(cluster string)
  • func (w *WatchCache) Delete(obj runtime.Object, clusterName string, ...) error
  • func (w *WatchCache) DeleteIndexer(clusterName string) bool
  • func (w *WatchCache) GetAllEventsSinceThreadUnsafe(resourceVersion *ClusterResourceVersion) ([]*watch.Event, error)
  • func (w *WatchCache) GetStores() map[string]cache.Indexer
  • func (w *WatchCache) Update(obj runtime.Object, clusterName string, ...) error
  • func (w *WatchCache) WaitUntilFreshAndGet(cluster, namespace, name string) (*StoreElement, error)
  • func (w *WatchCache) WaitUntilFreshAndList(opts *internal.ListOptions) ([]*StoreElement, *ClusterResourceVersion, error)

Functions

func GetKeyFunc

func GetKeyFunc(gvr schema.GroupVersionResource, isNamespaced bool) keyFunc

Types

type CacheWatcher

type CacheWatcher struct {
	// contains filtered or unexported fields
}

CacheWatcher implements watch.Interface

func NewCacheWatcher

func NewCacheWatcher(chanSize int) *CacheWatcher

func (*CacheWatcher) Add

func (c *CacheWatcher) Add(event *watch.Event, timer *time.Timer) bool

Nil timer means that add will not block (if it can't send event immediately, it will break the watcher)

func (*CacheWatcher) NonblockingAdd

func (c *CacheWatcher) NonblockingAdd(event *watch.Event) bool

func (*CacheWatcher) Process

func (c *CacheWatcher) Process(ctx context.Context, initEvents []*watch.Event)

Process send the events which stored in watchCache into the result channel,and select the event from input channel into result channel continuously.

func (*CacheWatcher) ResultChan

func (c *CacheWatcher) ResultChan() <-chan watch.Event

ResultChan implements watch.Interface.

func (*CacheWatcher) Stop

func (c *CacheWatcher) Stop()

Stop implements watch.Interface.

func (*CacheWatcher) StopThreadUnsafe

func (c *CacheWatcher) StopThreadUnsafe()

type ClusterResourceVersion

type ClusterResourceVersion struct {
	// contains filtered or unexported fields
}

ClusterResourceVersion holds the pediaCluster name and its latest resourceVersion

func GetClusterResourceVersionFromEvent

func GetClusterResourceVersionFromEvent(event *watch.Event) (*ClusterResourceVersion, error)

GetClusterResourceVersionFromEvent return a ClusterResourceVersion from watch event

func NewClusterResourceVersion

func NewClusterResourceVersion(cluster string) *ClusterResourceVersion

func NewClusterResourceVersionFromString

func NewClusterResourceVersionFromString(rv string) (*ClusterResourceVersion, error)

func (*ClusterResourceVersion) GetClusterResourceVersion

func (crv *ClusterResourceVersion) GetClusterResourceVersion() string

GetClusterResourceVersion return a base64 encode string of ClusterResourceVersion

func (*ClusterResourceVersion) IsEmpty

func (crv *ClusterResourceVersion) IsEmpty() bool

func (*ClusterResourceVersion) IsEqual

func (crv *ClusterResourceVersion) IsEqual(another *ClusterResourceVersion) bool

type ClusterResourceVersionSynchro

type ClusterResourceVersionSynchro struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewClusterResourceVersionSynchro

func NewClusterResourceVersionSynchro(cluster string) *ClusterResourceVersionSynchro

func (*ClusterResourceVersionSynchro) RemoveCluster

func (crvs *ClusterResourceVersionSynchro) RemoveCluster(clusterName string)

func (*ClusterResourceVersionSynchro) SetClusterResourceVersion

func (crvs *ClusterResourceVersionSynchro) SetClusterResourceVersion(clusterName string, resourceVersion string)

func (*ClusterResourceVersionSynchro) UpdateClusterResourceVersion

func (crvs *ClusterResourceVersionSynchro) UpdateClusterResourceVersion(obj runtime.Object, cluster string) (*ClusterResourceVersion, error)

UpdateClusterResourceVersion update the resourceVersion in ClusterResourceVersionSynchro to the latest

type StoreElement

type StoreElement struct {
	Key    string
	Object runtime.Object
	Labels labels.Set
	Fields fields.Set
}

StoreElement keeping the structs of resource in k8s(key, object, labels, fields).

type WatchCache

type WatchCache struct {
	sync.RWMutex

	// KeyFunc is used to get a key in the underlying storage for a given object.
	KeyFunc func(runtime.Object) (string, error)

	WatchersLock sync.RWMutex

	// watchersBuffer is a list of watchers potentially interested in currently
	// dispatched event.
	WatchersBuffer []*CacheWatcher
	// blockedWatchers is a list of watchers whose buffer is currently full.
	BlockedWatchers []*CacheWatcher
	IsNamespaced    bool
	// contains filtered or unexported fields
}

WatchCache implements a Store interface. However, it depends on the elements implementing runtime.Object interface.

WatchCache is a "sliding window" (with a limited capacity) of objects observed from a watch.

func NewWatchCache

func NewWatchCache(capacity int, gvr schema.GroupVersionResource, isNamespaced bool) *WatchCache

func (*WatchCache) Add

func (w *WatchCache) Add(obj runtime.Object, clusterName string, resourceVersion *ClusterResourceVersion,
	codec runtime.Codec, memoryVersion schema.GroupVersion) error

Add takes runtime.Object as an argument.

func (*WatchCache) AddIndexer

func (w *WatchCache) AddIndexer(clusterName string, indexers *cache.Indexers)

func (*WatchCache) CleanCluster

func (w *WatchCache) CleanCluster(cluster string)

func (*WatchCache) Delete

func (w *WatchCache) Delete(obj runtime.Object, clusterName string, resourceVersion *ClusterResourceVersion,
	codec runtime.Codec, memoryVersion schema.GroupVersion) error

Delete takes runtime.Object as an argument.

func (*WatchCache) DeleteIndexer

func (w *WatchCache) DeleteIndexer(clusterName string) bool

func (*WatchCache) GetAllEventsSinceThreadUnsafe

func (w *WatchCache) GetAllEventsSinceThreadUnsafe(resourceVersion *ClusterResourceVersion) ([]*watch.Event, error)

GetAllEventsSinceThreadUnsafe returns watch event from slice window in watchCache by the resourceVersion

func (*WatchCache) GetStores

func (w *WatchCache) GetStores() map[string]cache.Indexer

func (*WatchCache) Update

func (w *WatchCache) Update(obj runtime.Object, clusterName string, resourceVersion *ClusterResourceVersion,
	codec runtime.Codec, memoryVersion schema.GroupVersion) error

Update takes runtime.Object as an argument.

func (*WatchCache) WaitUntilFreshAndGet

func (w *WatchCache) WaitUntilFreshAndGet(cluster, namespace, name string) (*StoreElement, error)

WaitUntilFreshAndGet returns list of pointers to <storeElement> objects.

func (*WatchCache) WaitUntilFreshAndList

func (w *WatchCache) WaitUntilFreshAndList(opts *internal.ListOptions) ([]*StoreElement, *ClusterResourceVersion, error)

WaitUntilFreshAndList returns list of pointers to <storeElement> objects.