1. 前言

轉(zhuǎn)載請說明原文出處, 尊重他人勞動成果!

源碼位置: https://github.com/nicktming/kubernetes/blob/tming-v1.13/pkg/controller/controller_ref_manager.go
分支: tming-v1.13 (基于v1.13版本)

本文將分析controller中的一些公共結(jié)構(gòu)體, 因為很多的controller會共用這些結(jié)構(gòu)體, 所以在分析各種controller之前了解這些結(jié)構(gòu)體的作用很有必要. 本文將主要涉及pkg/controller/controller_utils.go.

2. ControllerExpectationsInterface

architecture.png

該接口會在controller中被用到, 比如在replicaset的controller中會被用來決定是否需要重新計算replicas數(shù)與實際數(shù)目是否已經(jīng)相等, 如果不相等, 則需要進行計算需要刪除多少pod或者需要增加多少pod等等.

2.1 ControlleeExpectations

type ControllerExpectationsInterface interface {
    GetExpectations(controllerKey string) (*ControlleeExpectations, bool, error)
    SatisfiedExpectations(controllerKey string) bool
    DeleteExpectations(controllerKey string)
    SetExpectations(controllerKey string, add, del int) error
    ExpectCreations(controllerKey string, adds int) error
    ExpectDeletions(controllerKey string, dels int) error
    CreationObserved(controllerKey string)
    DeletionObserved(controllerKey string)
    RaiseExpectations(controllerKey string, add, del int)
    LowerExpectations(controllerKey string, add, del int)
}
type Expectations interface {
    Fulfilled() bool
}
// ControlleeExpectations track controllee creates/deletes.
type ControlleeExpectations struct {
    add       int64
    del       int64
    key       string
    timestamp time.Time
}
func (e *ControlleeExpectations) Add(add, del int64) {
    atomic.AddInt64(&e.add, add)
    atomic.AddInt64(&e.del, del)
}
func (e *ControlleeExpectations) Fulfilled() bool {
    return atomic.LoadInt64(&e.add) <= 0 && atomic.LoadInt64(&e.del) <= 0
}
func (e *ControlleeExpectations) GetExpectations() (int64, int64) {
    return atomic.LoadInt64(&e.add), atomic.LoadInt64(&e.del)
}
func (exp *ControlleeExpectations) isExpired() bool {
    return clock.RealClock{}.Since(exp.timestamp) > ExpectationsTimeout
}

ControlleeExpectations: 用來維護某個key的add和del這兩個變量. 另外Fulfilled的條件是這兩個都已經(jīng)小于等于0了.

2.2 ControllerExpectations

type ControllerExpectations struct {
    cache.Store
}
func NewControllerExpectations() *ControllerExpectations {
    return &ControllerExpectations{cache.NewStore(ExpKeyFunc)}
}
func (r *ControllerExpectations) SetExpectations(controllerKey string, add, del int) error {
    exp := &ControlleeExpectations{add: int64(add), del: int64(del), key: controllerKey, timestamp: clock.RealClock{}.Now()}
    klog.V(4).Infof("Setting expectations %#v", exp)
    return r.Add(exp)
}
func (r *ControllerExpectations) DeleteExpectations(controllerKey string) {
    if exp, exists, err := r.GetByKey(controllerKey); err == nil && exists {
        if err := r.Delete(exp); err != nil {
            klog.V(2).Infof("Error deleting expectations for controller %v: %v", controllerKey, err)
        }
    }
}
func (r *ControllerExpectations) GetExpectations(controllerKey string) (*ControlleeExpectations, bool, error) {
    if exp, exists, err := r.GetByKey(controllerKey); err == nil && exists {
        return exp.(*ControlleeExpectations), true, nil
    } else {
        return nil, false, err
    }
}

該結(jié)構(gòu)體ControllerExpectations只有一個變量cache.Store, 用來存儲的, 明顯存儲的是ControlleeExpectations類型.
SetExpectations: 構(gòu)造了一個ControlleeExpectations方法進行存儲起來.
DeleteExpectations: 刪除這個key對應(yīng)的ControlleeExpectations.
GetExpectations: 獲取該controllerKey對應(yīng)的ControlleeExpectations.

接下來看一下它的更新方法, 都是常規(guī)操作, 操作store里面的內(nèi)容.

func (r *ControllerExpectations) ExpectCreations(controllerKey string, adds int) error {
    return r.SetExpectations(controllerKey, adds, 0)
}
func (r *ControllerExpectations) ExpectDeletions(controllerKey string, dels int) error {
    return r.SetExpectations(controllerKey, 0, dels)
}
func (r *ControllerExpectations) LowerExpectations(controllerKey string, add, del int) {
    if exp, exists, err := r.GetExpectations(controllerKey); err == nil && exists {
        exp.Add(int64(-add), int64(-del))
        // The expectations might've been modified since the update on the previous line.
        klog.V(4).Infof("Lowered expectations %#v", exp)
    }
}
func (r *ControllerExpectations) RaiseExpectations(controllerKey string, add, del int) {
    if exp, exists, err := r.GetExpectations(controllerKey); err == nil && exists {
        exp.Add(int64(add), int64(del))
        // The expectations might've been modified since the update on the previous line.
        klog.V(4).Infof("Raised expectations %#v", exp)
    }
}
func (r *ControllerExpectations) CreationObserved(controllerKey string) {
    r.LowerExpectations(controllerKey, 1, 0)
}
func (r *ControllerExpectations) DeletionObserved(controllerKey string) {
    r.LowerExpectations(controllerKey, 0, 1)
}

