plegManager | 南风未起｜云原生/Go技术分享博客

# 1.简介

pleg(pod lifecycle event generate)是kubelet的事件来源之一，负责周期调用containerRuntime获取底层的container状态，生成pod生命周期事件及推送到plegChan，驱动kubelet进行相应处理。

const (
	// ContainerStarted - event type when the new state of container is running.
	ContainerStarted PodLifeCycleEventType = "ContainerStarted"
	// ContainerDied - event type when the new state of container is exited.
	ContainerDied PodLifeCycleEventType = "ContainerDied"
	// ContainerRemoved - event type when the old state of container is exited.
	ContainerRemoved PodLifeCycleEventType = "ContainerRemoved"
	// PodSync is used to trigger syncing of a pod when the observed change of
	// the state of the pod cannot be captured by any single event above.
	PodSync PodLifeCycleEventType = "PodSync"
	// ContainerChanged - event type when the new state of container is unknown.
	ContainerChanged PodLifeCycleEventType = "ContainerChanged"
)

// PodLifecycleEvent is an event that reflects the change of the pod state.
type PodLifecycleEvent struct {
	// The pod ID.
	ID types.UID
	// The type of the event.
	Type PodLifeCycleEventType
	// The accompanied data which varies based on the event type.
	//   - ContainerStarted/ContainerStopped: the container name (string).
	//   - All other event types: unused.
	Data interface{}
}

// PodLifecycleEventGenerator contains functions for generating pod life cycle events.
type PodLifecycleEventGenerator interface {
    // 后台生成pod event
	Start()
    // pleg事件管道
	Watch() chan *PodLifecycleEvent
    // 单次执行reList时间是否超出3min(响应慢/pod太多/runtime异常)
	Healthy() (bool, error)
}

// GenericPLEG is an extremely simple generic PLEG that relies solely on
// periodic listing to discover container changes.
type GenericPLEG struct {
	// The period for relisting.
	relistPeriod time.Duration
	// The container runtime.
	runtime kubecontainer.Runtime
	// The channel from which the subscriber listens events.
	eventChannel chan *PodLifecycleEvent
	// The internal cache for pod/container information.
	podRecords podRecords
	// Time of the last relisting.
	relistTime atomic.Value
	// Cache for storing the runtime states required for syncing pods.
	cache kubecontainer.Cache
	// For testability.
	clock clock.Clock
	// Pods that failed to have their status retrieved during a relist. These pods will be
	// retried during the next relisting.
	podsToReinspect map[types.UID]*kubecontainer.Pod
}

