189 8069 5689
k8s为实现容器探活worker的管理构建了一个Manager组件，该组件负责底层探活worker的管理,并且缓存当前的容器的状态，并对外同步容器的当前状态，今天我们就来分析下其部分核心组件
创新互联成立十多年来，这条路我们正越走越好，积累了技术与客户资源，形成了良好的口碑。为客户提供成都网站建设、成都做网站、网站策划、网页设计、主机域名、网络营销、VI设计、网站改版、漏洞修补等服务。网站是否美观、功能强大、用户体验好、性价比高、打开快等等，这些对于网站建设都非常重要，创新互联通过对建站技术性的掌握、对创意设计的研究为客户提供一站式互联网解决方案，携手广大客户，共同发展进步。
Manager缓存的状态主要是会被kubelet、状态组件消费，并且在Pod同步状态的时候，会通过当前Manager里面的探测状态来更新Pod的容器的就绪与启动状态的更新，让我们一起看看Manager自身的一些关键实现吧
即prober/results/results_manager组件，其主要作用是：存储探测结果和通知探测结果
cache负责容器的探测结果的保存，updates则负责对外更新状态的订阅，其通过新的结果和cache中的状态进行对比，从而决定是否对外通知
// Manager implementation.
type manager struct {
// 保护cache
sync.RWMutex
// 容器ID->探测结果
cache map[kubecontainer.ContainerID]Result
// 更新管道
updates chan Update
}
更新缓存的时候回通过对比前后状态来进行是否发布变更事件，从而通知到外部订阅容器变更的kubelet核心流程
func (m *manager) Set(id kubecontainer.ContainerID, result Result, pod *v1.Pod) {
// 修改内部状态
if m.setInternal(id, result) {
// 同步更新事件
m.updates <- Update{id, result, pod.UID}
}
}内部状态修改与判断是否进行同步实现
// 如果之前的缓存不存在，或者前后状态不一致则会返回true触发更新
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
}func (m *manager) Updates() <-chan Update {
return m.updates
}探测管理器是指的prober/prober)manager的Manager组件，其负责当前kubelet上面探活组件的管理，并且进行探测状态结果的缓存与同步，并且内部还通过statusManager来进行apiserver状态的同步
每个探测Key包含要探测的目标信息：pod的ID、容器名、探测类型
type probeKey struct {
podUID types.UID
containerName string
probeType probeType
}
statusManager组件在后续章节里面会进行详细分析,说下livenessManager该组件即探活的结果，所以当一个容器探测失败，则会由kubelet本地先进行处理，而readlinessManager和startupManager则需要通过statusManager同步给apiserver进行同步
type manager struct {
//探测Key与worker映射
workers map[probeKey]*worker
// 读写锁
workerLock sync.RWMutex
//statusManager缓存为探测提供pod IP和容器id。
statusManager status.Manager
// 存储readiness探测结果
readinessManager results.Manager
// 存储liveness探测结果
livenessManager results.Manager
// 存储startup探测结果
startupManager results.Manager
// 执行探测操作
prober *prober
}func (m *manager) updateStartup() {
// 从管道获取数据进行同步
update := <-m.startupManager.Updates()
started := update.Result == results.Success
m.statusManager.SetContainerStartup(update.PodUID, update.ContainerID, started)
}func (m *manager) updateReadiness() {
update := <-m.readinessManager.Updates()
ready := update.Result == results.Success
m.statusManager.SetContainerReadiness(update.PodUID, update.ContainerID, ready)
}func (m *manager) Start() {
// Start syncing readiness.
go wait.Forever(m.updateReadiness, 0)
// Start syncing startup.
go wait.Forever(m.updateStartup, 0)
}
添加 Pod的时候会遍历Pod的所有容器，并根据探测类型来进行对应探测worker的构建
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的探测任务的构建
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
}
// 构建新的worker
w := newWorker(m, startup, pod, c)
m.workers[key] = w
go w.run()
}
// 针对ReadinessProbe的探测任务的构建
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的探测任务的构建
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()
}
}
}更新Pod状态主要是根据当前Manager里面缓存的之前的状态信息来更新Pod里面对应容器的状态，这些状态是Pod里面容器最新的探测状态，获取这些状态则是检测当前的容器是否已经就绪和启动，为后续更新流程做基础数据
for i, c := range podStatus.ContainerStatuses {
var ready bool
// 检测容器状态
if c.State.Running == nil {
ready = false
} else if result, ok := m.readinessManager.Get(kubecontainer.ParseContainerID(c.ContainerID)); ok {
// 检测readinessMnager里面的状态，如果是成功则就是已经就绪
ready = result == results.Success
} else {
// 检查是否有尚未运行的探测器。只要存在则认为就绪
_, 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里面的状态是成功的则认为是已经启动的
started = result == results.Success
} else {
// 检查是否有尚未运行的探测器。
_, exists := m.getWorker(podUID, c.Name, startup)
started = !exists
}
podStatus.ContainerStatuses[i].Started = &started
}针对初始化容器主要容器已经终止并且退出的状态码为0,则认为初始化容器已经就绪
for i, c := range podStatus.InitContainerStatuses {
var ready bool
if c.State.Terminated != nil && c.State.Terminated.ExitCode == 0 {
// 容器状态
ready = true
}
podStatus.InitContainerStatuses[i].Ready = ready
}存活状态通知主要是在kubelet的核心流程循环中进行的，如果检测到容器的状态失败，会立刻进行对应pod的容器状态的同步，从而决定下一步的操作是做什么
case update := <-kl.livenessManager.Updates():
// 如果探测状态失败
if update.Result == proberesults.Failure {
// 省略代码
handler.HandlePodSyncs([]*v1.Pod{pod})
}探活整体的设计大概就是这样，接下来会分期其statusManager组件，即将将探测的状态与apiserver的同步的实现, k8s源码阅读电子书地址: https://www.yuque.com/baxiaoshi/tyado3