kubelet 原理分析

kubelet 简介

kubernetes 分为控制面和数据面，kubelet 就是数据面最主要的组件，在每个节点上启动，主要负责容器的创建、启停、监控、日志收集等工作。它是一个在每个集群节点上运行的代理，负责确保节点上的容器根据PodSpec（Pod定义文件）正确运行。
Kubelet执行以下几项重要功能：

• Pod生命周期管理：Kubelet根据从API服务器接收到的PodSpecs创建、启动、终止容器。它负责启动Pod中的容器，并确保它们按预期运行。
  
• 节点状态监控：Kubelet定期监控节点和容器的状态，并将状态报告回集群的控制平面。这使得集群中的其他组件能够做出相应的调度决策。
  
• 资源管理：Kubelet负责管理分配给每个Pod的资源。这包括CPU、内存和磁盘存储资源。
  
• 健康检查：Kubelet可以执行容器健康检查，并根据检查结果决定是否需要重启容器。
  
• 与容器运行时的通信：Kubelet与容器运行时（如Docker、containerd等）通信，以管理容器的生命周期。
  
• 秘密和配置管理：Kubelet负责将秘密、配置映射等挂载到Pod的容器中，以便应用程序可以访问这些配置。
  
• 服务发现和负载均衡：尽管Kubelet本身不直接处理服务发现，但它通过设置网络规则和环境变量来支持容器内的服务发现机制。
  

kubelet 架构


kubelet 的架构由 N 多的组件组成，下面简单介绍下比较重要的几个：

• Sync Loop: 这是Kubelet活动的核心，负责同步Pod的状态。同步循环会定期从API服务器获取PodSpecs，并确保容器的当前状态与这些规格相匹配。
  
• PodConfig: 负责将各个配置源转换成 PodSpecs，可以选择的配置源包括：Kube-apiserver、本地文件、HTTP。
  
• PLEG(Pod Lifecycle Event Generator)： 负责监测和缓存Pod生命周期事件，如创建、启动或停止容器，然后将这些事件通知 Sync Loop。
  
• PodWorkers: 负责管理 Pod 的生命周期事件处理。当 Pod 生命周期事件 PLEG 检测到新的事件时，PodWorkers 会被调用来处理这些事件，包括启动新的 Pod、更新现有的 Pod、或者停止和清理不再需要的 Pod。
  
• PodManager: 存储 Pod 的期望状态，kubelet 服务的不同渠道的 Pod。
  
• ContainerRuntime: 顾名思义，容器运行时。与遵循 CRI 规范的高级容器运行时进行交互。
  
• StatsProvider: 提供节点和容器的统计信息，有 cAdvisor 和 CRI 两种实现。
  
• ProbeManager: 负责执行容器的健康检查，包括 Liveness，Startup 和 Readiness 检查。
  
• VolumeManager: 负责管理 Pod 的卷，包括挂载和卸载卷。
  
• ImageManager: 负责管理镜像，包括拉取、删除、镜像 GC 等。
  
• DeviceManager: 负责管理设备，包括 GPU、RDMA 等。
  
• PluginManager: PluginManager 运行一组异步循环，根据此节点确定哪些插件需要注册/取消注册并执行。如 CSI 驱动和设备管理器插件（Device Plugin）。
  
• CertificateManager: 处理证书轮换。
  
• OOMWatcher: 从系统日志中获取容器的 OOM 日志，将其封装成事件并记录。
  

流程

首先在 cmd/kubelet 中使用传入命令行参数的方式初始化配置，然后创建 pkg/kubelet 中的 Bootstrap inferface, kubelet struct 实现了这个接口， 然后调用 Run 方法启动 kubelet。

1. func startKubelet(k kubelet.Bootstrap, podCfg *config.PodConfig, kubeCfg *kubeletconfiginternal.KubeletConfiguration, kubeDeps *kubelet.Dependencies, enableServer bool) {
2.         // start the kubelet
3.         go k.Run(podCfg.Updates())
5.         // start the kubelet server
6.         if enableServer {
7.                 go k.ListenAndServe(kubeCfg, kubeDeps.TLSOptions, kubeDeps.Auth, kubeDeps.TracerProvider)
8.         }
9.         if kubeCfg.ReadOnlyPort > 0 {
10.                 go k.ListenAndServeReadOnly(netutils.ParseIPSloppy(kubeCfg.Address), uint(kubeCfg.ReadOnlyPort))
11.         }
12.         go k.ListenAndServePodResources()
13. }

复制代码

Bootstrap

