Kubernetes源码分析之kubelet

本节全部的代码基于1.13.4版本。

启动分析

Kubelet的启动参数有两种，kubeletFlags和kubeletConfig。其中，kubeletFlags与咱们使用的kubelet的--参数命令保持一致；kubeletConfig经过解析特定的配置文件完成参数的配置，它们共同构成kubelet启动参数的配置。如图

基本参数配置完成以后，接下来就是配置启动的 Run方法。Kubelet启动的Run方法代较长，以下

Run: func(cmd *cobra.Command, args []string) {
			// initial flag parse, since we disable cobra's flag parsing
			if err := cleanFlagSet.Parse(args); err != nil {
				cmd.Usage()
				klog.Fatal(err)
			}

			// check if there are non-flag arguments in the command line
			cmds := cleanFlagSet.Args()
			if len(cmds) > 0 {
				cmd.Usage()
				klog.Fatalf("unknown command: %s", cmds[0])
			}

			// short-circuit on help
			help, err := cleanFlagSet.GetBool("help")
			if err != nil {
				klog.Fatal(`"help" flag is non-bool, programmer error, please correct`)
			}
			if help {
				cmd.Help()
				return
			}

			// short-circuit on verflag
			verflag.PrintAndExitIfRequested()
			utilflag.PrintFlags(cleanFlagSet)

			// set feature gates from initial flags-based config
			if err := utilfeature.DefaultFeatureGate.SetFromMap(kubeletConfig.FeatureGates); err != nil {
				klog.Fatal(err)
			}

			// validate the initial KubeletFlags
			if err := options.ValidateKubeletFlags(kubeletFlags); err != nil {
				klog.Fatal(err)
			}

			if kubeletFlags.ContainerRuntime == "remote" && cleanFlagSet.Changed("pod-infra-container-image") {
				klog.Warning("Warning: For remote container runtime, --pod-infra-container-image is ignored in kubelet, which should be set in that remote runtime instead")
			}

			// load kubelet config file, if provided
			if configFile := kubeletFlags.KubeletConfigFile; len(configFile) > 0 {
				kubeletConfig, err = loadConfigFile(configFile)
				if err != nil {
					klog.Fatal(err)
				}
				// We must enforce flag precedence by re-parsing the command line into the new object.
				// This is necessary to preserve backwards-compatibility across binary upgrades.
				// See issue #56171 for more details.
				if err := kubeletConfigFlagPrecedence(kubeletConfig, args); err != nil {
					klog.Fatal(err)
				}
				// update feature gates based on new config
				if err := utilfeature.DefaultFeatureGate.SetFromMap(kubeletConfig.FeatureGates); err != nil {
					klog.Fatal(err)
				}
			}

			// We always validate the local configuration (command line + config file).
			// This is the default "last-known-good" config for dynamic config, and must always remain valid.
			if err := kubeletconfigvalidation.ValidateKubeletConfiguration(kubeletConfig); err != nil {
				klog.Fatal(err)
			}

			// use dynamic kubelet config, if enabled
			var kubeletConfigController *dynamickubeletconfig.Controller
			if dynamicConfigDir := kubeletFlags.DynamicConfigDir.Value(); len(dynamicConfigDir) > 0 {
				var dynamicKubeletConfig *kubeletconfiginternal.KubeletConfiguration
				dynamicKubeletConfig, kubeletConfigController, err = BootstrapKubeletConfigController(dynamicConfigDir,
					func(kc *kubeletconfiginternal.KubeletConfiguration) error {
						// Here, we enforce flag precedence inside the controller, prior to the controller's validation sequence,
						// so that we get a complete validation at the same point where we can decide to reject dynamic config.
						// This fixes the flag-precedence component of issue #63305.
						// See issue #56171 for general details on flag precedence.
						return kubeletConfigFlagPrecedence(kc, args)
					})
				if err != nil {
					klog.Fatal(err)
				}
				// If we should just use our existing, local config, the controller will return a nil config
				if dynamicKubeletConfig != nil {
					kubeletConfig = dynamicKubeletConfig
					// Note: flag precedence was already enforced in the controller, prior to validation,
					// by our above transform function. Now we simply update feature gates from the new config.
					if err := utilfeature.DefaultFeatureGate.SetFromMap(kubeletConfig.FeatureGates); err != nil {
						klog.Fatal(err)
					}
				}
			}

			// construct a KubeletServer from kubeletFlags and kubeletConfig
			kubeletServer := &options.KubeletServer{
				KubeletFlags:         *kubeletFlags,
				KubeletConfiguration: *kubeletConfig,
			}

			// use kubeletServer to construct the default KubeletDeps
			kubeletDeps, err := UnsecuredDependencies(kubeletServer)
			if err != nil {
				klog.Fatal(err)
			}

			// add the kubelet config controller to kubeletDeps
			kubeletDeps.KubeletConfigController = kubeletConfigController

			// start the experimental docker shim, if enabled
			if kubeletServer.KubeletFlags.ExperimentalDockershim {
				if err := RunDockershim(&kubeletServer.KubeletFlags, kubeletConfig, stopCh); err != nil {
					klog.Fatal(err)
				}
				return
			}

			// run the kubelet
			klog.V(5).Infof("KubeletConfiguration: %#v", kubeletServer.KubeletConfiguration)
			if err := Run(kubeletServer, kubeletDeps, stopCh); err != nil {
				klog.Fatal(err)
			}
复制代码

主要包括如下步骤：
一、解析参数，对参数的合法性进行判断；
二、根据kubeletConfig解析一些特殊的特性所须要配置的参数；
三、配置kubeletServer，包括KubeletFlags和KubeletConfiguration两个参数；
四、构造kubeletDeps结构体；
五、启动最终的Run方法。
除了最终的Run方法，其他的步骤仍是为kubelet的启动构建初始化的参数，无非就是换一个名称，换一个不一样的结构体，并配置相依赖的参数。

启动

启动调用的是Run方法，如图

initForOS经过对操做系统的判断，若是是windows系统须要作一些预先的特殊处理； run方法即经过传入的 kubeDeps参数开始执行启动操做。
进入 run方法，开始主要执行对参数的再一次验证，以及新的结构体的初始化。后续开始构建一些重要的客户端，包括 eventClient主要处理事件的上报，与apiserver打交道； heartbeatClient主要处理心跳操做，与以后的PLEG相关； csiClient主要与CSI接口相关。配置完成以后，最终进入 RunKubelet方法。
RunKubelet方法最重要的方法有两个： CreateAndInitKubelet和 startKubelet，能够理解为 CreateAndInitKubelet为参数的配置， startKubelet为最终的启动（说来讲去仍是把参数封装一遍，从新构造新的结构体）。
CreateAndInitKubelet方法经过调用 NewMainKubelet返回 Kubelet结构体。在 NewMainKubelet中，主要的配置有：
一、PodConfig。经过 makePodSourceConfig能够发现kubelet获取Pod的来源有如下途径： 静态Pod、 静态Pod的URL地址以及 kube-apiserver；
二、容器与镜像的GC参数。
三、驱逐Pod策略。
最终经过参数填充 Kubelet结构体，完成kubelet结构体参数的最终配置。
接下来就是启动了，不过在启动以前会有一个判断

判断是之后台daemon进程一直运行仍是只启动一次，即runOnce，基本上都是之后台daemon启动的方式，因此大部分调用的是 startKubelet方法。
startKubelet方法内部调用了最终的 Run方法，以下

func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) {
	if kl.logServer == nil {
		kl.logServer = http.StripPrefix("/logs/", http.FileServer(http.Dir("/var/log/")))
	}
	if kl.kubeClient == nil {
		klog.Warning("No api server defined - no node status update will be sent.")
	}

	// Start the cloud provider sync manager
	if kl.cloudResourceSyncManager != nil {
		go kl.cloudResourceSyncManager.Run(wait.NeverStop)
	}

	if err := kl.initializeModules(); err != nil {
		kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.KubeletSetupFailed, err.Error())
		klog.Fatal(err)
	}

	// Start volume manager
	go kl.volumeManager.Run(kl.sourcesReady, wait.NeverStop)

	if kl.kubeClient != nil {
		// Start syncing node status immediately, this may set up things the runtime needs to run.
		go wait.Until(kl.syncNodeStatus, kl.nodeStatusUpdateFrequency, wait.NeverStop)
		go kl.fastStatusUpdateOnce()

		// start syncing lease
		if utilfeature.DefaultFeatureGate.Enabled(features.NodeLease) {
			go kl.nodeLeaseController.Run(wait.NeverStop)
		}
	}
	go wait.Until(kl.updateRuntimeUp, 5*time.Second, wait.NeverStop)

	// Start loop to sync iptables util rules
	if kl.makeIPTablesUtilChains {
		go wait.Until(kl.syncNetworkUtil, 1*time.Minute, wait.NeverStop)
	}

	// Start a goroutine responsible for killing pods (that are not properly
	// handled by pod workers).
	go wait.Until(kl.podKiller, 1*time.Second, wait.NeverStop)

	// Start component sync loops.
	kl.statusManager.Start()
	kl.probeManager.Start()

	// Start syncing RuntimeClasses if enabled.
	if kl.runtimeClassManager != nil {
		go kl.runtimeClassManager.Run(wait.NeverStop)
	}

	// Start the pod lifecycle event generator.
	kl.pleg.Start()
	kl.syncLoop(updates, kl)
}
复制代码