// NewGenericPLEG instantiates a new GenericPLEG object and return it.
func NewGenericPLEG(runtime kubecontainer.Runtime, channelCapacity int,
	relistPeriod time.Duration, cache kubecontainer.Cache, clock clock.Clock) PodLifecycleEventGenerator {
	return &GenericPLEG{
        // 执行周期(1s)
		relistPeriod: relistPeriod,
        // containerRuntime
		runtime:      runtime,
		eventChannel: make(chan *PodLifecycleEvent, channelCapacity),
        // 缓存来自运行时的pod
		podRecords:   make(podRecords),
        // 缓存来自运行时的podStatus
		cache:        cache,
		clock:        clock,
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

注意

1.pleg的出现更多的是解决kubelet感知runtime状态，避免podWorker独立轮询容器状态造成的大量CPU开销，集中处理容器状态感知

2.pleg默认间隔1s集中轮询容器状态，重新将重启组装为pod与缓存数据对比，生成的事件放入kubelet感知的plegChan进行处理

# 2.pleg

# 2.1.Start

kubelet启动期间调用pleg.Start()间隔1s周期执行reList获取运行时的pod/container，基于内部维护的pod/container进行对比生成相应事件。

// plegRelistPeriod = time.Second * 1
// Start spawns a goroutine to relist periodically.
func (g *GenericPLEG) Start() {
	go wait.Until(g.relist, g.relistPeriod, wait.NeverStop)
}

// relist queries the container runtime for list of pods/containers, compare
// with the internal pods/containers, and generates events accordingly.
func (g *GenericPLEG) relist() {
	
    ...
	// Get all the pods.
	podList, err := g.runtime.GetPods(true)
	...

	pods := kubecontainer.Pods(podList)
	...
    // 本次获取的pods设置为current
	g.podRecords.setCurrent(pods)

	// Compare the old and the current pods, and generate events.、
    // podRecords[pid].current保存的是pod最新的状态
	// podRecords[pid].old保存的是上次relist时pod的状态
	eventsByPodID := map[types.UID][]*PodLifecycleEvent{}
	for pid := range g.podRecords {
		oldPod := g.podRecords.getOld(pid)
		pod := g.podRecords.getCurrent(pid)
		// Get all containers in the old and the new pod.
		allContainers := getContainersFromPods(oldPod, pod)
		for _, container := range allContainers {
            // 生成事件
			events := computeEvents(oldPod, pod, &container.ID)
			for _, e := range events {
                // 记录事件
				updateEvents(eventsByPodID, e)
			}
		}
	}
	...

	// If there are events associated with a pod, we should update the podCache.
	for pid, events := range eventsByPodID {
        // 获取事件对应pod
		pod := g.podRecords.getCurrent(pid)
        // kubelet传入的podCache不为空
		if g.cacheEnabled() {
			// 更新podCache
			if err := g.updateCache(pod, pid); err != nil {
				// make sure we try to reinspect the pod during the next relisting
				needsReinspection[pid] = pod
				continue
			} else {
				// 更新成功,清理重试队列的pod
				delete(g.podsToReinspect, pid)
			}
		}
		// Update the internal storage and send out the events.
		g.podRecords.update(pid)

		// Map from containerId to exit code; used as a temporary cache for lookup
		containerExitCode := make(map[string]int)

        // 推送事件
		for i := range events {
			// Filter out events that are not reliable and no other components use yet.
            // containerChanged事件代表不稳定
			if events[i].Type == ContainerChanged {
				continue
			}
            
			select {
            // 推送事件至podLifecycleEvent
			case g.eventChannel <- events[i]:
			default:
				metrics.PLEGDiscardEvents.Inc()
				klog.ErrorS(nil, "Event channel is full, discard this relist() cycle event")
			}
            ...
		}
	}

    // kubelet传入podCache不为空
	if g.cacheEnabled() {
		// reinspect any pods that failed inspection during the previous relist
		if len(g.podsToReinspect) > 0 {
			// 重试更新podCache
			for pid, pod := range g.podsToReinspect {
				if err := g.updateCache(pod, pid); err != nil {
					// 记录到下次reList
					needsReinspection[pid] = pod
				}
			}
		}
	}

	// make sure we retain the list of pods that need reinspecting the next time relist is called
	g.podsToReinspect = needsReinspection
}

func computeEvents(oldPod, newPod *kubecontainer.Pod, cid *kubecontainer.ContainerID) []*PodLifecycleEvent {
	...
    // 获取container旧状态
	oldState := getContainerState(oldPod, cid)
    // 获取container新状态
	newState := getContainerState(newPod, cid)
    // 生成事件
	return generateEvents(pid, cid.ID, oldState, newState)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

# 2.2.Healthy

kubelet会将pleg.Healthy()作为检查方法注册到runtimeState，用于检查感知运行时状态，以影响syncPod调度。

// klet.runtimeState.addHealthCheck("PLEG", klet.pleg.Healthy)
// Healthy check if PLEG work properly. relistThreshold is the maximum interval between two relist.
func (g *GenericPLEG) Healthy() (bool, error) {
    // 上一次reList时间
	relistTime := g.getRelistTime()
	...
	// 距离当前时间是否超出3min
	elapsed := g.clock.Since(relistTime)
    // 超出,reList异常
	if elapsed > relistThreshold {
		return false, fmt.Errorf("pleg was last seen active %v ago; threshold is %v", elapsed, relistThreshold)
	}
	return true, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

kubelet在syncLoop()调度会定时调用runtimeState.runtimeErrors()进行健康检测。

func (kl *Kubelet) syncLoop(updates <-chan kubetypes.PodUpdate, handler SyncHandler) {
	...
	for {
        // 执行runtimeStateHandle
		if err := kl.runtimeState.runtimeErrors(); err != nil {
            // 休眠min(5s,2^n*100ms)
			time.Sleep(duration)
            // 更新挂起时间
			duration = time.Duration(math.Min(float64(max), factor*float64(duration)))
			continue
		}
		// reset backoff if we have a success
		duration = base

		kl.syncLoopMonitor.Store(kl.clock.Now())
		if !kl.syncLoopIteration(updates, handler, syncTicker.C, housekeepingTicker.C, plegCh) {
			break
		}
		kl.syncLoopMonitor.Store(kl.clock.Now())
	}
}

func (s *runtimeState) runtimeErrors() error {
	s.RLock()
	defer s.RUnlock()
	errs := []error{}
	...
	for _, hc := range s.healthChecks {
        // 执行Handle
		if ok, err := hc.fn(); !ok {
			errs = append(errs, fmt.Errorf("%s is not healthy: %v", hc.name, err))
		}
	}
	if s.runtimeError != nil {
		errs = append(errs, s.runtimeError)
	}

	return utilerrors.NewAggregate(errs)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

注意

pleg.Healthy()用于主循环进入syncLoopIteration调度pod前检查运行时是否正常