Kubernetes 源码剖析 — kubelet 启动流程

func Run(
  ctx context.Context,
  s *options.KubeletServer,
  kubeDeps *kubelet.Dependencies,
  featureGate featuregate.FeatureGate
) error {
    // 1. .. 省略: 根据 options.KubeletServer 中的参数配置 klog
    // 2. .. 省略: 执行不同 OS 的初始化操作，目前仅有 Window 下有初始化操作，其他系统直接空函数
    // 3. 调用私有方法 `run`
    if err := run(ctx, s, kubeDeps, featureGate); err != nil {
        return fmt.Errorf("failed to run Kubelet: %v", err)
    }
    return nil
}

func run(
  ctx context.Context, 
  s *options.KubeletServer,
  kubeDeps *kubelet.Dependencies,
  featureGate featuregate.FeatureGate
) (err error) {
    // 1. .. 省略: 设置 FeatureGates, 然后检查 options 合法性
    // 2. .. 省略:  如果启动参数中包含 --lock-file 则会尝试获取文件锁，失败则退出
    // 如果文件锁获取成功，且参数中指定了 --exit-on-lock-contention
    // 则会创建一个 goroutine 借助 inotify 去 watch 文件，若产生 `inotify.InOpen|inotify.InDeleteSelf` 事件
    // 则 kubelet 退出
    // 3. .. 省略: 将 s.KubeletConfiguration 注册到 configz
    // 4. .. 省略: 根据 --kubeconfig 判断是按照 API server mode 还是 standalone mode 启动
    // 5. .. 省略: 初始化 CloudProvider, 1.23 之后就没了，可以不管
    // 6. .. 省略: 获取 hostName 和 nodeName
    // hostName 如若不指定，则获取的是 trim 后转小写的 `kern.hostname` 
    // nodeName 由 cloud-provider 提供，没有则使用 hostName
    // 7. .. 省略: 创建 KubeClient / EventClient / HeartbeatClient，
    // 如果是 standalone mode 则不需要。KubeClient 会向 API Server 发送节点的状态信息
    // 8. .. 省略: 将以下的 cgroups 加入 cadvisor 的 metrics 中
    // - root cgroup 可选，由 --cgroup-root 指定，比如 Systemd 和 cgroupfs
    // - kubelet cgroup 可选，由 --kubelet-cgroups 指定，默认是 kubelet 进程所在的，参考 GetKubeletContainer 函数
    // - runtime cgroup 可选，由 --runtime-cgroups 指定，一般是使用 docker 进程所在的
    // - system cgroup 可选，由 --system-cgroups 指定 
    // 9. 创建 ContainerManager
	if kubeDeps.ContainerManager == nil {
        // .. 省略计算 reserved-cpus 的部分
		kubeDeps.ContainerManager, err = cm.NewContainerManager(
			kubeDeps.Mounter,
			kubeDeps.CAdvisorInterface,
			cm.NodeConfig{
				RuntimeCgroupsName:    s.RuntimeCgroups,
				SystemCgroupsName:     s.SystemCgroups,
				KubeletCgroupsName:    s.KubeletCgroups,
				ContainerRuntime:      s.ContainerRuntime,
				CgroupsPerQOS:         s.CgroupsPerQOS,
				CgroupRoot:            s.CgroupRoot,
				CgroupDriver:          s.CgroupDriver,
				KubeletRootDir:        s.RootDirectory,
				ProtectKernelDefaults: s.ProtectKernelDefaults,
				NodeAllocatableConfig: cm.NodeAllocatableConfig{
					KubeReservedCgroupName:   s.KubeReservedCgroup,
					SystemReservedCgroupName: s.SystemReservedCgroup,
					EnforceNodeAllocatable:   sets.NewString(s.EnforceNodeAllocatable...),
					KubeReserved:             kubeReserved,
					SystemReserved:           systemReserved,
					ReservedSystemCPUs:       reservedSystemCPUs,
					HardEvictionThresholds:   hardEvictionThresholds,
				},
				QOSReserved:                             *experimentalQOSReserved,
				ExperimentalCPUManagerPolicy:            s.CPUManagerPolicy,
				ExperimentalCPUManagerReconcilePeriod:   s.CPUManagerReconcilePeriod.Duration,
				ExperimentalMemoryManagerPolicy:         s.MemoryManagerPolicy,
				ExperimentalMemoryManagerReservedMemory: s.ReservedMemory,
				ExperimentalPodPidsLimit:                s.PodPidsLimit,
				EnforceCPULimits:                        s.CPUCFSQuota,
				CPUCFSQuotaPeriod:                       s.CPUCFSQuotaPeriod.Duration,
				ExperimentalTopologyManagerPolicy:       s.TopologyManagerPolicy,
				ExperimentalTopologyManagerScope:        s.TopologyManagerScope,
			},
			s.FailSwapOn,
			devicePluginEnabled,
			kubeDeps.Recorder)
        )
    }
    // 10. 调用 RunKubelet，内部会新建 goroutine
	err = kubelet.PreInitRuntimeService(&s.KubeletConfiguration,
		kubeDeps, &s.ContainerRuntimeOptions,
		s.ContainerRuntime,
		s.RuntimeCgroups,
		s.RemoteRuntimeEndpoint,
		s.RemoteImageEndpoint,
		s.NonMasqueradeCIDR)
	if err != nil {
		return err
	}

	if err := RunKubelet(s, kubeDeps, s.RunOnce); err != nil {
		return err
	}
    // 11. .. 省略: 如果开启健康检查，那么会临时启一个暴露 /healthz 的 HTTPServer 的 goroutine
    // 12. .. 省略: 此 goroutine 等待 channel 中的 message 来结束
}