能够看到，在该方法内，完成的就是最终的kubelet的任务，经过多个goroutine完成。包括如下系列：
一、volumeManager，volume相关管理；
二、syncNodeStatus，定时同步Node状态；
三、updateRuntimeUp，定时更新Runtime状态；
四、syncNetworkUtil，定时同步网络状态；
五、podKiller，定时清理死亡的pod；
六、statusManager，pod状态管理；
七、probeManager，pod探针管理；
八、启动PLEG；
九、syncLoop，最重要的主进程，不停监听外部数据的变化执行pod的相应操做。
至此，kubelet启动过程完成。启动主要完成的任务就是参数的配置和多个任务的启动，经过构造一个循环进程不停监听外部事件的变化，执行对应的pod处理工做，这也就是kubelet所须要负责的任务。

Pod启动流程

Pod的启动在syncLoop方法下调用的syncLoopIteration方法开始。在syncLoopIteration方法内，有5个重要的参数

一、configCh：获取Pod信息的channel，关于Pod相关的事件都从该channel获取；
二、handler：处理Pod的handler；
三、syncCh：同步全部等待同步的Pod；
四、houseKeepingCh：清理Pod的channel；
五、plegCh：获取PLEG信息，同步Pod。
每一个参数都是一个channel，经过select判断某个channel获取到信息，处理相应的操做。Pod的启动显然与configCh相关。
经过获取configCh信息，获取Pod整个生命周期中的多种状态

