kubernetes 之 kubelet 深入浅出

 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

func (kl *Kubelet) syncLoopIteration(configCh <-chan kubetypes.PodUpdate, handler SyncHandler,
	syncCh <-chan time.Time, housekeepingCh <-chan time.Time, plegCh <-chan *pleg.PodLifecycleEvent) bool {
	select {
	case u, open := <-configCh:
		switch u.Op {
		case kubetypes.ADD:
			handler.HandlePodAdditions(u.Pods)
		case kubetypes.UPDATE:
		case kubetypes.REMOVE:
		case kubetypes.RECONCILE:
		case kubetypes.DELETE:
		case kubetypes.SET:
		default:
		}
		kl.sourcesReady.AddSource(u.Source)
	case e := <-plegCh:
		...
	case <-syncCh:
		...
	case update := <-kl.livenessManager.Updates():
		...
	case <-housekeepingCh:
        ...
	}
	return true
}

1
2
3
4
5
6
7
8

type SyncHandler interface {
	HandlePodAdditions(pods []*v1.Pod)
	HandlePodUpdates(pods []*v1.Pod)
	HandlePodRemoves(pods []*v1.Pod)
	HandlePodReconcile(pods []*v1.Pod)
	HandlePodSyncs(pods []*v1.Pod)
	HandlePodCleanups() error
}

1

HandlePodAdditions -> dispatchWork -> podWorkers

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

func (kl *Kubelet) dispatchWork(pod *v1.Pod, syncType kubetypes.SyncPodType, mirrorPod *v1.Pod, start time.Time) {
    ...
    // Run the sync in an async worker.
	kl.podWorkers.UpdatePod(&UpdatePodOptions{
		Pod:        pod,
		MirrorPod:  mirrorPod,
		UpdateType: syncType,
		OnCompleteFunc: func(err error) {
			...
		},
	})
    ...
}

 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

func (p *podWorkers) UpdatePod(options *UpdatePodOptions) {
	pod := options.Pod
	uid := pod.UID
	var podUpdates chan UpdatePodOptions
	var exists bool

	p.podLock.Lock()
	defer p.podLock.Unlock()
	if podUpdates, exists = p.podUpdates[uid]; !exists {
        // 如果当前 pod 还没有启动过 goroutine ，则启动 goroutine，并且创建 channel

		podUpdates = make(chan UpdatePodOptions, 1)
		p.podUpdates[uid] = podUpdates

		go func() {
			defer runtime.HandleCrash()
			p.managePodLoop(podUpdates)
		}()
	}
	if !p.isWorking[pod.UID] {
		p.isWorking[pod.UID] = true
		podUpdates <- *options
	} else {
        // 如果一个杀死 Pod 的请求正在等待，我们不会让任何东西覆盖该请求
		update, found := p.lastUndeliveredWorkUpdate[pod.UID]
		if !found || update.UpdateType != kubetypes.SyncPodKill {
			p.lastUndeliveredWorkUpdate[pod.UID] = *options
		}
	}
}

 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

// pkg/kubelet/kubelet.go:1388
func (kl *Kubelet) syncPod(o syncPodOptions) error {
	...
	// 如果是删除 pod，立即执行并返回。
	if updateType == kubetypes.SyncPodKill {
		...
		if err := kl.killPod(pod, nil, podStatus, killPodOptions.PodTerminationGracePeriodSecondsOverride); err != nil {
			...
			return err
		}
		return nil
	}

	...

	runnable := kl.canRunPod(pod)
	// 检查 Pod 是否能运行在本节点，主要是权限检查（是否能使用主机网络模式，是否可以以 privileged 权限运行等）。如果没有权限，就删除本地旧的 Pod 并返回错误信息；

	// 如果是 static Pod，就创建或者更新对应的 mirrorPod，如果存在先删除
	if kubetypes.IsStaticPod(pod) {
		...
		if err := kl.podManager.CreateMirrorPod(pod); err != nil {
			...
		}
		...
	}

	// 创建 Pod 的数据目录，存放 volume 和 plugin 信息，即目录 /var/lib/kubelet/uuid
	if err := kl.makePodDataDirs(pod); err != nil {
		...
		return err
	}

	// 如果定义了 PV，等待所有该 Pod 依赖的 Volume mount 完成（volumeManager 会在后台做这些事情）
	if !kl.podIsTerminated(pod) {
		if err := kl.volumeManager.WaitForAttachAndMount(pod); err != nil {
			...
			return err
		}
	}

	// 如果有 Image secrets，去 apiserver 获取对应的 Secrets 数据
	pullSecrets := kl.getPullSecretsForPod(pod)

	// 调用容器运行时的 SyncPod 回调
	result := kl.containerRuntime.SyncPod(pod, podStatus, pullSecrets, kl.backOff)
	...

	return nil
}