Kubelet | PLEG模块

 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

// PodLifeCycleEventType define the event type of pod life cycle events.
type PodLifeCycleEventType string

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 {
	Start()
	Watch() chan *PodLifecycleEvent
	Healthy() (bool, error)
}

1
2
3

func (g *GenericPLEG) Start() {
	go wait.Until(g.relist, g.relistPeriod, wait.NeverStop)
}

 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

func (g *GenericPLEG) relist() {
    // 获取当前时间
	timestamp := g.clock.Now()  
	// 调用container runtime获取所有pod
	podList, err := g.runtime.GetPods(true)
    // 将当前时间更新到GenericPLEG的relistTime字段，该字段类型是`atomic.Value`
	g.updateRelistTime(timestamp)
    // 将podList转换为kubecontainer.Pods类型
	pods := kubecontainer.Pods(podList)
    // 存到podRecords的current中
	g.podRecords.setCurrent(pods)
    // 比较old及current的pods，生成事件填充到eventsByPodID的map中
	eventsByPodID := map[types.UID][]*PodLifecycleEvent{}
	for pid := range g.podRecords {
		oldPod := g.podRecords.getOld(pid)
		pod := g.podRecords.getCurrent(pid)
		// 获取所有的container，内部使用了sets去重
		allContainers := getContainersFromPods(oldPod, pod)
		for _, container := range allContainers {
            // 根据pod等信息生成出事件，比如pod的state是`plegContainerRunning`，那么生成出的事件的类型即为`ContainerStarted`
			events := computeEvents(oldPod, pod, &container.ID)
			for _, e := range events {
				updateEvents(eventsByPodID, e)
			}
		}
	}
    // 创建需要重新探测的map映射
	var needsReinspection map[types.UID]*kubecontainer.Pod
	for pid, events := range eventsByPodID {
        // 从podRecords中获取所有的pod
		pod := g.podRecords.getCurrent(pid)
		if g.cacheEnabled() {
			// updateCache是执行探查的主逻辑
			if err := g.updateCache(pod, pid); err != nil {
                // 如果失败则将pod放到needsReinspection中
				needsReinspection[pid] = pod
				continue
			} else {
                // 如果成功则从podsToReinspect中删除该pod
				delete(g.podsToReinspect, pid)
			}
		}
		// Update the internal storage and send out the events.
		g.podRecords.update(pid)
		for i := range events {
			select {
               // 将产生的事件发送到eventChannel中，供外部使用
			case g.eventChannel <- events[i]:
			default:
                // 如果eventChannel已满，则将`pleg_discard_events`指标+1，并且产生一条错误
				metrics.PLEGDiscardEvents.Inc()
				klog.Error("event channel is full, discard this relist() cycle event")
			}
		}
	}
	if g.cacheEnabled() {
		// 如果podsToReinspect非空则遍历继续执行updateCache，如果失败仍然放到needsReinspection中
		if len(g.podsToReinspect) > 0 {
			for pid, pod := range g.podsToReinspect {
                // 对需要重新inspect的再次尝试
				if err := g.updateCache(pod, pid); err != nil {
					needsReinspection[pid] = pod
				}
			}
		}	
        // 更新缓存时间戳
		g.cache.UpdateTime(timestamp)
	}
	// 将needsReinspection赋值给genericpleg的podsToReinspect字段供下次循环使用
	g.podsToReinspect = needsReinspection
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

func (m *kubeGenericRuntimeManager) GetPodStatus(uid kubetypes.UID, name, namespace string) (*kubecontainer.PodStatus, error) {
	// 获取sandboxID
	podSandboxIDs, err := m.getSandboxIDByPodUID(uid, nil)
	sandboxStatuses := make([]*runtimeapi.PodSandboxStatus, len(podSandboxIDs))
	podIPs := []string{}
	for idx, podSandboxID := range podSandboxIDs {
        // 获取所有sandbox container的状态
		podSandboxStatus,_ := m.runtimeService.PodSandboxStatus(podSandboxID)
		sandboxStatuses[idx] = podSandboxStatus
	}
	// 获取所有container的状态
	containerStatuses, err := m.getPodContainerStatuses(uid, name, namespace)
	return &kubecontainer.PodStatus{
		ID:                uid,
		Name:              name,
		Namespace:         namespace,
		IPs:               podIPs,
		SandboxStatuses:   sandboxStatuses,
		ContainerStatuses: containerStatuses,
	}, nil
}

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

func (g *GenericPLEG) Healthy() (bool, error) {
	relistTime := g.getRelistTime()
	if relistTime.IsZero() {
		return false, fmt.Errorf("pleg has yet to be successful")
	}
	// Expose as metric so you can alert on `time()-pleg_last_seen_seconds > nn`
	metrics.PLEGLastSeen.Set(float64(relistTime.Unix()))
	elapsed := g.clock.Since(relistTime)
	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

// The threshold needs to be greater than the relisting period + the
// relisting time, which can vary significantly. Set a conservative
// threshold to avoid flipping between healthy and unhealthy.
relistThreshold = 3 * time.Minute

1
2
3

klet.pleg = pleg.NewGenericPLEG(klet.containerRuntime, plegChannelCapacity, plegRelistPeriod, klet.podCache, clock.RealClock{})
klet.runtimeState = newRuntimeState(maxWaitForContainerRuntime)
klet.runtimeState.addHealthCheck("PLEG", klet.pleg.Healthy)

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

func (s *runtimeState) runtimeErrors() error {
	for _, hc := range s.healthChecks {
		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
40
41

func (kl *Kubelet) syncLoop(updates <-chan kubetypes.PodUpdate, handler SyncHandler) {
	klog.Info("Starting kubelet main sync loop.")
	// The syncTicker wakes up kubelet to checks if there are any pod workers
	// that need to be sync'd. A one-second period is sufficient because the
	// sync interval is defaulted to 10s.
	syncTicker := time.NewTicker(time.Second)
	defer syncTicker.Stop()
	housekeepingTicker := time.NewTicker(housekeepingPeriod)
	defer housekeepingTicker.Stop()
	plegCh := kl.pleg.Watch()
	const (
		base   = 100 * time.Millisecond
		max    = 5 * time.Second
		factor = 2
	)
	duration := base
	// Responsible for checking limits in resolv.conf
	// The limits do not have anything to do with individual pods
	// Since this is called in syncLoop, we don't need to call it anywhere else
	if kl.dnsConfigurer != nil && kl.dnsConfigurer.ResolverConfig != "" {
		kl.dnsConfigurer.CheckLimitsForResolvConf()
	}

	for {
		if err := kl.runtimeState.runtimeErrors(); err != nil {
			klog.Errorf("skipping pod synchronization - %v", err)
			// exponential backoff
			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())
	}
}

1
2
3
4
5
6
7
8
9

if aggregatedErr := errors.NewAggregate(errs); aggregatedErr != nil {
			newNodeReadyCondition = v1.NodeCondition{
				Type:              v1.NodeReady,
				Status:            v1.ConditionFalse,
				Reason:            "KubeletNotReady",
				Message:           aggregatedErr.Error(),
				LastHeartbeatTime: currentTime,
			}
		}

 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

readyConditionUpdated := false
needToRecordEvent := false
for i := range node.Status.Conditions {
	if node.Status.Conditions[i].Type == v1.NodeReady {
		if node.Status.Conditions[i].Status == newNodeReadyCondition.Status {
			newNodeReadyCondition.LastTransitionTime = node.Status.Conditions[i].LastTransitionTime
		} else {
			newNodeReadyCondition.LastTransitionTime = currentTime
			needToRecordEvent = true
		}
		node.Status.Conditions[i] = newNodeReadyCondition
		readyConditionUpdated = true
		break
	}
}
if !readyConditionUpdated {
	newNodeReadyCondition.LastTransitionTime = currentTime
	node.Status.Conditions = append(node.Status.Conditions, newNodeReadyCondition)
}
if needToRecordEvent {
	if newNodeReadyCondition.Status == v1.ConditionTrue {
		recordEventFunc(v1.EventTypeNormal, events.NodeReady)
	} else {
		recordEventFunc(v1.EventTypeNormal, events.NodeNotReady)
		klog.Infof("Node became not ready: %+v", newNodeReadyCondition)
	}
}
return nil

 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

conditions:
  - lastHeartbeatTime: "2024-06-26T00:22:18Z"
    lastTransitionTime: "2024-06-25T06:14:04Z"
    message: kubelet has sufficient memory available
    reason: KubeletHasSufficientMemory
    status: "False"
    type: MemoryPressure
  - lastHeartbeatTime: "2024-06-26T00:22:18Z"
    lastTransitionTime: "2024-06-25T06:14:04Z"
    message: kubelet has no disk pressure
    reason: KubeletHasNoDiskPressure
    status: "False"
    type: DiskPressure
  - lastHeartbeatTime: "2024-06-26T00:22:18Z"
    lastTransitionTime: "2024-06-25T06:14:04Z"
    message: kubelet has sufficient PID available
    reason: KubeletHasSufficientPID
    status: "False"
    type: PIDPressure
  - lastHeartbeatTime: "2024-06-26T00:22:18Z"
    lastTransitionTime: "2024-06-25T06:14:04Z"
    message: kubelet is posting ready status
    reason: KubeletReady
    status: "True"
    type: Ready

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

var setters []func(n *v1.Node) error

func (kl *Kubelet) defaultNodeStatusFuncs() []func(*v1.Node) error {
	...
	setters = append(setters,
		nodestatus.MemoryPressureCondition(kl.clock.Now, kl.evictionManager.IsUnderMemoryPressure, kl.recordNodeStatusEvent),
		nodestatus.DiskPressureCondition(kl.clock.Now, kl.evictionManager.IsUnderDiskPressure, kl.recordNodeStatusEvent),
		nodestatus.PIDPressureCondition(kl.clock.Now, kl.evictionManager.IsUnderPIDPressure, kl.recordNodeStatusEvent),
		nodestatus.ReadyCondition(kl.clock.Now, kl.runtimeState.runtimeErrors, kl.runtimeState.networkErrors, kl.runtimeState.storageErrors, validateHostFunc, kl.containerManager.Status, kl.shutdownManager.ShutdownStatus, kl.recordNodeStatusEvent),
		nodestatus.VolumesInUse(kl.volumeManager.ReconcilerStatesHasBeenSynced, kl.volumeManager.GetVolumesInUse),
		kl.recordNodeSchedulableEvent,
	)
	return setters
}

1

klet.setNodeStatusFuncs = klet.defaultNodeStatusFuncs()

1
2
3
4
5
6
7
8

func (kl *Kubelet) setNodeStatus(node *v1.Node) {
	for i, f := range kl.setNodeStatusFuncs {
		klog.V(5).Infof("Setting node status at position %v", i)
		if err := f(node); err != nil {
			klog.Errorf("Failed to set some node status fields: %s", err)
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

go wait.Until(kl.syncNodeStatus, kl.nodeStatusUpdateFrequency, wait.NeverStop)
      |
      |
kl.updateNodeStatus() 
      | 
      | 
kl.tryUpdateNodeStatus() [重试5次调用tryUpdateNodeStatus]
      |
      |
kl.setNodeStatus(node)

1
2
3

if obj.NodeStatusUpdateFrequency == zeroDuration {
	obj.NodeStatusUpdateFrequency = metav1.Duration{Duration: 10 * time.Second}
}

 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

// GetContainerEvents gets container events from the CRI runtime
rpc  GetContainerEvents(GetEventsRequest) returns (stream ContainerEventResponse) {}

message GetEventsRequest {}

message ContainerEventResponse {
    // ID of the container
    string container_id = 1;
    // Type of the container event
    ContainerEventType container_event_type = 2;
    // Creation timestamp of this event
    int64 created_at = 3;
    // ID of the sandbox container
    PodSandboxMetadata pod_sandbox_metadata = 4;
}

enum ContainerEventType {
    // Container created
    CONTAINER_CREATED_EVENT = 0;
    // Container started
    CONTAINER_STARTED_EVENT = 1;
    // Container stopped
    CONTAINER_STOPPED_EVENT = 2;
    // Container deleted
    CONTAINER_DELETED_EVENT = 3;
}

 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

klet.podCache = kubecontainer.NewCache()
...
if utilfeature.DefaultFeatureGate.Enabled(features.EventedPLEG) {
	// adjust Generic PLEG relisting period and threshold to higher value when Evented PLEG is turned on
	genericRelistDuration := &pleg.RelistDuration{
		RelistPeriod:    eventedPlegRelistPeriod, // 300s
		RelistThreshold: eventedPlegRelistThreshold, // 10m
	}
	klet.pleg = pleg.NewGenericPLEG(klet.containerRuntime, eventChannel, genericRelistDuration, klet.podCache, clock.RealClock{})
	// In case Evented PLEG has to fall back on Generic PLEG due to an error,
	// Evented PLEG should be able to reset the Generic PLEG relisting duration
	// to the default value.
	eventedRelistDuration := &pleg.RelistDuration{
		RelistPeriod:    genericPlegRelistPeriod, // 1s
		RelistThreshold: genericPlegRelistThreshold, // 3m
	}
	klet.eventedPleg = pleg.NewEventedPLEG(klet.containerRuntime, klet.runtimeService, eventChannel,
		klet.podCache, klet.pleg, eventedPlegMaxStreamRetries, eventedRelistDuration, clock.RealClock{})
} else {
	genericRelistDuration := &pleg.RelistDuration{
		RelistPeriod:    genericPlegRelistPeriod,
		RelistThreshold: genericPlegRelistThreshold,
	}
	klet.pleg = pleg.NewGenericPLEG(klet.containerRuntime, eventChannel, genericRelistDuration, klet.podCache, clock.RealClock{})
}

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

// Start starts the Evented PLEG
func (e *EventedPLEG) Start() {
	e.runningMu.Lock()
	defer e.runningMu.Unlock()
	if isEventedPLEGInUse() {
		return
	}
	setEventedPLEGUsage(true)
	e.stopCh = make(chan struct{})
	e.stopCacheUpdateCh = make(chan struct{})
	go wait.Until(e.watchEventsChannel, 0, e.stopCh)
	go wait.Until(e.updateGlobalCache, globalCacheUpdatePeriod, e.stopCacheUpdateCh)
}

1
2
3
4
5
6
7
8

// In case the Evented PLEG experiences undetectable issues in the underlying
// GRPC connection there is a remote chance the pod might get stuck in a
// given state while it has progressed in its life cycle. This function will be
// called periodically to update the global timestamp of the cache so that those
// pods stuck at GetNewerThan in pod workers will get unstuck.
func (e *EventedPLEG) updateGlobalCache() {
	e.cache.UpdateTime(time.Now())
}

 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

func (e *EventedPLEG) watchEventsChannel() {
	containerEventsResponseCh := make(chan *runtimeapi.ContainerEventResponse, cap(e.eventChannel))
	defer close(containerEventsResponseCh)

	// Get the container events from the runtime.
	go func() {
		numAttempts := 0
		for {
			if numAttempts >= e.eventedPlegMaxStreamRetries {
				if isEventedPLEGInUse() {
					// Fall back to Generic PLEG relisting since Evented PLEG is not working.
					klog.V(4).InfoS("Fall back to Generic PLEG relisting since Evented PLEG is not working")
					e.Stop()
					e.genericPleg.Stop()       // Stop the existing Generic PLEG which runs with longer relisting period when Evented PLEG is in use.
					e.Update(e.relistDuration) // Update the relisting period to the default value for the Generic PLEG.
					e.genericPleg.Start()
					break
				}
			}

			err := e.runtimeService.GetContainerEvents(containerEventsResponseCh)
			if err != nil {
				metrics.EventedPLEGConnErr.Inc()
				numAttempts++
				e.Relist() // Force a relist to get the latest container and pods running metric.
				klog.V(4).InfoS("Evented PLEG: Failed to get container events, retrying: ", "err", err)
			}
		}
	}()

	if isEventedPLEGInUse() {
		e.processCRIEvents(containerEventsResponseCh)
	}
}

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

// Set sets the PodStatus for the pod only if the data is newer than the cache
func (c *cache) Set(id types.UID, status *PodStatus, err error, timestamp time.Time) (updated bool) {
	c.lock.Lock()
	defer c.lock.Unlock()

	if utilfeature.DefaultFeatureGate.Enabled(features.EventedPLEG) {
		// Set the value in the cache only if it's not present already
		// or the timestamp in the cache is older than the current update timestamp
		if cachedVal, ok := c.pods[id]; ok && cachedVal.modified.After(timestamp) {
			return false
		}
	}

	c.pods[id] = &data{status: status, err: err, modified: timestamp}
	c.notify(id, timestamp)
	return true
}