1. // Bootstrap is a bootstrapping interface for kubelet, targets the initialization protocol
2. type Bootstrap interface {
3.         GetConfiguration() kubeletconfiginternal.KubeletConfiguration
4.         BirthCry()
5.         StartGarbageCollection()
6.         ListenAndServe(kubeCfg *kubeletconfiginternal.KubeletConfiguration, tlsOptions *server.TLSOptions, auth server.AuthInterface, tp trace.TracerProvider)
7.         ListenAndServeReadOnly(address net.IP, port uint)
8.         ListenAndServePodResources()
9.         Run(<-chan kubetypes.PodUpdate)
10.         RunOnce(<-chan kubetypes.PodUpdate) ([]RunPodResult, error)
11. }
13. type Kubelet struct {
14.     // ...
15. }

复制代码

首先解释一下这个函数的参数：

• configCh: 将配置更改的 Pod 分派给适当的处理程序回调函数
  
• plegCh: 更新运行时缓存；同步 Pod
  
• syncCh: 同步所有等待同步的 Pod
  
• housekeepingCh: 触发 Pod 的清理
  
• health manager: 同步失败的 Pod 或其中一个或多个容器的健康检查失败的 Pod
  

代码流程为：

• 如果 updates channel 有消息 则使用 handler 调用对应方法做处理
  
• 如果 plegCh 有消息 则使用 handler 的 HandlePodSyncs 做同步
  
• 如果 syncCh 有消息 代表到了同步时间 做同步操作
  
• 如果是 三种 probe 的更新 则使用 handleProbeSync 做同步
  
• 如果 housekeepingCh 有消息 则使用 handler 的 HandlePodCleanups 做清理
  

1. func (kl *Kubelet) StartGarbageCollection() {
2.         loggedContainerGCFailure := false
3.         go wait.Until(func() {
4.                 ctx := context.Background()
5.                 if err := kl.containerGC.GarbageCollect(ctx); err != nil {
6.                         klog.ErrorS(err, "Container garbage collection failed")
7.                         kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.ContainerGCFailed, err.Error())
8.                         loggedContainerGCFailure = true
9.                 } else {
10.                         var vLevel klog.Level = 4
11.                         if loggedContainerGCFailure {
12.                                 vLevel = 1
13.                                 loggedContainerGCFailure = false
14.                         }
16.                         klog.V(vLevel).InfoS("Container garbage collection succeeded")
17.                 }
18.         }, ContainerGCPeriod, wait.NeverStop)
20.         // when the high threshold is set to 100, stub the image GC manager
21.         if kl.kubeletConfiguration.ImageGCHighThresholdPercent == 100 {
22.                 klog.V(2).InfoS("ImageGCHighThresholdPercent is set 100, Disable image GC")
23.                 return
24.         }
26.         prevImageGCFailed := false
27.         go wait.Until(func() {
28.                 ctx := context.Background()
29.                 if err := kl.imageManager.GarbageCollect(ctx); err != nil {
30.                         if prevImageGCFailed {
31.                                 klog.ErrorS(err, "Image garbage collection failed multiple times in a row")
32.                                 // Only create an event for repeated failures
33.                                 kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.ImageGCFailed, err.Error())
34.                         } else {
35.                                 klog.ErrorS(err, "Image garbage collection failed once. Stats initialization may not have completed yet")
36.                         }
37.                         prevImageGCFailed = true
38.                 } else {
39.                         var vLevel klog.Level = 4
40.                         if prevImageGCFailed {
41.                                 vLevel = 1
42.                                 prevImageGCFailed = false
43.                         }
45.                         klog.V(vLevel).InfoS("Image garbage collection succeeded")
46.                 }
47.         }, ImageGCPeriod, wait.NeverStop)
48. }

复制代码

handleProbeSync也是使用 handler 的 HandlePodSyncs 做同步
handle (SyncHandler)

