1. const (
    2.     // 主要是若给定版本的obj不在cache，需要去etcd fetch数据
    3.     // 此参数代表了fetch的超时时间
    4.     blockTimeout = 3 * time.Second
    6.     // 客户端收到too high resource version异常需要先block的时长，然后再重拾
    7.     resourceVersionTooHighRetrySeconds = 1
    8. )
    10. // 为了避免多次计算obj的key/label/field，需要缓存这些值
    11. type storeElement struct {
    12.     Key    string
    13.     Object runtime.Object
    14.     Labels labels.Set
    15.     Fields fields.Set
    16. }
    18. func storeElementKey(obj interface{}) (string, error) {
    19.     elem, ok := obj.(*storeElement)
    20.     if !ok {
    21.         return "", fmt.Errorf("not a storeElement: %v", obj)
    22.     }
    23.     return elem.Key, nil
    24. }
    26. func storeElementObject(obj interface{}) (runtime.Object, error) {
    27.     elem, ok := obj.(*storeElement)
    28.     if !ok {
    29.         return nil, fmt.Errorf("not a storeElement: %v", obj)
    30.     }
    31.     return elem.Object, nil
    32. }
    34. func storeElementIndexFunc(objIndexFunc cache.IndexFunc) cache.IndexFunc {
    35.     return func(obj interface{}) (strings []string, e error) {
    36.         seo, err := storeElementObject(obj)
    37.         if err != nil {
    38.             return nil, err
    39.         }
    40.         return objIndexFunc(seo)
    41.     }
    42. }
    44. func storeElementIndexers(indexers *cache.Indexers) cache.Indexers {
    45.     if indexers == nil {
    46.         return cache.Indexers{}
    47.     }
    48.     ret := cache.Indexers{}
    49.     for indexName, indexFunc := range *indexers {
    50.         ret[indexName] = storeElementIndexFunc(indexFunc)
    51.     }
    52.     return ret
    53. }
    55. // watchCache实际上是一个滑动窗口
    56. // 维护固定数量的obj
    57. type watchCache struct {
    58.     sync.RWMutex
    60.     // 等待刷新足够的版本数据到cache
    61.     cond *sync.Cond
    62.     // 滑动窗口的容量
    63.     capacity     int
    64.     keyFunc      func(runtime.Object) (string, error)
    65.     getAttrsFunc func(runtime.Object) (labels.Set, fields.Set, error)
    67.     // cache is used a cyclic buffer - its first element (with the smallest
    68.     // resourceVersion) is defined by startIndex, its last element is defined
    69.     // by endIndex (if cache is full it will be startIndex + capacity).
    70.     // Both startIndex and endIndex can be greater than buffer capacity -
    71.     // you should always apply modulo capacity to get an index in cache array.
    72.     cache      []*watchCacheEvent
    73.     startIndex int
    74.     endIndex   int
    76.     // store will effectively support LIST operation from the "end of cache
    77.     // history" i.e. from the moment just after the newest cached watched event.
    78.     // It is necessary to effectively allow clients to start watching at now.
    79.     // NOTE: We assume that <store> is thread-safe.
    80.     store cache.Indexer
    82.     // ResourceVersion up to which the watchCache is propagated.
    83.     resourceVersion uint64
    85.     // ResourceVersion of the last list result (populated via Replace() method).
    86.     listResourceVersion uint64
    88.     // This handler is run at the end of every successful Replace() method.
    89.     onReplace func()
    90.     // 事件处理器
    91.     eventHandler func(*watchCacheEvent)
    92.     clock        clock.Clock
    93.     // 获取obj的resourceversion信息
    94.     versioner storage.Versioner
    95. }
    97. type watchCacheEvent struct {
    98.     Key             string
    99.     ResourceVersion uint64
    100.     Type            watch.EventType
    101.     Object          runtime.Object
    102.     ObjLabels       labels.Set
    103.     ObjFields       fields.Set
    104.     PrevObject      runtime.Object
    105.     PrevObjLabels   labels.Set
    106.     PrevObjFields   fields.Set
    107. }
    109. func newWatchCache(
    110.     capacity int,
    111.     keyFunc func(runtime.Object) (string, error),
    112.     eventHandler func(*watchCacheEvent),
    113.     getAttrsFunc func(runtime.Object) (labels.Set, fields.Set, error),
    114.     versioner storage.Versioner,
    115.     indexers *cache.Indexers) *watchCache {
    116.     wc := &watchCache{
    117.         capacity:            capacity,
    118.         keyFunc:             keyFunc,
    119.         getAttrsFunc:        getAttrsFunc,
    120.         cache:               make([]*watchCacheEvent, capacity),
    121.         startIndex:          0,
    122.         endIndex:            0,
    123.         store:               cache.NewIndexer(storeElementKey, storeElementIndexers(indexers)),
    124.         resourceVersion:     0,
    125.         listResourceVersion: 0,
    126.         eventHandler:        eventHandler,
    127.         clock:               clock.RealClock{},
    128.         versioner:           versioner,
    129.     }
    130.     wc.cond = sync.NewCond(wc.RLocker())
    131.     return wc
    132. }
    134. func (w *watchCache) Add(obj interface{}) error {
    135.     object, resourceVersion, err := w.objectToVersionedRuntimeObject(obj)
    136.     if err != nil {
    137.         return err
    138.     }
    139.     event := watch.Event{Type: watch.Added, Object: object}
    140.     f := func(elem *storeElement) error { return w.store.Add(elem) }
    141.     return w.processEvent(event, resourceVersion, f)
    142. }
    144. func (w *watchCache) Update(obj interface{}) error {
    145.     object, resourceVersion, err := w.objectToVersionedRuntimeObject(obj)
    146.     if err != nil {
    147.         return err
    148.     }
    149.     event := watch.Event{Type: watch.Modified, Object: object}
    150.     f := func(elem *storeElement) error { return w.store.Update(elem) }
    151.     return w.processEvent(event, resourceVersion, f)
    152. }
    154. func (w *watchCache) Delete(obj interface{}) error {
    155.     object, resourceVersion, err := w.objectToVersionedRuntimeObject(obj)
    156.     if err != nil {
    157.         return err
    158.     }
    159.     event := watch.Event{Type: watch.Deleted, Object: object}
    160.     f := func(elem *storeElement) error { return w.store.Delete(elem) }
    161.     return w.processEvent(event, resourceVersion, f)
    162. }
    164. func (w *watchCache) List() []interface{} {
    165.     return w.store.List()
    166. }
    168. func (w *watchCache) ListKeys() []string {
    169.     return w.store.ListKeys()
    170. }
    172. func (w *watchCache) WaitUntilFreshAndList(resourceVersion uint64, matchValues []storage.MatchValue, trace *utiltrace.Trace) ([]interface{}, uint64, error) {
    173.     err := w.waitUntilFreshAndBlock(resourceVersion, trace)
    174.     defer w.RUnlock()
    175.     if err != nil {
    176.         return nil, 0, err
    177.     }
    179.     // This isn't the place where we do "final filtering" - only some "prefiltering" is happening here. So the only
    180.     // requirement here is to NOT miss anything that should be returned. We can return as many non-matching items as we
    181.     // want - they will be filtered out later. The fact that we return less things is only further performance improvement.
    182.     // TODO: if multiple indexes match, return the one with the fewest items, so as to do as much filtering as possible.
    183.     for _, matchValue := range matchValues {
    184.         if result, err := w.store.ByIndex(matchValue.IndexName, matchValue.Value); err == nil {
    185.             return result, w.resourceVersion, nil
    186.         }
    187.     }
    188.     return w.store.List(), w.resourceVersion, nil
    189. }
    191. func (w *watchCache) WaitUntilFreshAndGet(resourceVersion uint64, key string, trace *utiltrace.Trace) (interface{}, bool, uint64, error) {
    192.     err := w.waitUntilFreshAndBlock(resourceVersion, trace)
    193.     defer w.RUnlock()
    194.     if err != nil {
    195.         return nil, false, 0, err
    196.     }
    197.     value, exists, err := w.store.GetByKey(key)
    198.     return value, exists, w.resourceVersion, err
    199. }
    201. func (w *watchCache) Get(obj interface{}) (interface{}, bool, error) {
    202.     object, ok := obj.(runtime.Object)
    203.     if !ok {
    204.         return nil, false, fmt.Errorf("obj does not implement runtime.Object interface: %v", obj)
    205.     }
    206.     key, err := w.keyFunc(object)
    207.     if err != nil {
    208.         return nil, false, fmt.Errorf("couldn't compute key: %v", err)
    209.     }
    211.     return w.store.Get(&storeElement{Key: key, Object: object})
    212. }
    214. func (w *watchCache) GetByKey(key string) (interface{}, bool, error) {
    215.     return w.store.GetByKey(key)
    216. }
    218. func (w *watchCache) Replace(objs []interface{}, resourceVersion string) error {
    219.     version, err := w.versioner.ParseResourceVersion(resourceVersion)
    220.     if err != nil {
    221.         return err
    222.     }
    224.     toReplace := make([]interface{}, 0, len(objs))
    225.     for _, obj := range objs {
    226.         object, ok := obj.(runtime.Object)
    227.         if !ok {
    228.             return fmt.Errorf("didn't get runtime.Object for replace: %#v", obj)
    229.         }
    230.         key, err := w.keyFunc(object)
    231.         if err != nil {
    232.             return fmt.Errorf("couldn't compute key: %v", err)
    233.         }
    234.         objLabels, objFields, err := w.getAttrsFunc(object)
    235.         if err != nil {
    236.             return err
    237.         }
    238.         toReplace = append(toReplace, &storeElement{
    239.             Key:    key,
    240.             Object: object,
    241.             Labels: objLabels,
    242.             Fields: objFields,
    243.         })
    244.     }
    246.     w.Lock()
    247.     defer w.Unlock()
    249.     w.startIndex = 0
    250.     w.endIndex = 0
    251.     if err := w.store.Replace(toReplace, resourceVersion); err != nil {
    252.         return err
    253.     }
    254.     w.listResourceVersion = version
    255.     w.resourceVersion = version
    256.     if w.onReplace != nil {
    257.         w.onReplace()
    258.     }
    259.     w.cond.Broadcast()
    260.     return nil
    261. }
    263. func (w *watchCache) SetOnReplace(onReplace func()) {
    264.     w.Lock()
    265.     defer w.Unlock()
    266.     w.onReplace = onReplace
    267. }
    269. func (w *watchCache) GetAllEventsSinceThreadUnsafe(resourceVersion uint64) ([]*watchCacheEvent, error) {
    270.     size := w.endIndex - w.startIndex
    271.     var oldest uint64
    272.     switch {
    273.     case size >= w.capacity:
    274.         // Once the watch event buffer is full, the oldest watch event we can deliver
    275.         // is the first one in the buffer.
    276.         oldest = w.cache[w.startIndex%w.capacity].ResourceVersion
    277.     case w.listResourceVersion > 0:
    278.         // If the watch event buffer isn't full, the oldest watch event we can deliver
    279.         // is one greater than the resource version of the last full list.
    280.         oldest = w.listResourceVersion + 1
    281.     case size > 0:
    282.         // If we've never completed a list, use the resourceVersion of the oldest event
    283.         // in the buffer.
    284.         // This should only happen in unit tests that populate the buffer without
    285.         // performing list/replace operations.
    286.         oldest = w.cache[w.startIndex%w.capacity].ResourceVersion
    287.     default:
    288.         return nil, fmt.Errorf("watch cache isn't correctly initialized")
    289.     }
    291.     if resourceVersion == 0 {
    292.         // resourceVersion = 0 means that we don't require any specific starting point
    293.         // and we would like to start watching from ~now.
    294.         // However, to keep backward compatibility, we additionally need to return the
    295.         // current state and only then start watching from that point.
    296.         //
    297.         // TODO: In v2 api, we should stop returning the current state - #13969.
    298.         allItems := w.store.List()
    299.         result := make([]*watchCacheEvent, len(allItems))
    300.         for i, item := range allItems {
    301.             elem, ok := item.(*storeElement)
    302.             if !ok {
    303.                 return nil, fmt.Errorf("not a storeElement: %v", elem)
    304.             }
    305.             objLabels, objFields, err := w.getAttrsFunc(elem.Object)
    306.             if err != nil {
    307.                 return nil, err
    308.             }
    309.             result[i] = &watchCacheEvent{
    310.                 Type:            watch.Added,
    311.                 Object:          elem.Object,
    312.                 ObjLabels:       objLabels,
    313.                 ObjFields:       objFields,
    314.                 Key:             elem.Key,
    315.                 ResourceVersion: w.resourceVersion,
    316.             }
    317.         }
    318.         return result, nil
    319.     }
    320.     if resourceVersion < oldest-1 {
    321.         return nil, errors.NewResourceExpired(fmt.Sprintf("too old resource version: %d (%d)", resourceVersion, oldest-1))
    322.     }
    324.     // Binary search the smallest index at which resourceVersion is greater than the given one.
    325.     f := func(i int) bool {
    326.         return w.cache[(w.startIndex+i)%w.capacity].ResourceVersion > resourceVersion
    327.     }
    328.     first := sort.Search(size, f)
    329.     result := make([]*watchCacheEvent, size-first)
    330.     for i := 0; i < size-first; i++ {
    331.         result[i] = w.cache[(w.startIndex+first+i)%w.capacity]
    332.     }
    333.     return result, nil
    334. }
    336. func (w *watchCache) GetAllEventsSince(resourceVersion uint64) ([]*watchCacheEvent, error) {
    337.     w.RLock()
    338.     defer w.RUnlock()
    339.     return w.GetAllEventsSinceThreadUnsafe(resourceVersion)
    340. }
    342. func (w *watchCache) Resync() error {
    343.     return nil
    344. }
    346. func (w *watchCache) processEvent(event watch.Event, resourceVersion uint64, updateFunc func(*storeElement) error) error {
    347.     key, err := w.keyFunc(event.Object)
    348.     if err != nil {
    349.         return fmt.Errorf("couldn't compute key: %v", err)
    350.     }
    351.     elem := &storeElement{Key: key, Object: event.Object}
    352.     elem.Labels, elem.Fields, err = w.getAttrsFunc(event.Object)
    353.     if err != nil {
    354.         return err
    355.     }
    357.     wcEvent := &watchCacheEvent{
    358.         Type:            event.Type,
    359.         Object:          elem.Object,
    360.         ObjLabels:       elem.Labels,
    361.         ObjFields:       elem.Fields,
    362.         Key:             key,
    363.         ResourceVersion: resourceVersion,
    364.     }
    366.     if err := func() error {
    367.         // TODO: We should consider moving this lock below after the watchCacheEvent
    368.         // is created. In such situation, the only problematic scenario is Replace(
    369.         // happening after getting object from store and before acquiring a lock.
    370.         // Maybe introduce another lock for this purpose.
    371.         w.Lock()
    372.         defer w.Unlock()
    374.         previous, exists, err := w.store.Get(elem)
    375.         if err != nil {
    376.             return err
    377.         }
    378.         if exists {
    379.             previousElem := previous.(*storeElement)
    380.             wcEvent.PrevObject = previousElem.Object
    381.             wcEvent.PrevObjLabels = previousElem.Labels
    382.             wcEvent.PrevObjFields = previousElem.Fields
    383.         }
    385.         w.updateCache(wcEvent)
    386.         w.resourceVersion = resourceVersion
    387.         defer w.cond.Broadcast()
    389.         return updateFunc(elem)
    390.     }(); err != nil {
    391.         return err
    392.     }
    394.     // Avoid calling event handler under lock.
    395.     // This is safe as long as there is at most one call to processEvent in flight
    396.     // at any point in time.
    397.     if w.eventHandler != nil {
    398.         w.eventHandler(wcEvent)
    399.     }
    400.     return nil
    401. }
    403. func (w *watchCache) updateCache(event *watchCacheEvent) {
    404.     if w.endIndex == w.startIndex+w.capacity {
    405.         // Cache is full - remove the oldest element.
    406.         w.startIndex++
    407.     }
    408.     w.cache[w.endIndex%w.capacity] = event
    409.     w.endIndex++
    410. }
    412. func (w *watchCache) waitUntilFreshAndBlock(resourceVersion uint64, trace *utiltrace.Trace) error {
    413.     startTime := w.clock.Now()
    414.     go func() {
    415.         // Wake us up when the time limit has expired.  The docs
    416.         // promise that time.After (well, NewTimer, which it calls)
    417.         // will wait *at least* the duration given. Since this go
    418.         // routine starts sometime after we record the start time, and
    419.         // it will wake up the loop below sometime after the broadcast,
    420.         // we don't need to worry about waking it up before the time
    421.         // has expired accidentally.
    422.         <-w.clock.After(blockTimeout)
    423.         w.cond.Broadcast()
    424.     }()
    426.     w.RLock()
    427.     if trace != nil {
    428.         trace.Step("watchCache locked acquired")
    429.     }
    430.     for w.resourceVersion < resourceVersion {
    431.         if w.clock.Since(startTime) >= blockTimeout {
    432.             // Request that the client retry after 'resourceVersionTooHighRetrySeconds' seconds.
    433.             return storage.NewTooLargeResourceVersionError(resourceVersion, w.resourceVersion, resourceVersionTooHighRetrySeconds)
    434.         }
    435.         w.cond.Wait()
    436.     }
    437.     if trace != nil {
    438.         trace.Step("watchCache fresh enough")
    439.     }
    440.     return nil
    441. }
    443. // 解析obj为runtime.Object、version
    444. func (w *watchCache) objectToVersionedRuntimeObject(obj interface{}) (runtime.Object, uint64, error) {
    445.     object, ok := obj.(runtime.Object)
    446.     if !ok {
    447.         return nil, 0, fmt.Errorf("obj does not implement runtime.Object interface: %v", obj)
    448.     }
    449.     resourceVersion, err := w.versioner.ObjectResourceVersion(object)
    450.     if err != nil {
    451.         return nil, 0, err
    452.     }
    453.     return object, resourceVersion, nil
    454. }