相对应的，每一个状态对应相应的处理方法

其中， ADD操做对应Pod的建立，其对应的处理方法为 HandlePodAdditions。
进入 HandlePodAdditions方法，主要如下几个步骤：
一、根据Pod的建立时间对Pod进行排序；
二、podManager添加Pod；（对Pod的管理依赖于podManager）
三、处理mirrorPod，即静态Pod的处理；
四、经过 dispatchWork方法分发任务，处理Pod的建立；
五、probeManager添加Pod。（readiness和liveness探针）
dispatchWork方法内，最核心的是调用了 kl.podWorkers.UpdatePod方法对Pod进行建立。 UpdatePod方法经过 podUpdates的map类型获取相对应的Pod，map的key为Pod的UID，value为 UpdatePodOptions的结构体channel。经过获取到须要建立的Pod以后，单独起一个goroutine调用 managePodLoop方法完成Pod的建立， managePodLoop方法最终调用 syncPodFn完成Pod的建立， syncPodFn对应的就是Kubelet的 syncPod方法，位于 kubernetes/pkg/kubelet/kubelet.go下。通过层层环绕， syncPod就是最终处理Pod建立的方法。
syncPod主要的工做流如注释

一、更新Pod的状态，对应 generateAPIPodStatus和 statusManager.SetPodStatus方法；
二、建立Pod存储的目录，对应 makePodDataDirs方法；
三、挂载对应的volume，对应 volumeManager.WaitForAttachAndMount方法；
四、获取ImagePullSecrets，对应 getPullSecretsForPod方法；
五、建立容器，对应 containerRuntime.SyncPod方法，以下

至此，Pod的启动到建立过程完成。经过 kubectl describe pod命令能够查看Pod建立的整个生命周期。