1. // RunOnce polls from one configuration update and run the associated pods.
2. func (kl *Kubelet) RunOnce(updates <-chan kubetypes.PodUpdate) ([]RunPodResult, error) {
3.         ctx := context.Background()
4.         // Setup filesystem directories.
5.         if err := kl.setupDataDirs(); err != nil {
6.                 return nil, err
7.         }
9.         // If the container logs directory does not exist, create it.
10.         if _, err := os.Stat(ContainerLogsDir); err != nil {
11.                 if err := kl.os.MkdirAll(ContainerLogsDir, 0755); err != nil {
12.                         klog.ErrorS(err, "Failed to create directory", "path", ContainerLogsDir)
13.                 }
14.         }
16.         select {
17.         case u := <-updates:
18.                 klog.InfoS("Processing manifest with pods", "numPods", len(u.Pods))
19.                 result, err := kl.runOnce(ctx, u.Pods, runOnceRetryDelay)
20.                 klog.InfoS("Finished processing pods", "numPods", len(u.Pods))
21.                 return result, err
22.         case <-time.After(runOnceManifestDelay):
23.                 return nil, fmt.Errorf("no pod manifest update after %v", runOnceManifestDelay)
24.         }
25. }
27. // runOnce runs a given set of pods and returns their status.
28. func (kl *Kubelet) runOnce(ctx context.Context, pods []*v1.Pod, retryDelay time.Duration) (results []RunPodResult, err error) {
29.         ch := make(chan RunPodResult)
30.         admitted := []*v1.Pod{}
31.         for _, pod := range pods {
32.                 // Check if we can admit the pod.
33.                 if ok, reason, message := kl.canAdmitPod(admitted, pod); !ok {
34.                         kl.rejectPod(pod, reason, message)
35.                         results = append(results, RunPodResult{pod, nil})
36.                         continue
37.                 }
39.                 admitted = append(admitted, pod)
40.                 go func(pod *v1.Pod) {
41.                         err := kl.runPod(ctx, pod, retryDelay)
42.                         ch <- RunPodResult{pod, err}
43.                 }(pod)
44.         }
46.         klog.InfoS("Waiting for pods", "numPods", len(admitted))
47.         failedPods := []string{}
48.         for i := 0; i < len(admitted); i++ {
49.                 res := <-ch
50.                 results = append(results, res)
51.                 if res.Err != nil {
52.                         failedContainerName, err := kl.getFailedContainers(ctx, res.Pod)
53.                         if err != nil {
54.                                 klog.InfoS("Unable to get failed containers' names for pod", "pod", klog.KObj(res.Pod), "err", err)
55.                         } else {
56.                                 klog.InfoS("Unable to start pod because container failed", "pod", klog.KObj(res.Pod), "containerName", failedContainerName)
57.                         }
58.                         failedPods = append(failedPods, format.Pod(res.Pod))
59.                 } else {
60.                         klog.InfoS("Started pod", "pod", klog.KObj(res.Pod))
61.                 }
62.         }
63.         if len(failedPods) > 0 {
64.                 return results, fmt.Errorf("error running pods: %v", failedPods)
65.         }
66.         klog.InfoS("Pods started", "numPods", len(pods))
67.         return results, err
68. }

复制代码

也是 kubelet struct 实现了这个接口

1. func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) {
2.         ctx := context.Background()
3.         if kl.logServer == nil {
4.                 file := http.FileServer(http.Dir(nodeLogDir))
5.                 if utilfeature.DefaultFeatureGate.Enabled(features.NodeLogQuery) && kl.kubeletConfiguration.EnableSystemLogQuery {
6.                         kl.logServer = http.StripPrefix("/logs/", http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
7.                                 if nlq, errs := newNodeLogQuery(req.URL.Query()); len(errs) > 0 {
8.                                         http.Error(w, errs.ToAggregate().Error(), http.StatusBadRequest)
9.                                         return
10.                                 } else if nlq != nil {
11.                                         if req.URL.Path != "/" && req.URL.Path != "" {
12.                                                 http.Error(w, "path not allowed in query mode", http.StatusNotAcceptable)
13.                                                 return
14.                                         }
15.                                         if errs := nlq.validate(); len(errs) > 0 {
16.                                                 http.Error(w, errs.ToAggregate().Error(), http.StatusNotAcceptable)
17.                                                 return
18.                                         }
19.                                         // Validation ensures that the request does not query services and files at the same time
20.                                         if len(nlq.Services) > 0 {
21.                                                 journal.ServeHTTP(w, req)
22.                                                 return
23.                                         }
24.                                         // Validation ensures that the request does not explicitly query multiple files at the same time
25.                                         if len(nlq.Files) == 1 {
26.                                                 // Account for the \ being used on Windows clients
27.                                                 req.URL.Path = filepath.ToSlash(nlq.Files[0])
28.                                         }
29.                                 }
30.                                 // Fall back in case the caller is directly trying to query a file
31.                                 // Example: kubectl get --raw /api/v1/nodes/$name/proxy/logs/foo.log
32.                                 file.ServeHTTP(w, req)
33.                         }))
34.                 } else {
35.                         kl.logServer = http.StripPrefix("/logs/", file)
36.                 }
37.         }
38.         if kl.kubeClient == nil {
39.                 klog.InfoS("No API server defined - no node status update will be sent")
40.         }
42.         // Start the cloud provider sync manager
43.         if kl.cloudResourceSyncManager != nil {
44.                 go kl.cloudResourceSyncManager.Run(wait.NeverStop)
45.         }
47.         if err := kl.initializeModules(); err != nil {
48.                 kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.KubeletSetupFailed, err.Error())
49.                 klog.ErrorS(err, "Failed to initialize internal modules")
50.                 os.Exit(1)
51.         }
53.         // Start volume manager
54.         go kl.volumeManager.Run(kl.sourcesReady, wait.NeverStop)
56.         if kl.kubeClient != nil {
57.                 // Start two go-routines to update the status.
58.                 //
59.                 // The first will report to the apiserver every nodeStatusUpdateFrequency and is aimed to provide regular status intervals,
60.                 // while the second is used to provide a more timely status update during initialization and runs an one-shot update to the apiserver
61.                 // once the node becomes ready, then exits afterwards.
62.                 //
63.                 // Introduce some small jittering to ensure that over time the requests won't start
64.                 // accumulating at approximately the same time from the set of nodes due to priority and
65.                 // fairness effect.
66.                 go wait.JitterUntil(kl.syncNodeStatus, kl.nodeStatusUpdateFrequency, 0.04, true, wait.NeverStop)
67.                 go kl.fastStatusUpdateOnce()
69.                 // start syncing lease
70.                 go kl.nodeLeaseController.Run(context.Background())
71.         }
72.         go wait.Until(kl.updateRuntimeUp, 5*time.Second, wait.NeverStop)
74.         // Set up iptables util rules
75.         if kl.makeIPTablesUtilChains {
76.                 kl.initNetworkUtil()
77.         }
79.         // Start component sync loops.
80.         kl.statusManager.Start()
82.         // Start syncing RuntimeClasses if enabled.
83.         if kl.runtimeClassManager != nil {
84.                 kl.runtimeClassManager.Start(wait.NeverStop)
85.         }
87.         // Start the pod lifecycle event generator.
88.         kl.pleg.Start()
90.         // Start eventedPLEG only if EventedPLEG feature gate is enabled.
91.         if utilfeature.DefaultFeatureGate.Enabled(features.EventedPLEG) {
92.                 kl.eventedPleg.Start()
93.         }
95.         kl.syncLoop(ctx, updates, kl)
96. }

