# IT明星不是夢 #圖解kubernetes容器探活機(jī)制核

k8s為實現(xiàn)容器探活worker的管理構(gòu)建了一個Manager組件，該組件負(fù)責(zé)底層探活worker的管理,并且緩存當(dāng)前的容器的狀態(tài)，并對外同步容器的當(dāng)前狀態(tài)，今天我們就來分析下其部分核心組件

1. 核心原理實現(xiàn)

Manager緩存的狀態(tài)主要是會被kubelet、狀態(tài)組件消費，并且在Pod同步狀態(tài)的時候，會通過當(dāng)前Manager里面的探測狀態(tài)來更新Pod的容器的就緒與啟動狀態(tài)的更新，讓我們一起看看Manager自身的一些關(guān)鍵實現(xiàn)吧

2. 探活結(jié)果管理

即prober/results/results_manager組件，其主要作用是：存儲探測結(jié)果和通知探測結(jié)果

2.1 核心數(shù)據(jù)結(jié)構(gòu)

cache負(fù)責(zé)容器的探測結(jié)果的保存，updates則負(fù)責(zé)對外更新狀態(tài)的訂閱，其通過新的結(jié)果和cache中的狀態(tài)進(jìn)行對比，從而決定是否對外通知

// Manager implementation.
type manager struct {
    // 保護(hù)cache
    sync.RWMutex
    // 容器ID->探測結(jié)果
    cache map[kubecontainer.ContainerID]Result
    // 更新管道
    updates chan Update
}

2.2 更新緩存通知事件

更新緩存的時候回通過對比前后狀態(tài)來進(jìn)行是否發(fā)布變更事件，從而通知到外部訂閱容器變更的kubelet核心流程

func (m *manager) Set(id kubecontainer.ContainerID, result Result, pod *v1.Pod) {
    // 修改內(nèi)部狀態(tài)
    if m.setInternal(id, result) {
        // 同步更新事件
        m.updates <- Update{id, result, pod.UID}
    }
}

內(nèi)部狀態(tài)修改與判斷是否進(jìn)行同步實現(xiàn)

// 如果之前的緩存不存在，或者前后狀態(tài)不一致則會返回true觸發(fā)更新
func (m *manager) setInternal(id kubecontainer.ContainerID, result Result) bool {
    m.Lock()
    defer m.Unlock()
    prev, exists := m.cache[id]
    if !exists || prev != result {
        m.cache[id] = result
        return true
    }
    return false
}

2.3 對外更新管道

func (m *manager) Updates() <-chan Update {
    return m.updates
}

3.探測管理器

探測管理器是指的prober/prober)manager的Manager組件，其負(fù)責(zé)當(dāng)前kubelet上面探活組件的管理，并且進(jìn)行探測狀態(tài)結(jié)果的緩存與同步，并且內(nèi)部還通過statusManager來進(jìn)行apiserver狀態(tài)的同步

3.1 容器探測Key

每個探測Key包含要探測的目標(biāo)信息：pod的ID、容器名、探測類型

type probeKey struct {
    podUID        types.UID
    containerName string
    probeType     probeType
}

3.2 核心數(shù)據(jù)結(jié)構(gòu)

statusManager組件在后續(xù)章節(jié)里面會進(jìn)行詳細(xì)分析,說下livenessManager該組件即探活的結(jié)果，所以當(dāng)一個容器探測失敗，則會由kubelet本地先進(jìn)行處理，而readlinessManager和startupManager則需要通過statusManager同步給apiserver進(jìn)行同步

type manager struct {
    //探測Key與worker映射
    workers map[probeKey]*worker
    // 讀寫鎖
    workerLock sync.RWMutex

    //statusManager緩存為探測提供pod IP和容器id。
    statusManager status.Manager

    // 存儲readiness探測結(jié)果
    readinessManager results.Manager

    // 存儲liveness探測結(jié)果
    livenessManager results.Manager

    // 存儲startup探測結(jié)果
    startupManager results.Manager

    // 執(zhí)行探測操作
    prober *prober
}

3.3 同步startup探測結(jié)果

func (m *manager) updateStartup() {
    // 從管道獲取數(shù)據(jù)進(jìn)行同步
    update := <-m.startupManager.Updates()

    started := update.Result == results.Success
    m.statusManager.SetContainerStartup(update.PodUID, update.ContainerID, started)
}

3.4 同步readiness探測結(jié)果

func (m *manager) updateReadiness() {
    update := <-m.readinessManager.Updates()

    ready := update.Result == results.Success
    m.statusManager.SetContainerReadiness(update.PodUID, update.ContainerID, ready)
}

3.5 啟動同步探測結(jié)果后臺任務(wù)

func (m *manager) Start() {
    // Start syncing readiness.
    go wait.Forever(m.updateReadiness, 0)
    // Start syncing startup.
    go wait.Forever(m.updateStartup, 0)
}

3.6 添加Pod探測

