kubelet 源码
进入 cmd/kubelet/app/server.go 调用 NewKubeletCommand 函数，使用默认的参数创建一个 cobra.Command 指针：
func NewKubeletCommand() *cobra.Command {
...
cmd := &cobra.Command{
Use: componentKubelet,
...
// Kubelet具有特殊的flag解析要求来强制执行flag优先级规则，因此我们在下面的Run中手动完成所有解析。DisableFlagParsing=true提供传递给`args` arg中的kubelet的完整flagset来运行，没有Cobra的干扰。
DisableFlagParsing: true,
Run: func(cmd *cobra.Command, args []string) {
// 初始化flag解析，因为我们禁用了cobra的flag解析
if err := cleanFlagSet.Parse(args); err != nil {
cmd.Usage()
klog.Fatal(err)
}
......
// 校验初始的 KubeletFlags
// 加载 kubelet config 文件(如果开启)
// 使用动态的 kubelet config(如果开启)
// 通过kubeletFlags和kubeletConfig封装KubeletServer结构
// 使用 kubeletServer 来构造默认的 KubeletDeps
// 添加 kubelet 配置控制器到 kubeletDeps
// 添加 stopCh Channel 供 kubelet 和 docker shim 复用
// 开启docker shim特性(如果开启)
if kubeletServer.KubeletFlags.ExperimentalDockershim {
if err := RunDockershim(&kubeletServer.KubeletFlags, kubeletConfig, stopCh); err != nil {
klog.Fatal(err)
}
// 注意这里的return，意味着启用 dockershim only 模式，kubelet只会启动 dockershim 进程。此标志仅用于测试目的，除非您意识到自己在做什么，否则请不要使用它。 [默认= FALSE]
return
}
// 运行 kubelet
if err := Run(kubeletServer, kubeletDeps, stopCh); err != nil {
klog.Fatal(err)
}
}
}
...
return cmd
}其中 RunDockershim 函数仅在当前进程中启动 dockershim。这仅用于验证测试目的。
func RunDockershim(f *options.KubeletFlags, c *kubeletconfiginternal.KubeletConfiguration, stopCh <-chan struct{}) error {
// 初始化 docker 客户端配置
// 初始化网络插件配置
// 初始化 streaming 配置（现在不使用TLS）
// 独立的 dockershim 将始终会启动本地的streaming server。
ds, err := dockershim.NewDockerService(dockerClientConfig, r.PodSandboxImage, streamingConfig, &pluginSettings,
f.RuntimeCgroups, c.CgroupDriver, r.DockershimRootDirectory, true /*startLocalStreamingServer*/)
if err != nil {
return err
}
// 为 docker CRI shim 启动 GRPC 服务器
server := dockerremote.NewDockerServer(f.RemoteRuntimeEndpoint, ds)
if err := server.Start(); err != nil {
return err
}
<-stopCh
return nil
}然后是最核心的 kubelet 启动函数 Run：
// Run 函数运行一个具有给定依赖项的 KubeletServer，永远不会退出。
// kubeDeps 参数可能为 nil - 如果为空，则需要从 KubeletServer 上的设置进行初始化。
// 如果调用者已经设置了依赖对象了则不会生成默认的对象了。
func Run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chan struct{}) error {
if err := initForOS(s.KubeletFlags.WindowsService); err != nil {
return fmt.Errorf("failed OS init: %v", err)
}
if err := run(s, kubeDeps, stopCh); err != nil {
return fmt.Errorf("failed to run Kubelet: %v", err)
}
return nil
}调用下面的核心的 run 函数：
func run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chan struct{}) (err error) {
// 基于初始的KubeletServer的值设置全局的 feature gates 特性
// 校验初始KubeletServer（我们首先设置feature gates，因为该验证依赖feature gates）
if err := options.ValidateKubeletServer(s); err != nil {
return err
}
// 获取 Kubelet 锁文件
// 使用 /configz 端口注册当前配置
// 获取客户端等信息，检测独立(standaloneMode)模式
// 获取主机名、获取节点名
// 如果是独立模式，尽可能将所有客户端设置成nil
// 否则
// 创建 kubelet 客户端
// 为事件创建一个独立的客户端
// 为禁用限流和附加超时的心跳创建一个单独的客户端
// 如果启用了NodeLease功能，则超时是租约持续时间和状态更新频率的最小值
// NodeLease: Kubelet 使用新的 Lease API 报告节点心跳，节点生命周期控制器使用这些心跳作为节点健康信号。
// 如果 kubelet 配置控制器可用，并且启用了动态配置，则启动配置和状态同步循环
// DynamicConfigDir:
// Kubelet 将使用此目录检查下载的配置并跟踪配置运行状况。
// Kubelet 将创建此目录（如果该目录尚不存在）, 路径可以是绝对的或相对的; 相对路径位于 Kubelet 的当前工作目录下。
// 提供此标志可启用动态 kubelet 配置，要使用此标志，必须启用DynamicKubeletConfig feature gates。
if utilfeature.DefaultFeatureGate.Enabled(features.DynamicKubeletConfig) && len(s.DynamicConfigDir.Value()) > 0 &&
kubeDeps.KubeletConfigController != nil && !standaloneMode && !s.RunOnce {
// StartSync 函数告诉控制器开一个goroutine从apiserver同步status/config 到 /（位于pkg/kubelt/kubeletconfig/controller.go）
if err := kubeDeps.KubeletConfigController.StartSync(kubeDeps.KubeClient, kubeDeps.EventClient, string(nodeName)); err != nil {
return err
}
}
// 然后调用 BuildAuth 函数创建一个验证器，一个授权器，以及一个 kubelet的authorizer.RequestAttributesGetter
// 安装事件记录器
// 解析资源预留配置
// 构造 ContainerManager
// 然后调用 RunKubelet 函数
if err := RunKubelet(s, kubeDeps, s.RunOnce); err != nil {
return err
}
// 如果开启了 healthzPort 端口（本地 healthz endpint 端口，0表示禁用）,则开启 healthz server
// 如果使用了 systemd , 通知它我们已经启动了
go daemon.SdNotify(false, "READY=1")
select {
case <-done:
break
case <-stopCh:
break
}
return nil
}其中校验 KubeletServer 的函数 ValidateKubeletServer 位于cmd/kubelet/app/options/options.go：
// ValidateKubeletServer 函数校验 KubeletServer 的配置，如果配置无效则返回error。
func ValidateKubeletServer(s *KubeletServer) error {
// 任何的 KubeletConfiguration 校验都使用kubeletconfigvalidation.ValidateKubeletConfiguration函数
// ValidateKubeletConfiguration 函数位于 pkg/kubelet/apis/config/validation/validation.go
if err := kubeletconfigvalidation.ValidateKubeletConfiguration(&s.KubeletConfiguration); err != nil {
return err
}
// ValidateKubeletFlags 函数校验 Kubelet 的配置 flag，如果无效则返回 error
if err := ValidateKubeletFlags(&s.KubeletFlags); err != nil {
return err
}
return nil
}其中启动 Kubelet 核心的 RunKubelet 函数如下：
// RunKubelet 函数负责配置和运行 kubelet。它用于三种不同的应用：
// 1. 集成测试
// 2. Kubelet 二进制
// 3. 独立的 kubernetes 二进制
// 最终，#2将会被#3的实例所替换
func RunKubelet(kubeServer *options.KubeletServer, kubeDeps *kubelet.Dependencies, runOnce bool) error {
......
// 核心仍然是调用 createAndInitKubelet 函数将 Kubelet 配置、依赖等构造成 kubelet 的启动接口
// 然后调用 startKubelet 函数启动 Kubelet
startKubelet(k, podCfg, &kubeServer.KubeletConfiguration, kubeDeps, kubeServer.EnableCAdvisorJSONEndpoints, kubeServer.EnableServer)
return nil
}其中 createAndInitKubelet 函数调用了 kubelet.NewMainKubelet 函数来构造 kubelet 的启动接口，NewMainKubelet 函数位于pkg/kubelet/kubelet.go：
// NewMainKubelet 函数实例化一个新的 Kubelet 对象以及所有必需的内部模块。此处不会初始化 Kubelet 及其模块。
func NewMainKubelet(...) {
// 一些常规校验
// 指定容器垃圾回收策略
// 指定镜像垃圾回收策略
// 配置驱逐相关策略
// 配置kubelet客户端
// 配置集群DNS
// 组装Kubelet结构体
// 设置 secretManager
// 设置 configMapManager
// 设置 livenessManager
// 设置 podManager，podManager 还负责保持 secretManager 和 configMapManager 内容为最新。
// 设置 statusManager
// 启动 docker CRI shim GRPC 服务(包括RuntimeService和ImageService两个客户端):
// 默认 remoteRuntimeEndpoint=unix:///var/run/dockershim.sock，即默认使用本地的docker作为容器运行时；如果remoteImageEndpoint没有指定，则和remoteRuntimeEndpoint保持一致
// 然后创建一个新的 kubeGenericRuntimeManager
// 初始化一个新的 GenericPLEG 对象，用于配置pod生命周期事件生成器
// 然后配置容器回收策略
// 配置镜像回收策略
// 如果配置了 Kubelet 证书自动轮转(开启RotateKubeletServerCertificate特性)，则初始化证书管理器
// 然后配置probe管理器(探针管理)
// 生成一个对pod的service account token的管理器
// 调用如下方法配置volume插件管理器：
// NewInitializedVolumePluginMgr函数初始化Kubelet runtimeState上的一些storageErrors（在csi_plugin.go init中），这会影响节点d的就绪状态。必须在初始化Kubelet之前调用此函数，以便节点ReadyState在存储状态下准确无误。
klet.volumePluginMgr, err =
NewInitializedVolumePluginMgr(klet, secretManager, configMapManager, tokenManager, kubeDeps.VolumePlugins, kubeDeps.DynamicPluginProber)
// 然后配置volume管理器
// 配置eviction(驱逐)管理器
// 如果开启了NodeLease特性(节点生命周期控制器使用lease api报告的心跳来作为节点健康的信号))，则配置nodeLease控制器
// 最后，将最新版本的配置放在 Kubelet 上，以便可以看到它的配置方式:
klet.kubeletConfiguration = *kubeCfg
// 最后的最后，生成状态函数应该是我们做的最后一件事，因为这依赖于已经构造的Kubelet的其他部分。
// defaultNodeStatusFuncs 是一个工厂函数(位于pkg/kubelet/kubelet_node_status.go)，它生成一组默认的 setNodeStatus 函数
klet.setNodeStatusFuncs = klet.defaultNodeStatusFuncs()
return klet, nil
}其中 startKubelet 函数在调用上面的 createAndInitKubelet 函数获取到最新的 kubelet 启动对象以后，直接启动 kubelet 的一个server：
// startKubelet 函数很简单就是直接启动一个 kubelet 的server，
func startKubelet(k kubelet.Bootstrap, podCfg *config.PodConfig, kubeCfg *kubeletconfiginternal.KubeletConfiguration, kubeDeps *kubelet.Dependencies, enableCAdvisorJSONEndpoints, enableServer bool) {
// 开启一个goroutine，调用Run函数，不退出。
go wait.Until(func() {
k.Run(podCfg.Updates())
}, 0, wait.NeverStop)
// 开启 kubelet server
if enableServer {
go k.ListenAndServe(net.ParseIP(kubeCfg.Address), uint(kubeCfg.Port), kubeDeps.TLSOptions, kubeDeps.Auth, enableCAdvisorJSONEndpoints, kubeCfg.EnableDebuggingHandlers, kubeCfg.EnableContentionProfiling)
}
// 如果开启了只读端口，然后也会在一个gorouting 里面开启一个只读的 server
if kubeCfg.ReadOnlyPort > 0 {
go k.ListenAndServeReadOnly(net.ParseIP(kubeCfg.Address), uint(kubeCfg.ReadOnlyPort), enableCAdvisorJSONEndpoints)
}
// 如果开启了 kubelet 的 pod 资源 grpc 服务，则同样运行 一个kubelet 的 podresource grpc 服务
if utilfeature.DefaultFeatureGate.Enabled(features.KubeletPodResources) {
go k.ListenAndServePodResources()
}
}其中 Run 函数，位于pkg/kubelet/kubelet.go，运行启动 kubelet 响应配置更新：
func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) {
// 配置 logServer
// 检查 kubeClient
// 开启 cloud provider 同步管理器
// 调用 initializeModules 函数初始化内部模块（不需要启动容器运行时）
if err := kl.initializeModules(); err != nil {
kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.KubeletSetupFailed, err.Error())
klog.Fatal(err)
}
// 启动 volume 管理器
if kl.kubeClient != nil {
// 立即开始同步节点状态，这可能会设置运行时需要运行的内容
go wait.Until(kl.syncNodeStatus, kl.nodeStatusUpdateFrequency, wait.NeverStop)
// FastStatusUpdateOnce 函数启动一个循环，在应用CIDR时检查内部节点索引器缓存，并尝试立即更新POD CIDR。更新pod CIDR后，它会触发运行时更新和节点状态更新。函数在一次成功的节点状态更新后直接返回。该功能仅在 kubelet 启动期间执行，通过尽快更新 pod cidr、运行时状态和节点状态来提高准备就绪节点的延迟。
go kl.fastStatusUpdateOnce()
// 开始同步 lease
if utilfeature.DefaultFeatureGate.Enabled(features.NodeLease) {
go kl.nodeLeaseController.Run(wait.NeverStop)
}
}
// 每5s调用一次updateRuntimeUp函数检查容器运行时状态
go wait.Until(kl.updateRuntimeUp, 5*time.Second, wait.NeverStop)
// 启动循环同步 iptables 规则
if kl.makeIPTablesUtilChains {
// 不过当前syncNetworkUtil函数是一个空实现
go wait.Until(kl.syncNetworkUtil, 1*time.Minute, wait.NeverStop)
}
// 当 pod 没有被 podworker 正确处理的时候，启动一个 goroutine 杀死pod。
go wait.Until(kl.podKiller, 1*time.Second, wait.NeverStop)
// 启动组件同步循环
// 使用apiserver同步pods状态; 也用作状态缓存。
kl.statusManager.Start()
// 处理容器探针
kl.probeManager.Start()
// 启动 RuntimeClass 控制器
if kl.runtimeClassManager != nil {
kl.runtimeClassManager.Start(wait.NeverStop)
}
// 启动 Pod 生命周期事件生成器。
// PodLifecycleEventGenerator 是一个包含下面函数的pod生命周期事件生成器接口：
// type PodLifecycleEventGenerator interface {
// Start()
// Watch() chan *PodLifecycleEvent
// Healthy() (bool, error)
// }
kl.pleg.Start()
// 调用updates的同步循环
kl.syncLoop(updates, kl)
}其中 initializeModules 函数用来初始化内部得一些模块（不需要启动容器运行时）
func (kl *Kubelet) initializeModules() error {
// 注册 prometheus metrics 接口
// 创建文件系统目录:
// 1.根目录 2.pods目录 3.插件目录 4.pod-resources 目录
if err := kl.setupDataDirs(); err != nil {
return err
}
// 如果容器日志目录不存在，则创建该目录(/var/log/containers)
// 然后启动镜像管理器
kl.imageManager.Start()
// 开启 certificate 管理器
if kl.serverCertificateManager != nil {
kl.serverCertificateManager.Start()
}
// 开启 out of memory 监听器
if err := kl.oomWatcher.Start(kl.nodeRef); err != nil {
return fmt.Errorf("Failed to start OOM watcher %v", err)
}
// 开启资源分析器
kl.resourceAnalyzer.Start()
return nil
}然后就是调用 syncNodeStatus 函数失效节点状态同步，位于pkg/kubelet/kubelet_node_status.go:
// syncNodeStatus 函数定期被一个goroutine所调用。
// 如果和上次同步对比有任何变化或者上次同步经过的时间足够长，它会将节点状态同步到 master，如果需要请先注册 kubelet 。
func (kl *Kubelet) syncNodeStatus() {
kl.syncNodeStatusMux.Lock()
defer kl.syncNodeStatusMux.Unlock()
if kl.kubeClient == nil || kl.heartbeatClient == nil {
return
}
// 如果开启了registerNode
if kl.registerNode {
// 如果不需要执行任何操作，它将立即退出。registerWithApiServer 函数向群集 APIServer 注册节点。可以安全地多次调用，但不能同时调用（kl.registrationcompleted 未加锁）。
kl.registerWithAPIServer()
}
if err := kl.updateNodeStatus(); err != nil {
klog.Errorf("Unable to update node status: %v", err)
}
}调用 updateNodeStatus 函数更新节点状态:
// updateNodeStatus 函数更新节点状态到 master，如果上次同步有任何更改或经过足够的时间，则重试一次，最多重试5次
func (kl *Kubelet) updateNodeStatus() error {
klog.V(5).Infof("Updating node status")
for i := 0; i < nodeStatusUpdateRetry; i++ {
// 调用 tryUpdateNodeStatus 函数尝试进行状态更新
if err := kl.tryUpdateNodeStatus(i); err != nil {
// 当心跳操作失败多次时，调用 onRepeatedHeartBeatFailure 函数
if i > 0 && kl.onRepeatedHeartbeatFailure != nil {
kl.onRepeatedHeartbeatFailure()
}
klog.Errorf("Error updating node status, will retry: %v", err)
} else {
return nil
}
}
return fmt.Errorf("update node status exceeds retry count")
}调用 tryUpdateNodeStatus 函数尝试去进行真正的节点状态更新：
func (kl *Kubelet) tryUpdateNodeStatus(tryNumber int) error {
// 在大型集群中，这里的节点对象的 GET 和 PUT 操作是 apiserver 和 etcd 上的主要负载。
// 为了减少 etcd 上的负载，我们提供了从 apiserver 缓存获取操作（数据可能稍有延迟，但似乎不会造成更多冲突-延迟非常小）。
// 如果导致冲突，则所有重试都直接从 etcd 提供。
opts := metav1.GetOptions{}
if tryNumber == 0 {
util.FromApiserverCache(&opts)
}
node, err := kl.heartbeatClient.CoreV1().Nodes().Get(string(kl.nodeName), opts)
if err != nil {
return fmt.Errorf("error getting node %q: %v", kl.nodeName, err)
}
originalNode := node.DeepCopy()
if originalNode == nil {
return fmt.Errorf("nil %q node object", kl.nodeName)
}
podCIDRChanged := false
if node.Spec.PodCIDR != "" {
// Pod CIDR 之前就已经更新过了，因此我们需要判断node.Spec.PodCIDR是否为空，还需要知道 pod CIDR 是否真的已经改变了。
if podCIDRChanged, err = kl.updatePodCIDR(node.Spec.PodCIDR); err != nil {
klog.Errorf(err.Error())
}
}
// setNodeStatus 函数就是调用上面我们设置的 setNodeStatusFuncs 函数列表来进行状态更新
kl.setNodeStatus(node)
// 如果开启了 NodeLease 特性
now := kl.clock.Now()
if utilfeature.DefaultFeatureGate.Enabled(features.NodeLease) && now.Before(kl.lastStatusReportTime.Add(kl.nodeStatusReportFrequency)) {
if !podCIDRChanged && !nodeStatusHasChanged(&originalNode.Status, &node.Status) {
// 将指定的卷标记为已在节点的卷状态中成功报告为“正在使用”。
kl.volumeManager.MarkVolumesAsReportedInUse(node.Status.VolumesInUse)
return nil
}
}
// Patch 现在节点的状态到 APIServer
updatedNode, _, err := nodeutil.PatchNodeStatus(kl.heartbeatClient.CoreV1(), types.NodeName(kl.nodeName), originalNode, node)
if err != nil {
return err
}
// 设置当前时间为最近一次状态上报时间
kl.lastStatusReportTime = now
kl.setLastObservedNodeAddresses(updatedNode.Status.Addresses)
// 如果更新成功完成，请将卷使用标记为ReportedInUse以表明这些卷已在节点状态下更新
kl.volumeManager.MarkVolumesAsReportedInUse(updatedNode.Status.VolumesInUse)
return nil
}