type containerManagerImpl struct {
	sync.RWMutex
	cadvisorInterface cadvisor.Interface
	mountUtil         mount.Interface
	NodeConfig
	status Status
	// External containers being managed.
	systemContainers []*systemContainer
	// Tasks that are run periodically
	periodicTasks []func()
	// Holds all the mounted cgroup subsystems
	subsystems *CgroupSubsystems
	nodeInfo   *v1.Node
	// Interface for cgroup management
	cgroupManager CgroupManager
	// Capacity of this node.
	capacity v1.ResourceList
	// Capacity of this node, including internal resources.
	internalCapacity v1.ResourceList
	// Absolute cgroupfs path to a cgroup that Kubelet needs to place all pods under.
	// This path include a top level container for enforcing Node Allocatable.
	cgroupRoot CgroupName
	// Event recorder interface.
	recorder record.EventRecorder
	// Interface for QoS cgroup management
	qosContainerManager QOSContainerManager
	// Interface for exporting and allocating devices reported by device plugins.
	deviceManager devicemanager.Manager
	// Interface for CPU affinity management.
	cpuManager cpumanager.Manager
	// Interface for memory affinity management.
	memoryManager memorymanager.Manager
	// Interface for Topology resource co-ordination
	topologyManager topologymanager.Manager
}

func RunKubelet(
    kubeServer *options.KubeletServer,
    kubeDeps *kubelet.Dependencies,
    runOnce bool
) error {
    // 1. .. 省略: 获取当前 node 的 ip，可以由参数 --node-ip 指定，默认值是本机 IPv4 地址。并进行 dual-stack 的一些校验
    // 2. 创建并初始化 kubelet 结构体
	k, err := createAndInitKubelet(
    	// ... 省略参数
    )
	if err != nil {
		return fmt.Errorf("failed to create kubelet: %v", err)
	}

	// NewMainKubelet should have set up a pod source config if one didn't exist
	// when the builder was run. This is just a precaution.
	if kubeDeps.PodConfig == nil {
		return fmt.Errorf("failed to create kubelet, pod source config was nil")
	}
	podCfg := kubeDeps.PodConfig

	if err := rlimit.SetNumFiles(uint64(kubeServer.MaxOpenFiles)); err != nil {
		klog.ErrorS(err, "Failed to set rlimit on max file handles")
	}

    // 3. 启动 startKubelet
	if runOnce {
		if _, err := k.RunOnce(podCfg.Updates()); err != nil {
			return fmt.Errorf("runonce failed: %v", err)
		}
		klog.InfoS("Started kubelet as runonce")
	} else {
		startKubelet(k, podCfg, &kubeServer.KubeletConfiguration, kubeDeps, kubeServer.EnableServer)
		klog.InfoS("Started kubelet")
	}
	return nil
}

func startKubelet(k kubelet.Bootstrap, podCfg *config.PodConfig, kubeCfg *kubeletconfiginternal.KubeletConfiguration, kubeDeps *kubelet.Dependencies, enableServer bool) {
	// start the kubelet
	go k.Run(podCfg.Updates())

	// start the kubelet server
	if enableServer {
		go k.ListenAndServe(kubeCfg, kubeDeps.TLSOptions, kubeDeps.Auth)
	}
	if kubeCfg.ReadOnlyPort > 0 {
		go k.ListenAndServeReadOnly(net.ParseIP(kubeCfg.Address), uint(kubeCfg.ReadOnlyPort))
	}
	if utilfeature.DefaultFeatureGate.Enabled(features.KubeletPodResources) {
		go k.ListenAndServePodResources()
	}
}

// Run starts the kubelet reacting to config updates
func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) {
    // 1. 初始化 logServer，这个是一个 net.http.FileServer，将 /var/log 挂在到 HTTP的 /logs/ 路径下面
	if kl.logServer == nil {
		kl.logServer = http.StripPrefix("/logs/", http.FileServer(http.Dir("/var/log/")))
	}
    // 2. Start the cloud provider sync manager
	if kl.cloudResourceSyncManager != nil {
		go kl.cloudResourceSyncManager.Run(wait.NeverStop)
	}

    // 3. 内部初始化，具体是以下逻辑:
    //    - 注册 Prometheus metrics
    //    - 创建 kubelet 所使用的目录，比如 /var/lib/kubelet，权限均是 0750
    //    - 确保 /var/log/containers 存在，否则创建，权限是 0755
    //    - 启动 imageManager，这个会负责镜像的 GC 逻辑
    //    - Server 的权限认证管理
    //    - 启动 OOMWatcher 检测 System 的事件 实现参考 pkg/kubelet/oom/oom_watcher_linux.go
    //    - 启动 ResourceAnalyzer，
	if err := kl.initializeModules(); err != nil {
		kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.KubeletSetupFailed, err.Error())
		klog.ErrorS(err, "failed to intialize internal modules")
		os.Exit(1)
	}

	// 4. Start volume manager, 循环判断哪些 POD 需要 mounted/unmount volume
	go kl.volumeManager.Run(kl.sourcesReady, wait.NeverStop)

    // 5. 上报节点信息
	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
		go kl.nodeLeaseController.Run(wait.NeverStop)
	}
    // 6. 检查 container runtime status，会产生 grpc 调用到 docker
	go wait.Until(kl.updateRuntimeUp, 5*time.Second, wait.NeverStop)

	// 7. Set up iptables util rules
	if kl.makeIPTablesUtilChains {
		kl.initNetworkUtil()
	}

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

	// 9. StatusManager is the Source of truth for kubelet pod status, and should be kept up-to-date with
    // the latest v1.PodStatus. It also syncs updates back to the API server.
	kl.statusManager.Start()

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

	// 11. Start the pod lifecycle event generator.
	kl.pleg.Start()
    // 12. 主循环
	kl.syncLoop(updates, kl)
}