添加 Pod的時候會遍歷Pod的所有容器，并根據(jù)探測類型來進(jìn)行對應(yīng)探測worker的構(gòu)建

func (m *manager) AddPod(pod *v1.Pod) {
    m.workerLock.Lock()
    defer m.workerLock.Unlock()

    key := probeKey{podUID: pod.UID}
    for _, c := range pod.Spec.Containers {
        key.containerName = c.Name

        // 針對startupProbe的探測任務(wù)的構(gòu)建
        if c.StartupProbe != nil && utilfeature.DefaultFeatureGate.Enabled(features.StartupProbe) {
            key.probeType = startup
            if _, ok := m.workers[key]; ok {
                klog.Errorf("Startup probe already exists! %v - %v",
                    format.Pod(pod), c.Name)
                return
            }
            // 構(gòu)建新的worker
            w := newWorker(m, startup, pod, c)
            m.workers[key] = w
            go w.run()
        }

        // 針對ReadinessProbe的探測任務(wù)的構(gòu)建
        if c.ReadinessProbe != nil {
            key.probeType = readiness
            if _, ok := m.workers[key]; ok {
                klog.Errorf("Readiness probe already exists! %v - %v",
                    format.Pod(pod), c.Name)
                return
            }
            w := newWorker(m, readiness, pod, c)
            m.workers[key] = w
            go w.run()
        }

        // 針對LivenessProbe的探測任務(wù)的構(gòu)建
        if c.LivenessProbe != nil {
            key.probeType = liveness
            if _, ok := m.workers[key]; ok {
                klog.Errorf("Liveness probe already exists! %v - %v",
                    format.Pod(pod), c.Name)
                return
            }
            w := newWorker(m, liveness, pod, c)
            m.workers[key] = w
            go w.run()
        }
    }
}

3.7 更新Pod狀態(tài)

更新Pod狀態(tài)主要是根據(jù)當(dāng)前Manager里面緩存的之前的狀態(tài)信息來更新Pod里面對應(yīng)容器的狀態(tài)，這些狀態(tài)是Pod里面容器最新的探測狀態(tài)，獲取這些狀態(tài)則是檢測當(dāng)前的容器是否已經(jīng)就緒和啟動，為后續(xù)更新流程做基礎(chǔ)數(shù)據(jù)

3.7.1 容器狀態(tài)更新

    for i, c := range podStatus.ContainerStatuses {
        var ready bool
        // 檢測容器狀態(tài)
        if c.State.Running == nil {
            ready = false
        } else if result, ok := m.readinessManager.Get(kubecontainer.ParseContainerID(c.ContainerID)); ok {
            // 檢測readinessMnager里面的狀態(tài)，如果是成功則就是已經(jīng)就緒
            ready = result == results.Success
        } else {
            // 檢查是否有尚未運行的探測器。只要存在則認(rèn)為就緒
            _, exists := m.getWorker(podUID, c.Name, readiness)
            ready = !exists
        }
        podStatus.ContainerStatuses[i].Ready = ready

        var started bool
        if c.State.Running == nil {
            started = false
        } else if !utilfeature.DefaultFeatureGate.Enabled(features.StartupProbe) {
            // 容器正在運行，如果StartupProbe功能被禁用，則假定它已啟動
            started = true
        } else if result, ok := m.startupManager.Get(kubecontainer.ParseContainerID(c.ContainerID)); ok {
            // 如果startupManager里面的狀態(tài)是成功的則認(rèn)為是已經(jīng)啟動的
            started = result == results.Success
        } else {
            // 檢查是否有尚未運行的探測器。
            _, exists := m.getWorker(podUID, c.Name, startup)
            started = !exists
        }
        podStatus.ContainerStatuses[i].Started = &started
    }

3.7.2 初始化容器狀態(tài)更新

針對初始化容器主要容器已經(jīng)終止并且退出的狀態(tài)碼為0,則認(rèn)為初始化容器已經(jīng)就緒

    for i, c := range podStatus.InitContainerStatuses {
        var ready bool
        if c.State.Terminated != nil && c.State.Terminated.ExitCode == 0 {
            // 容器狀態(tài)
            ready = true
        }
        podStatus.InitContainerStatuses[i].Ready = ready
    }

3.8 存活狀態(tài)通知

存活狀態(tài)通知主要是在kubelet的核心流程循環(huán)中進(jìn)行的，如果檢測到容器的狀態(tài)失敗，會立刻進(jìn)行對應(yīng)pod的容器狀態(tài)的同步，從而決定下一步的操作是做什么

    case update := <-kl.livenessManager.Updates():
        // 如果探測狀態(tài)失敗
        if update.Result == proberesults.Failure {
            // 省略代碼
            handler.HandlePodSyncs([]*v1.Pod{pod})
        }

探活整體的設(shè)計大概就是這樣，接下來會分期其statusManager組件，即將將探測的狀態(tài)與apiserver的同步的實現(xiàn), k8s源碼閱讀電子書地址: https://www.yuque.com/baxiaoshi/tyado3