這里都是更新方法, 方法很簡單, 但是每個方法背后的意義還是需要結(jié)合真正的controller來進行說明. (后面分析controller的時候會有說明)

這里一個比較重要的方法就是SatisfiedExpectations方法

// 返回true時 代表需要去同步
func (r *ControllerExpectations) SatisfiedExpectations(controllerKey string) bool {
    if exp, exists, err := r.GetExpectations(controllerKey); exists {
        if exp.Fulfilled() {
            klog.V(4).Infof("Controller expectations fulfilled %#v", exp)
            return true
        } else if exp.isExpired() {
            // 已經(jīng)過期了 需要去同步
            klog.V(4).Infof("Controller expectations expired %#v", exp)
            return true
        } else {
            klog.V(4).Infof("Controller still waiting on expectations %#v", exp)
            return false
        }
    } else if err != nil {
        klog.V(2).Infof("Error encountered while checking expectations %#v, forcing sync", err)
    } else {
        // When a new controller is created, it doesn't have expectations.
        // When it doesn't see expected watch events for > TTL, the expectations expire.
        //  - In this case it wakes up, creates/deletes controllees, and sets expectations again.
        // When it has satisfied expectations and no controllees need to be created/destroyed > TTL, the expectations expire.
        //  - In this case it continues without setting expectations till it needs to create/delete controllees.
        klog.V(4).Infof("Controller %v either never recorded expectations, or the ttl expired.", controllerKey)
    }
    // Trigger a sync if we either encountered and error (which shouldn't happen since we're
    // getting from local store) or this controller hasn't established expectations.
    return true
}

該方法會在分析replicaset的controller時進行說明.

2.3 UIDTrackingControllerExpectations

type UIDTrackingControllerExpectations struct {
    ControllerExpectationsInterface
    uidStoreLock sync.Mutex
    uidStore cache.Store
}
func NewUIDTrackingControllerExpectations(ce ControllerExpectationsInterface) *UIDTrackingControllerExpectations {
    return &UIDTrackingControllerExpectations{ControllerExpectationsInterface: ce, uidStore: cache.NewStore(UIDSetKeyFunc)}
}
type UIDSet struct {
    sets.String
    key string
}

該結(jié)構(gòu)體的設(shè)計是為了期望刪除的時候需要知道刪除哪些元素, 因為ControllerExpectationsInterface中只能用存儲的del屬性只能是一個數(shù)字, 并不能知道具體要刪除哪些對象.
所以在此基礎(chǔ)上加入UIDSet表明該key具體要刪除哪些對象. 因此除了維護ControlleeExpectations之外還要維護UIDSet.

看一下如何存儲的, 保存了ControlleeExpectations和UIDSet.

func (u *UIDTrackingControllerExpectations) ExpectDeletions(rcKey string, deletedKeys []string) error {
    u.uidStoreLock.Lock()
    defer u.uidStoreLock.Unlock()
    // 如果以前存在
    if existing := u.GetUIDs(rcKey); existing != nil && existing.Len() != 0 {
        klog.Errorf("Clobbering existing delete keys: %+v", existing)
    }
    expectedUIDs := sets.NewString()
    for _, k := range deletedKeys {
        expectedUIDs.Insert(k)
    }
    klog.V(4).Infof("Controller %v waiting on deletions for: %+v", rcKey, deletedKeys)
    // 維護一個UIDSet
    if err := u.uidStore.Add(&UIDSet{expectedUIDs, rcKey}); err != nil {
        return err
    }
    // 維護ControlleeExpectations
    return u.ControllerExpectationsInterface.ExpectDeletions(rcKey, expectedUIDs.Len())
}
func (u *UIDTrackingControllerExpectations) GetUIDs(controllerKey string) sets.String {
    if uid, exists, err := u.uidStore.GetByKey(controllerKey); err == nil && exists {
        return uid.(*UIDSet).String
    }
    return nil
}

具體刪除某一個

func (u *UIDTrackingControllerExpectations) DeletionObserved(rcKey, deleteKey string) {
    u.uidStoreLock.Lock()
    defer u.uidStoreLock.Unlock()

    uids := u.GetUIDs(rcKey)
    if uids != nil && uids.Has(deleteKey) {
        klog.V(4).Infof("Controller %v received delete for pod %v", rcKey, deleteKey)
        //維護ControlleeExpectations
        u.ControllerExpectationsInterface.DeletionObserved(rcKey)
        // 維護UIDSet
        uids.Delete(deleteKey)
    }
}

刪除全部某個controllerKey的全部內(nèi)容.

func (u *UIDTrackingControllerExpectations) DeleteExpectations(rcKey string) {
    u.uidStoreLock.Lock()
    defer u.uidStoreLock.Unlock()

    u.ControllerExpectationsInterface.DeleteExpectations(rcKey)
    if uidExp, exists, err := u.uidStore.GetByKey(rcKey); err == nil && exists {
        if err := u.uidStore.Delete(uidExp); err != nil {
            klog.V(2).Infof("Error deleting uid expectations for controller %v: %v", rcKey, err)
        }
    }
}