复制代码

可以看到这些函数基本都是把pod交给 podWorkers 去处理
podconfig

上文中的 updates channel 是从 podconfig 中获取的 那就来看看 podconfig 是怎么工作的

1. // syncLoop is the main loop for processing changes. It watches for changes from
2. // three channels (file, apiserver, and http) and creates a union of them. For
3. // any new change seen, will run a sync against desired state and running state. If
4. // no changes are seen to the configuration, will synchronize the last known desired
5. // state every sync-frequency seconds. Never returns.
6. func (kl *Kubelet) syncLoop(ctx context.Context, updates <-chan kubetypes.PodUpdate, handler SyncHandler) {
7.         klog.InfoS("Starting kubelet main sync loop")
8.         // The syncTicker wakes up kubelet to checks if there are any pod workers
9.         // that need to be sync'd. A one-second period is sufficient because the
10.         // sync interval is defaulted to 10s.
11.         syncTicker := time.NewTicker(time.Second)
12.         defer syncTicker.Stop()
13.         housekeepingTicker := time.NewTicker(housekeepingPeriod)
14.         defer housekeepingTicker.Stop()
15.         plegCh := kl.pleg.Watch()
16.         const (
17.                 base   = 100 * time.Millisecond
18.                 max    = 5 * time.Second
19.                 factor = 2
20.         )
21.         duration := base
22.         // Responsible for checking limits in resolv.conf
23.         // The limits do not have anything to do with individual pods
24.         // Since this is called in syncLoop, we don't need to call it anywhere else
25.         if kl.dnsConfigurer != nil && kl.dnsConfigurer.ResolverConfig != "" {
26.                 kl.dnsConfigurer.CheckLimitsForResolvConf()
27.         }
29.         for {
30.                 if err := kl.runtimeState.runtimeErrors(); err != nil {
31.                         klog.ErrorS(err, "Skipping pod synchronization")
32.                         // exponential backoff
33.                         time.Sleep(duration)
34.                         duration = time.Duration(math.Min(float64(max), factor*float64(duration)))
35.                         continue
36.                 }
37.                 // reset backoff if we have a success
38.                 duration = base
40.                 kl.syncLoopMonitor.Store(kl.clock.Now())
41.                 if !kl.syncLoopIteration(ctx, updates, handler, syncTicker.C, housekeepingTicker.C, plegCh) {
42.                         break
43.                 }
44.                 kl.syncLoopMonitor.Store(kl.clock.Now())
45.         }
46. }

复制代码

这里定义了接口 接口提中可以存储多个 source 那么有什么 source 呢
kube-apiserver

[code]func NewSourceApiserver(c clientset.Interface, nodeName types.NodeName, nodeHasSynced func() bool, updates chan