- 1. 介绍
- 2.Delta
- 3. FIFO
- 4. DeltaFIFO
Reflector中通过ListAndWatch获取到数据后传入到本地的存储中，也就是DeltaFIFO中.从DeltaFIFO的名字可以看出它是一个FIFO,也就是一个先进先出的队列，而Delta表示的是变化的资源对象存储，包含操作资源对象的类型和数据，Reflector就是这个队列的生产者。
2.Delta在了解DeltaFIFO之前我们需要了解下什么事Delta,我们先来看看client-go中是如何定义的，Delta的数据结构定义位于：staging/src/k8s.io/client-go/tools/cache/delta_fifo.go文件中：
staging/src/k8s.io/client-go/tools/cache/delta_fifo.go文
// 变化类型（增加、删除、更新等）
type DeltaType string
// 变化类型的定义
const (
Added DeltaType = "Added" //增加
Updated DeltaType = "Updated" //删除
Deleted DeltaType = "Deleted" //更新
// 当遇到watch错误，不得不进行重新lists时，就会触发Replaced。我们不知道被替换的对象是否发生了变化。
// 注意：以前版本的DeltaFIFO也会对Replace事件使用Sync。所以只有当选项EmitDelaTypeReplaced为真时才会触发Replaced。
Replaced DeltaType = "Replaced"
// Sync是针对周期重新同步期间的合成事件
Sync DeltaType = "Sync"
)
// Delta 是被存储在DeltaFIFO中的类型，告诉我们发生了什么变化，对象在改变之后的状态
// 除非变化是删除，已经会在删除前得到最终状态
type Delta struct {
Type DeltaType
Object interface{}
}
3. FIFO
在client-go中，FIFO、Queue和Store之间的关系：
// staging/src/k8s.io/client-go/tools/cache/store.go
//Store是一个通用的对象存储和处理接口。存储区保存从字符串键到累加器的映射，并具有向当前与给定键关联的累加器添加、更新和删除给定对象的操作。存储区还知道如何从给定对象中提取密钥，因此许多操作只针对该对象进行。 在最简单的存储实现中，每个累加器只是最后一个给定的对象，或者在删除后为空，因此存储的行为是简单存储。 Reflector知道如何监视服务器和更新存储。这个包提供了Store的各种实现。
type Store interface {
// Add adds the given object to the accumulator associated with the given object's key
Add(obj interface{}) error
// Update updates the given object in the accumulator associated with the given object's key
Update(obj interface{}) error
// Delete deletes the given object from the accumulator associated with the given object's key
Delete(obj interface{}) error
// List returns a list of all the currently non-empty accumulators
List() []interface{}
// ListKeys returns a list of all the keys currently associated with non-empty accumulators
ListKeys() []string
// Get returns the accumulator associated with the given object's key
Get(obj interface{}) (item interface{}, exists bool, err error)
// GetByKey returns the accumulator associated with the given key
GetByKey(key string) (item interface{}, exists bool, err error)
// Replace will delete the contents of the store, using instead the
// given list. Store takes ownership of the list, you should not reference
// it after calling this function.
// 使用给定的list替换store中被删除的内容，不应该在这个函数执行后其他函数引用它，store独占list
Replace([]interface{}, string) error
// Resync is meaningless in the terms appearing here but has
// meaning in some implementations that have non-trivial
// additional behavior (e.g., DeltaFIFO).
// Resync在当前情况下是无意义的，但是在某些有附加动作的情况下是有意义的。
Resync() error
}
Queue接口的定义
// Queue使用Store Key集合将Store扩展到“process”。每次添加、更新或删除都可能会将对象的key放入该集合中。
// 队列可以在给定累加器的情况下派生相应的key。
// 可以从多个goroutine同时访问队列。
// 队列可以“关闭”，然后Pop操作返回错误。
type Queue interface {
Store
// 直到至少有一个key要处理或队列关闭，Pop才会阻塞。在后一种情况下，Pop返回时出错。在前一种情况下，
// Pop以原子方式选择一个key进行处理，从Store中删除该（key，累加器）关联，并处理累加器。Pop返回已处理的累加器和处理结果。
// PopProcessFunc可以返回ErrRequeue{internal}，在这种情况下，Pop将（a）作为原子处理的一部分将该（键、累加器）关联返回到队列，并（b）从Pop返回内部错误。
Pop(PopProcessFunc) (interface{}, error)
// AddIfNotPresent将给定累加器放入队列（与累加器的键关联），当且仅当该键尚未与非空累加器关联时。
AddIfNotPresent(interface{}) error
// 如果第一批keys全部弹出，HasSynced将返回true。如果第一次替换操作发生在任何添加、更新或删除之前，
// 则第一批key是第一次替换操作的key；否则，第一批为空。
HasSynced() bool
// Close the queue
Close()
}
FIFO就是Queue
var (
_ = Queue(&FIFO{}) // FIFO is a Queue
)
FIFO在client-go中的定义即具体使用
// staging/src/k8s.io/client-go/tools/cache/fifo.go
// FIFO解决了这个用例：
//您希望（精确地）处理每个对象一次。
//您希望在处理对象时处理该对象的最新版本。
//您不想处理已删除的对象，应将其从队列中删除。
//您不希望定期重新处理对象。
//对于其他用例，请与DeltaFIFO进行比较
type FIFO struct {
lock sync.RWMutex
cond sync.Cond
// We depend on the property that every key in `items` is also in `queue`
// 存储资源对象的value,依赖于queue中存储的key
items map[string]interface{}
// 存储资源对象的key
queue []string
// populated is true if the first batch of items inserted by Replace() has been populated
// or Delete/Add/Update was called first.
// 如果Replace（）插入的第一批项已填充，或者首先调用了DeleteAddUpdate，则填充为true。
populated bool
// initialPopulationCount is the number of items inserted by the first call of Replace()
// initialPopulationCount是第一次调用Replace（）插入的项数
initialPopulationCount int
// keyFunc is used to make the key used for queued item insertion and retrieval, and
// should be deterministic.
// 生成queue中的key的函数
keyFunc KeyFunc
// Indication the queue is closed.
// Used to indicate a queue is closed so a control loop can exit when a queue is empty.
// Currently, not used to gate any of CRED operations.
// 标识queue是否被关闭
closed bool
closedLock sync.Mutex
}
// Add插入一个item，并将其放入队列。仅当该项在集合中尚不存在时，才会将其排入队列。
func (f *FIFO) Add(obj interface{}) error {
id, err := f.keyFunc(obj)
if err != nil {
return KeyError{obj, err}
}
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
if _, exists := f.items[id]; !exists {
f.queue = append(f.queue, id)
}
f.items[id] = obj
f.cond.Broadcast()
return nil
}
// AddifyNotPresent插入一个item，并将其放入队列。如果该项已存在于集合中，则它既不会排队，也不会添加到集合中。这在单个producerconsumer场景中很有用，这样消费者就可以安全地重试项目，而不会与生产者发生冲突，也不会让过期的item排队。
func (f *FIFO) AddIfNotPresent(obj interface{}) error {
id, err := f.keyFunc(obj)
if err != nil {
return KeyError{obj, err}
}
f.lock.Lock()
defer f.lock.Unlock()
f.addIfNotPresent(id, obj)
return nil
}
// // addIfNotPresent假定fifo锁已被持有，如果提供的项不存在，则将其添加到id下的队列中。
func (f *FIFO) addIfNotPresent(id string, obj interface{}) {
f.populated = true
if _, exists := f.items[id]; exists {
return
}
f.queue = append(f.queue, id)
f.items[id] = obj
f.cond.Broadcast()
}
func (f *FIFO) Update(obj interface{}) error {
return f.Add(obj)
}
func (f *FIFO) Delete(obj interface{}) error {
id, err := f.keyFunc(obj)
if err != nil {
return KeyError{obj, err}
}
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
delete(f.items, id)
return err
}
// Pop等待item准备就绪并进行处理。如果有多个项目准备就绪，则将按照添加的顺序返回这些items。该项目在处理之前将从队列（和存储）中删除，因此如果处理不成功，则应使用AddifyNotPresent（）将其添加回。进程函数是在锁定状态下调用的，因此可以安全地更新其中需要与队列同步的数据结构。
func (f *FIFO) Pop(process PopProcessFunc) (interface{}, error) {
f.lock.Lock()
defer f.lock.Unlock()
for {
for len(f.queue) == 0 {
// When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
// 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().
// 当队列为空时，Pop（）的调用将被阻止，直到新项目进入队列。调用Close（）时，将设置f.closed并广播条件。这将导致此循环继续并从Pop（）返回。
if f.IsClosed() {
return nil, ErrFIFOClosed
}
f.cond.Wait()
}
id := f.queue[0]
f.queue = f.queue[1:]
if f.initialPopulationCount > 0 {
f.initialPopulationCount--
}
item, ok := f.items[id]
if !ok {
// Item may have been deleted subsequently.
continue
}
delete(f.items, id)
err := process(item)
if e, ok := err.(ErrRequeue); ok {
f.addIfNotPresent(id, item)
err = e.Err
}
return item, err
}
}
4. DeltaFIFO
// DeltaFIFO与FIFO类似，但在两个方面有所不同。 DeltaFIFO是一个Deltas，它是该对象的Delta值的一部分。将对象应用于Deltas意味着附加Delta，
//除非潜在附加的Delta是Deleted，并且Delta已经以Deleted结尾。在这种情况下，增量不会增长，但是如果旧的已删除对象是DeletedFinalStateUnknown，
//则已删除的terminal将替换为新的已删除对象。另一个区别是DeltaFIFO还有一种将对象应用于累加器的方法，称为同步。
//DeltaFIFO是生产者-消费者队列，其中Reflector是生产者，消费者是调用Pop（）方法的任何对象。
//DeltaFIFO解决了这个用例：
// 您希望最多处理一次每个对象更改（delta）。
// 处理对象时，希望看到自上次处理对象以来发生的所有事情。
// 您要进行某些对象的删除操作。
//您可能希望定期重新处理对象。
//DeltaFIFO的Pop（）、Get（）和GetByKey（）方法返回接口{}，以满足StoreQueue接口的要求，但它们总是返回Deltas类型的对象。
//DeltaFIFO的knownObjects KeyListTargeter提供列出Store keys和通过Store keys获取对象的能力。所讨论的对象称为“"known objects" ”，
//这组对象修改了Delete、Replace和Resync方法的行为（每种方法都有不同的方式）。
//关于线程的一个注意事项：
//如果从多个线程并行调用Pop（），最终可能会有多个线程处理同一对象的略有不同的版本。
type DeltaFIFO struct {
// lock/cond protects access to 'items' and 'queue'.
lock sync.RWMutex
cond sync.Cond
// We depend on the property that items in the set are in
// the queue and vice versa, and that all Deltas in this
// map have at least one Delta.
// 我们依赖于集合中的item 在queue中的属性，反之亦然，并且此映射中的所有增量都至少有一个增量。
items map[string]Deltas
queue []string
// populated is true if the first batch of items inserted by Replace() has been populated
// or Delete/Add/Update was called first.
// 如果首批items插入，将由Replace()填充，或者首先调用的是Delete/Add/Update populated设置为true,
populated bool
// initialPopulationCount is the number of items inserted by the first call of Replace()
initialPopulationCount int
// keyFunc is used to make the key used for queued item
// insertion and retrieval, and should be deterministic.
keyFunc KeyFunc
// knownObjects list keys that are "known" --- affecting Delete(),
// Replace(), and Resync()
// knownObjects列出“已知”的keys，这些键影响Delete（）、Replace（）和Resync（）。 knownObjects实际就是indexer
knownObjects KeyListerGetter
// Indication the queue is closed.
// Used to indicate a queue is closed so a control loop can exit when a queue is empty.
// Currently, not used to gate any of CRED operations.
// 表示队列已关闭。用于指示队列已关闭，以便在队列为空时可以退出控制循环。目前，未用于关闭任何CRED操作。
closed bool
closedLock sync.Mutex
// emitDeltaTypeReplaced is whether to emit the Replaced or Sync
// DeltaType when Replace() is called (to preserve backwards compat).
// emitDeltaTypeReplaced是在调用Replace（）时发出替换的DeltaType还是同步DeltaType（以保留向后兼容）
emitDeltaTypeReplaced bool
}
DeltaFIFO也是一个Queue
var (
_ = Queue(&DeltaFIFO{}) // DeltaFIFO is a Queue
)
// KeyOf()是暴露f的keyFunc，但也检测Deltas对象或DeletedFinalStateUnknown对象的key。
func (f *DeltaFIFO) KeyOf(obj interface{}) (string, error) {
if d, ok := obj.(Deltas); ok {
if len(d) == 0 {
return "", KeyError{obj, ErrZeroLengthDeltasObject}
}
obj = d.Newest().Object
}
if d, ok := obj.(DeletedFinalStateUnknown); ok {
return d.Key, nil
}
return f.keyFunc(obj)
}
// 如果先调用Add/Update/Delete/AddIfyNotPresent，或先调用更新，但已弹出Replace（）插入的第一批项，则HasSynced返回true
func (f *DeltaFIFO) HasSynced() bool {
f.lock.Lock()
defer f.lock.Unlock()
return f.populated && f.initialPopulationCount == 0
}
// Add inserts an item, and puts it in the queue. The item is only enqueued
// if it doesn't already exist in the set.
// Add插入一个item，并将其放入队列。仅当该项在集合中尚不存在时，才会将其排入队列。
func (f *DeltaFIFO) Add(obj interface{}) error {
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
return f.queueActionLocked(Added, obj)
}
// Update is just like Add, but makes an Updated Delta.
func (f *DeltaFIFO) Update(obj interface{}) error {
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
return f.queueActionLocked(Updated, obj)
}
// Delete is just like Add, but makes a Deleted Delta. If the given
// object does not already exist, it will be ignored. (It may have
// already been deleted by a Replace (re-list), for example.) In this
// method `f.knownObjects`, if not nil, provides (via GetByKey)
// _additional_ objects that are considered to already exist.
func (f *DeltaFIFO) Delete(obj interface{}) error {
id, err := f.KeyOf(obj)
if err != nil {
return KeyError{obj, err}
}
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
if f.knownObjects == nil {
if _, exists := f.items[id]; !exists {
// Presumably, this was deleted when a relist happened.
// Don't provide a second report of the same deletion.
return nil
}
} else {
// We only want to skip the "deletion" action if the object doesn't
// exist in knownObjects and it doesn't have corresponding item in items.
// Note that even if there is a "deletion" action in items, we can ignore it,
// because it will be deduped automatically in "queueActionLocked"
_, exists, err := f.knownObjects.GetByKey(id)
_, itemsExist := f.items[id]
if err == nil && !exists && !itemsExist {
// Presumably, this was deleted when a relist happened.
// Don't provide a second report of the same deletion.
return nil
}
}
return f.queueActionLocked(Deleted, obj)
}
// AddIfNotPresent inserts an item, and puts it in the queue. If the item is already
// present in the set, it is neither enqueued nor added to the set.
//
// This is useful in a single producer/consumer scenario so that the consumer can
// safely retry items without contending with the producer and potentially enqueueing
// stale items.
//
// Important: obj must be a Deltas (the output of the Pop() function). Yes, this is
// different from the Add/Update/Delete functions.
func (f *DeltaFIFO) AddIfNotPresent(obj interface{}) error {
deltas, ok := obj.(Deltas)
if !ok {
return fmt.Errorf("object must be of type deltas, but got: %#v", obj)
}
id, err := f.KeyOf(deltas.Newest().Object)
if err != nil {
return KeyError{obj, err}
}
f.lock.Lock()
defer f.lock.Unlock()
f.addIfNotPresent(id, deltas)
return nil
}
// addIfNotPresent inserts deltas under id if it does not exist, and assumes the caller
// already holds the fifo lock.
func (f *DeltaFIFO) addIfNotPresent(id string, deltas Deltas) {
f.populated = true
if _, exists := f.items[id]; exists {
return
}
f.queue = append(f.queue, id)
f.items[id] = deltas
f.cond.Broadcast()
}
// re-listing and watching can deliver the same update multiple times in any
// order. This will combine the most recent two deltas if they are the same.
// 去重Deltas
func dedupDeltas(deltas Deltas) Deltas {
n := len(deltas)
if n < 2 {
return deltas
}
a := &deltas[n-1]
b := &deltas[n-2]
if out := isDup(a, b); out != nil {
d := append(Deltas{}, deltas[:n-2]...)
return append(d, *out)
}
return deltas
}
// If a & b represent the same event, returns the delta that ought to be kept.
// Otherwise, returns nil.
// TODO: is there anything other than deletions that need deduping?
func isDup(a, b *Delta) *Delta {
if out := isDeletionDup(a, b); out != nil {
return out
}
// TODO: Detect other duplicate situations? Are there any?
return nil
}
// keep the one with the most information if both are deletions.
// 检测到两次相同的的删除操作
func isDeletionDup(a, b *Delta) *Delta {
if b.Type != Deleted || a.Type != Deleted {
return nil
}
// Do more sophisticated checks, or is this sufficient?
if _, ok := b.Object.(DeletedFinalStateUnknown); ok {
return a
}
return b
}
// queueActionLocked appends to the delta list for the object.
// Caller must lock first.
// queueActionLocked附加到对象的增量列表。调用方必须先加锁。
func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
id, err := f.KeyOf(obj)
if err != nil {
return KeyError{obj, err}
}
newDeltas := append(f.items[id], Delta{actionType, obj})
newDeltas = dedupDeltas(newDeltas)
if len(newDeltas) > 0 {
if _, exists := f.items[id]; !exists {
f.queue = append(f.queue, id)
}
f.items[id] = newDeltas
f.cond.Broadcast()
} else {
// This never happens, because dedupDeltas never returns an empty list
// when given a non-empty list (as it is here).
// But if somehow it ever does return an empty list, then
// We need to remove this from our map (extra items in the queue are
// ignored if they are not in the map).
// 这种情况永远不会发生，因为当给定一个非空列表时，dedupDeltas从不返回空列表。但如果它确实返回了一个空列表，那么我们需要从映射中删除它（如果队列中的其他项目不在映射中，则忽略它们）。
delete(f.items, id)
}
return nil
}
// List returns a list of all the items; it returns the object
// from the most recent Delta.
// You should treat the items returned inside the deltas as immutable.
// List返回所有项目的列表；它从最近的增量返回对象。您应该将delta中返回的项视为不可变的。
func (f *DeltaFIFO) List() []interface{} {
f.lock.RLock()
defer f.lock.RUnlock()
return f.listLocked()
}
func (f *DeltaFIFO) listLocked() []interface{} {
list := make([]interface{}, 0, len(f.items))
for _, item := range f.items {
list = append(list, item.Newest().Object)
}
return list
}
// ListKeys returns a list of all the keys of the objects currently
// in the FIFO.
func (f *DeltaFIFO) ListKeys() []string {
f.lock.RLock()
defer f.lock.RUnlock()
list := make([]string, 0, len(f.items))
for key := range f.items {
list = append(list, key)
}
return list
}
// Get returns the complete list of deltas for the requested item,
// or sets exists=false.
// You should treat the items returned inside the deltas as immutable.
func (f *DeltaFIFO) Get(obj interface{}) (item interface{}, exists bool, err error) {
key, err := f.KeyOf(obj)
if err != nil {
return nil, false, KeyError{obj, err}
}
return f.GetByKey(key)
}
// GetByKey returns the complete list of deltas for the requested item,
// setting exists=false if that list is empty.
// You should treat the items returned inside the deltas as immutable.
func (f *DeltaFIFO) GetByKey(key string) (item interface{}, exists bool, err error) {
f.lock.RLock()
defer f.lock.RUnlock()
d, exists := f.items[key]
if exists {
// Copy item's slice so operations on this slice
// won't interfere with the object we return.
d = copyDeltas(d)
}
return d, exists, nil
}
func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
f.lock.Lock()
defer f.lock.Unlock()
for {
for len(f.queue) == 0 {
// When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
// 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, ErrFIFOClosed
}
f.cond.Wait()
}
id := f.queue[0]
f.queue = f.queue[1:]
if f.initialPopulationCount > 0 {
f.initialPopulationCount--
}
item, ok := f.items[id]
if !ok {
// Item may have been deleted subsequently.
continue
}
delete(f.items, id)
err := process(item)
if e, ok := err.(ErrRequeue); ok {
f.addIfNotPresent(id, item)
err = e.Err
}
// Don't need to copyDeltas here, because we're transferring
// ownership to the caller.
return item, err
}
}
func (f *DeltaFIFO) Replace(list []interface{}, resourceVersion string) error {
f.lock.Lock()
defer f.lock.Unlock()
keys := make(sets.String, len(list))
// keep backwards compat for old clients
action := Sync
if f.emitDeltaTypeReplaced {
action = Replaced
}
for _, item := range list {
key, err := f.KeyOf(item)
if err != nil {
return KeyError{item, err}
}
keys.Insert(key)
if err := f.queueActionLocked(action, item); err != nil {
return fmt.Errorf("couldn't enqueue object: %v", err)
}
}
if f.knownObjects == nil {
// Do deletion detection against our own list.
queuedDeletions := 0
for k, oldItem := range f.items {
if keys.Has(k) {
continue
}
var deletedObj interface{}
if n := oldItem.Newest(); n != nil {
deletedObj = n.Object
}
queuedDeletions++
if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil {
return err
}
}
if !f.populated {
f.populated = true
// While there shouldn't be any queued deletions in the initial
// population of the queue, it's better to be on the safe side.
f.initialPopulationCount = len(list) + queuedDeletions
}
return nil
}
// Detect deletions not already in the queue.
knownKeys := f.knownObjects.ListKeys()
queuedDeletions := 0
for _, k := range knownKeys {
if keys.Has(k) {
continue
}
deletedObj, exists, err := f.knownObjects.GetByKey(k)
if err != nil {
deletedObj = nil
klog.Errorf("Unexpected error %v during lookup of key %v, placing DeleteFinalStateUnknown marker without object", err, k)
} else if !exists {
deletedObj = nil
klog.Infof("Key %v does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", k)
}
queuedDeletions++
if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil {
return err
}
}
if !f.populated {
f.populated = true
f.initialPopulationCount = len(list) + queuedDeletions
}
return nil
}