PLEG分析

PLEG，即PodLifecycleEventGenerator，用来记录Pod生命周期中对应的各类事件。在kubelet中，启动主进程的syncLoop以前，先启动pleg，如图

Start方法经过启动一个定时的任务执行 relist方法

relist主要的工做就是经过比对Pod的原始状态和如今的状态，判断Pod当前所处的生命周期，核心代码以下

对每个Pod，比对Pod内的容器，经过 computeEvents-->generateEvents生成事件。在 generateEvents内，生成如下事件：
一、newState为 plegContainerRunning，对应 ContainerStarted事件；
二、newState为 plegContainerExited，对应 ContainerDied事件；
三、newState为 plegContainerUnknown，对应 ContainerChanged事件；
四、newState为 plegContainerNonExistent，查找oldState，若是对应 plegContainerExited，则生成的事件为 ContainerRemoved，不然事件为 ContainerDied和 ContainerRemoved；

生成完事件以后，将事件一一通知到 eventChannel，该channel对应的就是 syncLoopIteration方法下的plegCh

在 syncLoopIteration方法下，接收到plegCh channel传输过来的消息以后，执行 HandlePodSyncs同步方法，最终调用到 dispatchWork这个Pod的处理方法，对Pod的生命进行管理。

GC管理

Kubelet会定时去清理多余的container和image，完成ContainerGC和ImageGC。Kubelet在启动的Run方法里，会先去调用imageManager的Start方法，代码位于kubernetes/pkg/kubelet/kubelet.go下，调用了initializeModules方法。imageManager.Start方法主要执行两个步骤：
一、detectImages：主要用来监控images，判断镜像是可被发现的；
二、ListImages：主要用来获取镜像信息，写入到缓存imageCache中。
在启动的CreateAndInitKubelet方法中，开始执行镜像与容器的回收

StartGarbageCollection方法启用两个goroutine，一个用来作ContainerGC，一个用来作ImageGC，代码以下

func (kl *Kubelet) StartGarbageCollection() {
	loggedContainerGCFailure := false
	go wait.Until(func() {
		if err := kl.containerGC.GarbageCollect(); err != nil {
			klog.Errorf("Container garbage collection failed: %v", err)
			kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.ContainerGCFailed, err.Error())
			loggedContainerGCFailure = true
		} else {
			var vLevel klog.Level = 4
			if loggedContainerGCFailure {
				vLevel = 1
				loggedContainerGCFailure = false
			}

			klog.V(vLevel).Infof("Container garbage collection succeeded")
		}
	}, ContainerGCPeriod, wait.NeverStop)

	// when the high threshold is set to 100, stub the image GC manager
	if kl.kubeletConfiguration.ImageGCHighThresholdPercent == 100 {
		klog.V(2).Infof("ImageGCHighThresholdPercent is set 100, Disable image GC")
		return
	}

	prevImageGCFailed := false
	go wait.Until(func() {
		if err := kl.imageManager.GarbageCollect(); err != nil {
			if prevImageGCFailed {
				klog.Errorf("Image garbage collection failed multiple times in a row: %v", err)
				// Only create an event for repeated failures
				kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.ImageGCFailed, err.Error())
			} else {
				klog.Errorf("Image garbage collection failed once. Stats initialization may not have completed yet: %v", err)
			}
			prevImageGCFailed = true
		} else {
			var vLevel klog.Level = 4
			if prevImageGCFailed {
				vLevel = 1
				prevImageGCFailed = false
			}

			klog.V(vLevel).Infof("Image garbage collection succeeded")
		}
	}, ImageGCPeriod, wait.NeverStop)
}
复制代码

能够看到容器的GC默认是每分钟执行一次，镜像的GC默认是每5分钟执行一次，经过定时执行GC的清理完成容器与镜像的回收。
容器的GC主要完成的任务包括删除被驱除的容器、删除sandboxes以及清理Pod的sandbox的日志目录，代码位于kubernetes/pkg/kubelet/kuberuntime/kuberuntime_gc.go下，调用了GarbageCollect方法；镜像的GC主要完成多余镜像的删除和存储空间的释放，代码位于kubernetes/pkg/kubelet/images/image_gc_manager.go下，调用了GarbageCollect方法。