Kubernetes源码学习-Kubelet-P1-启动流程

矫情吗；* 2023-01-18 09:29 121阅读 0赞

前言

在大致分析过k8s的Scheduler、Controller、APIServer三个控制平面组件后，本篇开始进入数据交互平面的daemon组件kubelet部分，看看kubelet是如何在控制平面和数据平面中以承上启下的模式工作的。

启动流程

启动入口照旧，位于项目的cmd路径下，使用cobra做cmd封装：

cmd/kubelet/kubelet.go:39

  1. func main() { 
  2.     rand.Seed(time.Now().UnixNano())
  4.     command := app.NewKubeletCommand(server.SetupSignalHandler())
  5.     logs.InitLogs()
  6.     defer logs.FlushLogs()
  8.     if err := command.Execute(); err != nil { 
  9.         fmt.Fprintf(os.Stderr, "%v\n", err)
  10.         os.Exit(1)
  11.     }
  12. }

cmd/kubelet/app/server.go:112

NewKubeletFlags和NewKubeletConfiguration方法会初始化kubelet的很多默认flag和参数，来分别看下：

cmd/kubelet/app/options/options.go:214

  1. func NewKubeletFlags() *KubeletFlags { 
  2.     remoteRuntimeEndpoint := ""
  3.     if runtime.GOOS == "linux" { 
  4.         remoteRuntimeEndpoint = "unix:///var/run/dockershim.sock"
  5.     } else if runtime.GOOS == "windows" { 
  6.         remoteRuntimeEndpoint = "npipe:./pipe/dockershim"
  7.     }
  9.     return &KubeletFlags{ 
  10.         EnableServer:                        true,
  11.     // 容器运行时这个参数需要留意下
  12.         ContainerRuntimeOptions:             *NewContainerRuntimeOptions(),
  13.         CertDirectory:                       "/var/lib/kubelet/pki",
  14.         RootDirectory:                       defaultRootDir,
  15.         MasterServiceNamespace:              metav1.NamespaceDefault,
  16.         MaxContainerCount:                   -1,
  17.         MaxPerPodContainerCount:             1,
  18.         MinimumGCAge:                        metav1.Duration{ Duration: 0},
  19.         NonMasqueradeCIDR:                   "10.0.0.0/8",
  20.         RegisterSchedulable:                 true,
  21.         ExperimentalKernelMemcgNotification: false,
  22.         RemoteRuntimeEndpoint:               remoteRuntimeEndpoint,
  23.         NodeLabels:                          make(map[string]string),
  24.         VolumePluginDir:                     "/usr/libexec/kubernetes/kubelet-plugins/volume/exec/",
  25.         RegisterNode:                        true,
  26.         SeccompProfileRoot:                  filepath.Join(defaultRootDir, "seccomp"),
  27.         HostNetworkSources:                  []string{ kubetypes.AllSource},
  28.         HostPIDSources:                      []string{ kubetypes.AllSource},
  29.         HostIPCSources:                      []string{ kubetypes.AllSource},
  30.         // TODO(#58010:v1.13.0): Remove --allow-privileged, it is deprecated
  31.         AllowPrivileged: true,
  32.         // prior to the introduction of this flag, there was a hardcoded cap of 50 images
  33.         NodeStatusMaxImages: 50,
  34.     }
  35. }

ContainerRuntimeOptions有必要看看：

cmd/kubelet/app/options/container_runtime.go:41

  1. func NewContainerRuntimeOptions() *config.ContainerRuntimeOptions { 
  2.     dockerEndpoint := ""
  3.     if runtime.GOOS != "windows" { 
  4.     // 默认的容器驱动是docker
  5.         dockerEndpoint = "unix:///var/run/docker.sock"
  6.     }
  8.     return &config.ContainerRuntimeOptions{ 
  9.         ContainerRuntime:           kubetypes.DockerContainerRuntime,
  10.         RedirectContainerStreaming: false,
  11.         DockerEndpoint:             dockerEndpoint,
  12.     // dockershim路径，dockershim是容器运行中的实际载体，每个docker容器都会产生一个shim进程
  13.         DockershimRootDirectory:    "/var/lib/dockershim",
  14.     // pause容器
  15.         PodSandboxImage:            defaultPodSandboxImage,
  16.         ImagePullProgressDeadline:  metav1.Duration{ Duration: 1 * time.Minute},
  17.         ExperimentalDockershim:     false,
  19.         // 这个目录下都是网络相关功能工具的执行文件
  20.         CNIBinDir:  "/opt/cni/bin",
  21.     // 这里是cni的配置文件，如pod网段、网关、bridge等，一般由cni动态生成
  22.         CNIConfDir: "/etc/cni/net.d",
  23.     }
  24. }

cmd/kubelet/app/options/options.go:293

–> cmd/kubelet/app/options/options.go:311

  1. // `NewKubeletConfiguration`方法则会默认设置一些参数
  2. func applyLegacyDefaults(kc *kubeletconfig.KubeletConfiguration) { 
  3.     // --anonymous-auth
  4.     kc.Authentication.Anonymous.Enabled = true
  5.     // --authentication-token-webhook
  6.     kc.Authentication.Webhook.Enabled = false
  7.     // --authorization-mode
  8.   // apiserver认证篇提到的针对node设计的AlwaysAllow认证模式
  9.     kc.Authorization.Mode = kubeletconfig.KubeletAuthorizationModeAlwaysAllow
  10.     // 10255采集信息的接口，如prometheus采集cadvisor的metrics
  11.     kc.ReadOnlyPort = ports.KubeletReadOnlyPort
  12. }

再来看看Run方法里面做了哪些操作:

–> cmd/kubelet/app/server.go:148

  1. Run: func(cmd *cobra.Command, args []string) { 
  2.       ...
  3.             // 上面百来行代码都是默认的init flag和config相关处理，例如featureGates等，略过
  5.       // 加载kubelet配置文件，展开进去看可以看到即是--config参数对应指定的文件，一般kubeadm部署时使用的是/var/lib/kubelet/config.yaml
  6.             if configFile := kubeletFlags.KubeletConfigFile; len(configFile) > 0 { 
  7.                 kubeletConfig, err = loadConfigFile(configFile)
  8.                 if err != nil { 
  9.                     klog.Fatal(err)
  10.                 }
  11.       ... 
  12.             }
  14.             // 实例化KubeletServer
  15.             kubeletServer := &options.KubeletServer{ 
  16.                 KubeletFlags:         *kubeletFlags,
  17.                 KubeletConfiguration: *kubeletConfig,
  18.             }
  20.             // 构建一些kubelet的依赖插件，例如nsenter，连接dockershim的client端
  21.             kubeletDeps, err := UnsecuredDependencies(kubeletServer)
  22.             if err != nil { 
  23.                 klog.Fatal(err)
  24.             }
  26.             // add the kubelet config controller to kubeletDeps
  27.             kubeletDeps.KubeletConfigController = kubeletConfigController
  29.             // start the experimental docker shim, if enabled
  30.             if kubeletServer.KubeletFlags.ExperimentalDockershim { 
  31.                 if err := RunDockershim(&kubeletServer.KubeletFlags, kubeletConfig, stopCh); err != nil { 
  32.                     klog.Fatal(err)
  33.                 }
  34.                 return
  35.             }
  37.             // 启动kubelet
  38.             klog.V(5).Infof("KubeletConfiguration: %#v", kubeletServer.KubeletConfiguration)
  39.             if err := Run(kubeletServer, kubeletDeps, stopCh); err != nil { 
  40.                 klog.Fatal(err)
  41.             }
  42.         },
  43.     }
  45.     // 下面是一些cmd help信息，省略
  46.   ...
  48.     return cmd

–> cmd/kubelet/app/server.go:416

–> cmd/kubelet/app/server.go:479 这个函数代码段很长，两百多行，挑主要片段

  1. func run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chan struct{ }) (err error) { 
  2.  ...
  4.   // 独立模式指的是不与外部(如apiserver)交互的模式，一般在调试中使用，所以独立模式不需要起client
  5.     standaloneMode := true
  6.     if len(s.KubeConfig) > 0 { 
  7.         standaloneMode = false
  8.     }
  10.     if kubeDeps == nil { 
  11.         kubeDeps, err = UnsecuredDependencies(s)
  12.         if err != nil { 
  13.             return err
  14.         }
  15.     }
  17.   // 取得注册node名
  18.     hostName, err := nodeutil.GetHostname(s.HostnameOverride)
  19.     if err != nil { 
  20.         return err
  21.     }
  22.     nodeName, err := getNodeName(kubeDeps.Cloud, hostName)
  23.     if err != nil { 
  24.         return err
  25.     }
  28.     switch { 
  29.     // 独立模式，则所有client设为nil
  30.     case standaloneMode:
  31.         kubeDeps.KubeClient = nil
  32.         kubeDeps.EventClient = nil
  33.         kubeDeps.HeartbeatClient = nil
  34.         klog.Warningf("standalone mode, no API client")
  36.    // 正常模式，则初始化client，包括kubeClient/eventClient/heartBeatClient
  37.     case kubeDeps.KubeClient == nil, kubeDeps.EventClient == nil, kubeDeps.HeartbeatClient == nil:
  38.     // client的配置，主要是连接apiserver的cert相关的配置，cert文件默认放在/var/lib/kubelet/pki下，如果开启了循环续期证书，则相应的异步进程会从cert manager循环检测和更新证书。其他的配置诸如超时时间，长连接时间等。closeAllConns接收的是一个方法，用来断开连接。
  39.         clientConfig, closeAllConns, err := buildKubeletClientConfig(s, nodeName)
  40.         if err != nil { 
  41.             return err
  42.         }
  43.         if closeAllConns == nil { 
  44.             return errors.New("closeAllConns must be a valid function other than nil")
  45.         }
  46.         kubeDeps.OnHeartbeatFailure = closeAllConns
  47.         // 构建一个client-go里的clientset实例，访问各个GV和GVR对象使用
  48.         kubeDeps.KubeClient, err = clientset.NewForConfig(clientConfig)
  49.         if err != nil { 
  50.             return fmt.Errorf("failed to initialize kubelet client: %v", err)
  51.         }
  53.         // event事件使用独立的client，与上面的访问GVR使用的client区分开
  54.         eventClientConfig := *clientConfig
  55.         eventClientConfig.QPS = float32(s.EventRecordQPS)
  56.         eventClientConfig.Burst = int(s.EventBurst)
  57.         kubeDeps.EventClient, err = v1core.NewForConfig(&eventClientConfig)
  58.         if err != nil { 
  59.             return fmt.Errorf("failed to initialize kubelet event client: %v", err)
  60.         }
  62.         // 再开启一个心跳检测的client
  63.         heartbeatClientConfig := *clientConfig
  64.         heartbeatClientConfig.Timeout = s.KubeletConfiguration.NodeStatusUpdateFrequency.Duration
  65.     // 如果开启了NodeLease(node定期向apiserver汇报运行状态)，那么心跳间隔最大不超过NodeLease duration
  66.         if utilfeature.DefaultFeatureGate.Enabled(features.NodeLease) { 
  67.             leaseTimeout := time.Duration(s.KubeletConfiguration.NodeLeaseDurationSeconds) * time.Second
  68.             if heartbeatClientConfig.Timeout > leaseTimeout { 
  69.                 heartbeatClientConfig.Timeout = leaseTimeout
  70.             }
  71.         }
  72.     // 心跳1次/s
  73.         heartbeatClientConfig.QPS = float32(-1)
  74.         kubeDeps.HeartbeatClient, err = clientset.NewForConfig(&heartbeatClientConfig)
  75.         if err != nil { 
  76.             return fmt.Errorf("failed to initialize kubelet heartbeat client: %v", err)
  77.         }
  78.     }
  79.   // 向apiserver发起认证建立会话
  80.     if kubeDeps.Auth == nil { 
  81.         auth, err := BuildAuth(nodeName, kubeDeps.KubeClient, s.KubeletConfiguration)
  82.         if err != nil { 
  83.             return err
  84.         }
  85.         kubeDeps.Auth = auth
  86.     }
  87.   // 填充cadvisor接口
  88.     if kubeDeps.CAdvisorInterface == nil { 
  89.         imageFsInfoProvider := cadvisor.NewImageFsInfoProvider(s.ContainerRuntime, s.RemoteRuntimeEndpoint)
  90.         kubeDeps.CAdvisorInterface, err = cadvisor.New(imageFsInfoProvider, s.RootDirectory, cadvisor.UsingLegacyCadvisorStats(s.ContainerRuntime, s.RemoteRuntimeEndpoint))
  91.         if err != nil { 
  92.             return err
  93.         }
  94.     }
  96.     // Setup event recorder if required.
  97.     makeEventRecorder(kubeDeps, nodeName)
  99.     if kubeDeps.ContainerManager == nil { 
  100.         if s.CgroupsPerQOS && s.CgroupRoot == "" { 
  101.             klog.Info("--cgroups-per-qos enabled, but --cgroup-root was not specified. defaulting to /")
  102.             s.CgroupRoot = "/"
  103.         }
  104.     // /var/lib/kubelt/config.yaml里可以指定，为系统和kube组件指定不同的cgroup，为它们预留资源
  105.     // kubeReserved即为kube组件指定cgroup预留的资源
  106.         kubeReserved, err := parseResourceList(s.KubeReserved)
  107.         if err != nil { 
  108.             return err
  109.         }
  110.     // kubeReserved即为宿主机系统进程指定cgroup预留的资源
  111.         systemReserved, err := parseResourceList(s.SystemReserved)
  112.         if err != nil { 
  113.             return err
  114.         }
  115.     // 硬驱逐容器的资源阈值
  116.         var hardEvictionThresholds []evictionapi.Threshold
  117.         // If the user requested to ignore eviction thresholds, then do not set valid values for hardEvictionThresholds here.
  118.         if !s.ExperimentalNodeAllocatableIgnoreEvictionThreshold { 
  119.             hardEvictionThresholds, err = eviction.ParseThresholdConfig([]string{ }, s.EvictionHard, nil, nil, nil)
  120.             if err != nil { 
  121.                 return err
  122.             }
  123.         }
  124.         experimentalQOSReserved, err := cm.ParseQOSReserved(s.QOSReserved)
  125.         if err != nil { 
  126.             return err
  127.         }
  129.         devicePluginEnabled := utilfeature.DefaultFeatureGate.Enabled(features.DevicePlugins)
  131.     // 上面的参数汇集起来，初始化容器管理器
  132.         kubeDeps.ContainerManager, err = cm.NewContainerManager(
  133.             kubeDeps.Mounter,
  134.             kubeDeps.CAdvisorInterface,
  135.             cm.NodeConfig{ 
  136.                 RuntimeCgroupsName:    s.RuntimeCgroups,
  137.                 SystemCgroupsName:     s.SystemCgroups,
  138.                 KubeletCgroupsName:    s.KubeletCgroups,
  139.                 ContainerRuntime:      s.ContainerRuntime,
  140.                 CgroupsPerQOS:         s.CgroupsPerQOS,
  141.                 CgroupRoot:            s.CgroupRoot,
  142.                 CgroupDriver:          s.CgroupDriver,
  143.                 KubeletRootDir:        s.RootDirectory,
  144.                 ProtectKernelDefaults: s.ProtectKernelDefaults,
  145.                 NodeAllocatableConfig: cm.NodeAllocatableConfig{ 
  146.                     KubeReservedCgroupName:   s.KubeReservedCgroup,
  147.                     SystemReservedCgroupName: s.SystemReservedCgroup,
  148.                     EnforceNodeAllocatable:   sets.NewString(s.EnforceNodeAllocatable...),
  149.                     KubeReserved:             kubeReserved,
  150.                     SystemReserved:           systemReserved,
  151.                     HardEvictionThresholds:   hardEvictionThresholds,
  152.                 },
  153.                 QOSReserved:                           *experimentalQOSReserved,
  154.                 ExperimentalCPUManagerPolicy:          s.CPUManagerPolicy,
  155.                 ExperimentalCPUManagerReconcilePeriod: s.CPUManagerReconcilePeriod.Duration,
  156.                 ExperimentalPodPidsLimit:              s.PodPidsLimit,
  157.                 EnforceCPULimits:                      s.CPUCFSQuota,
  158.                 CPUCFSQuotaPeriod:                     s.CPUCFSQuotaPeriod.Duration,
  159.             },
  160.             s.FailSwapOn,
  161.             devicePluginEnabled,
  162.             kubeDeps.Recorder)
  164.         if err != nil { 
  165.             return err
  166.         }
  167.     }
  169.     if err := checkPermissions(); err != nil { 
  170.         klog.Error(err)
  171.     }
  173.     utilruntime.ReallyCrash = s.ReallyCrashForTesting
  175.     rand.Seed(time.Now().UnixNano())
  177.     // oom判定器给当前进程设置oom分数，容器内存资源管控的手段就是使用的oom，这里待会儿拎出来单独分析
  178.     oomAdjuster := kubeDeps.OOMAdjuster
  179.     if err := oomAdjuster.ApplyOOMScoreAdj(0, int(s.OOMScoreAdj)); err != nil { 
  180.         klog.Warning(err)
  181.     }
  182.   // RunKubelet接往下文
  183.     if err := RunKubelet(s, kubeDeps, s.RunOnce); err != nil { 
  184.         return err
  185.     }
  186.   // 起一个健康检查的http服务
  187.     if s.HealthzPort > 0 { 
  188.         healthz.DefaultHealthz()
  189.         go wait.Until(func() { 
  190.             err := http.ListenAndServe(net.JoinHostPort(s.HealthzBindAddress, strconv.Itoa(int(s.HealthzPort))), nil)
  191.             if err != nil { 
  192.                 klog.Errorf("Starting health server failed: %v", err)
  193.             }
  194.         }, 5*time.Second, wait.NeverStop)
  195.     }
  197.     if s.RunOnce { 
  198.         return nil
  199.     }
  201.     // If systemd is used, notify it that we have started
  202.     go daemon.SdNotify(false, "READY=1")
  204.     select { 
  205.     case <-done:
  206.         break
  207.     case <-stopCh:
  208.         break
  209.     }
  211.     return nil
  212. }

OOMAdjuster

–> pkg/util/oom/oom.go:22上面提到的oom判定器，这里分析一下，这个结构体有三个方法:

  1. // 这里目前用的还是结构体，看todo描述是后面要改成interface
  2. // TODO: make this an interface, and inject a mock ioutil struct for testing.
  3. type OOMAdjuster struct { 
  4.     pidLister                 func(cgroupName string) ([]int, error)
  5.     ApplyOOMScoreAdj          func(pid int, oomScoreAdj int) error
  6.     ApplyOOMScoreAdjContainer func(cgroupName string, oomScoreAdj, maxTries int) error
  7. }

–> pkg/util/oom/oom_linux.go:35实现方法

  1. func NewOOMAdjuster() *OOMAdjuster { 
  2.     oomAdjuster := &OOMAdjuster{ 
  3.         pidLister:        getPids,
  4.         ApplyOOMScoreAdj: applyOOMScoreAdj,
  5.     }
  6.     oomAdjuster.ApplyOOMScoreAdjContainer = oomAdjuster.applyOOMScoreAdjContainer
  7.     return oomAdjuster
  8. }
  9. // 获取cgroup下所有进程的pid
  10. func getPids(cgroupName string) ([]int, error) { 
  11.     return cmutil.GetPids(filepath.Join("/", cgroupName))
  12. }
  14. // 修改oom分数，在linux下即是修改/proc/<pid>/oom_score_adj对应的值，当内存紧张时由linux系统的oom机制去杀掉oom score最高的进程，默认情况下是使用内存越多的进程oom score越高越容易被kill，applyOOMScoreAdj函数就是用来修改oom score的。
  16. // Writes 'value' to /proc/<pid>/oom_score_adj. PID = 0 means self
  17. // Returns os.ErrNotExist if the `pid` does not exist.
  18. func applyOOMScoreAdj(pid int, oomScoreAdj int) error { 
  19.     if pid < 0 { 
  20.         return fmt.Errorf("invalid PID %d specified for oom_score_adj", pid)
  21.     }
  23.     var pidStr string
  24.     if pid == 0 { 
  25.         pidStr = "self"
  26.     } else { 
  27.         pidStr = strconv.Itoa(pid)
  28.     }
  30.     maxTries := 2
  31.     oomScoreAdjPath := path.Join("/proc", pidStr, "oom_score_adj")
  32.     value := strconv.Itoa(oomScoreAdj)
  33.     klog.V(4).Infof("attempting to set %q to %q", oomScoreAdjPath, value)
  34.     var err error
  35.     for i := 0; i < maxTries; i++ { 
  36.         err = ioutil.WriteFile(oomScoreAdjPath, []byte(value), 0700)
  37.         if err != nil { 
  38.             if os.IsNotExist(err) { 
  39.                 klog.V(2).Infof("%q does not exist", oomScoreAdjPath)
  40.                 return os.ErrNotExist
  41.             }
  43.             klog.V(3).Info(err)
  44.             time.Sleep(100 * time.Millisecond)
  45.             continue
  46.         }
  47.         return nil
  48.     }
  49.     if err != nil { 
  50.         klog.V(2).Infof("failed to set %q to %q: %v", oomScoreAdjPath, value, err)
  51.     }
  52.     return err
  53. }
  55. // 修改整个容器的oom评分，即修改某个cgroup下所有进程的评分，getPids取得所有pid遍历执行applyOOMScoreAdj
  56. // Writes 'value' to /proc/<pid>/oom_score_adj for all processes in cgroup cgroupName.
  57. // Keeps trying to write until the process list of the cgroup stabilizes, or until maxTries tries.
  58. func (oomAdjuster *OOMAdjuster) applyOOMScoreAdjContainer(cgroupName string, oomScoreAdj, maxTries int) error { 
  59.     adjustedProcessSet := make(map[int]bool)
  60.     for i := 0; i < maxTries; i++ { 
  61.         continueAdjusting := false
  62.         pidList, err := oomAdjuster.pidLister(cgroupName)
  63.         if err != nil { 
  64.             if os.IsNotExist(err) { 
  65.                 // Nothing to do since the container doesn't exist anymore.
  66.                 return os.ErrNotExist
  67.             }
  68.             continueAdjusting = true
  69.             klog.V(10).Infof("Error getting process list for cgroup %s: %+v", cgroupName, err)
  70.         } else if len(pidList) == 0 { 
  71.             klog.V(10).Infof("Pid list is empty")
  72.             continueAdjusting = true
  73.         } else { 
  74.             for _, pid := range pidList { 
  75.                 if !adjustedProcessSet[pid] { 
  76.                     klog.V(10).Infof("pid %d needs to be set", pid)
  77.                     if err = oomAdjuster.ApplyOOMScoreAdj(pid, oomScoreAdj); err == nil { 
  78.                         adjustedProcessSet[pid] = true
  79.                     } else if err == os.ErrNotExist { 
  80.                         continue
  81.                     } else { 
  82.                         klog.V(10).Infof("cannot adjust oom score for pid %d - %v", pid, err)
  83.                         continueAdjusting = true
  84.                     }
  85.                     // Processes can come and go while we try to apply oom score adjust value. So ignore errors here.
  86.                 }
  87.             }
  88.         }
  89.         if !continueAdjusting { 
  90.             return nil
  91.         }
  92.         // There's a slight race. A process might have forked just before we write its OOM score adjust.
  93.         // The fork might copy the parent process's old OOM score, then this function might execute and
  94.         // update the parent's OOM score, but the forked process id might not be reflected in cgroup.procs
  95.         // for a short amount of time. So this function might return without changing the forked process's
  96.         // OOM score. Very unlikely race, so ignoring this for now.
  97.     }
  98.     return fmt.Errorf("exceeded maxTries, some processes might not have desired OOM score")
  99. }

RunKubelet

–> cmd/kubelet/app/server.go:955 回到主线

  1. func RunKubelet(kubeServer *options.KubeletServer, kubeDeps *kubelet.Dependencies, runOnce bool) error { 
  2.     ...
  4.   // 这里的几个source都是"*"，意为接收api/file/http来源的pod更新
  5.     hostNetworkSources, err := kubetypes.GetValidatedSources(kubeServer.HostNetworkSources)
  6.     if err != nil { 
  7.         return err
  8.     }
  10.     hostPIDSources, err := kubetypes.GetValidatedSources(kubeServer.HostPIDSources)
  11.     if err != nil { 
  12.         return err
  13.     }
  15.     hostIPCSources, err := kubetypes.GetValidatedSources(kubeServer.HostIPCSources)
  16.     if err != nil { 
  17.         return err
  18.     }
  20.     privilegedSources := capabilities.PrivilegedSources{ 
  21.         HostNetworkSources: hostNetworkSources,
  22.         HostPIDSources:     hostPIDSources,
  23.         HostIPCSources:     hostIPCSources,
  24.     }
  25.     capabilities.Setup(kubeServer.AllowPrivileged, privilegedSources, 0)
  27.     credentialprovider.SetPreferredDockercfgPath(kubeServer.RootDirectory)
  28.     klog.V(2).Infof("Using root directory: %v", kubeServer.RootDirectory)
  30.     if kubeDeps.OSInterface == nil { 
  31.         kubeDeps.OSInterface = kubecontainer.RealOS{ }
  32.     }
  34.   // kubelet初始化，这个函数比较复杂，下面拎出来分析
  35.     k, err := createAndInitKubelet(&kubeServer.KubeletConfiguration,
  36.         kubeDeps,
  37.         &kubeServer.ContainerRuntimeOptions,
  38.         kubeServer.ContainerRuntime,
  39.         kubeServer.RuntimeCgroups,
  40.         kubeServer.HostnameOverride,
  41.         kubeServer.NodeIP,
  42.         kubeServer.ProviderID,
  43.         kubeServer.CloudProvider,
  44.         kubeServer.CertDirectory,
  45.         kubeServer.RootDirectory,
  46.         kubeServer.RegisterNode,
  47.         kubeServer.RegisterWithTaints,
  48.         kubeServer.AllowedUnsafeSysctls,
  49.         kubeServer.RemoteRuntimeEndpoint,
  50.         kubeServer.RemoteImageEndpoint,
  51.         kubeServer.ExperimentalMounterPath,
  52.         kubeServer.ExperimentalKernelMemcgNotification,
  53.         kubeServer.ExperimentalCheckNodeCapabilitiesBeforeMount,
  54.         kubeServer.ExperimentalNodeAllocatableIgnoreEvictionThreshold,
  55.         kubeServer.MinimumGCAge,
  56.         kubeServer.MaxPerPodContainerCount,
  57.         kubeServer.MaxContainerCount,
  58.         kubeServer.MasterServiceNamespace,
  59.         kubeServer.RegisterSchedulable,
  60.         kubeServer.NonMasqueradeCIDR,
  61.         kubeServer.KeepTerminatedPodVolumes,
  62.         kubeServer.NodeLabels,
  63.         kubeServer.SeccompProfileRoot,
  64.         kubeServer.BootstrapCheckpointPath,
  65.         kubeServer.NodeStatusMaxImages)
  66.     if err != nil { 
  67.         return fmt.Errorf("failed to create kubelet: %v", err)
  68.     }
  70.     // NewMainKubelet should have set up a pod source config if one didn't exist
  71.     // when the builder was run. This is just a precaution.
  72.     if kubeDeps.PodConfig == nil { 
  73.         return fmt.Errorf("failed to create kubelet, pod source config was nil")
  74.     }
  75.     podCfg := kubeDeps.PodConfig
  77.     rlimit.RlimitNumFiles(uint64(kubeServer.MaxOpenFiles))
  79.     // 只运行一次处理完pod就退出
  80.     if runOnce { 
  81.         if _, err := k.RunOnce(podCfg.Updates()); err != nil { 
  82.             return fmt.Errorf("runonce failed: %v", err)
  83.         }
  84.         klog.Info("Started kubelet as runonce")
  85.     } else { 
  86.     // 正常运行
  87.         startKubelet(k, podCfg, &kubeServer.KubeletConfiguration, kubeDeps, kubeServer.EnableServer)
  88.         klog.Info("Started kubelet")
  89.     }
  90.     return nil
  91. }

createAndInitKubelet

–> cmd/kubelet/app/server.go:1078，这里主要走到NewMainKubelet函数:

–> pkg/kubelet/kubelet.go:326 God！这个函数简直了，500+行代码…挑重要的说一下吧

  1. func NewMainKubelet(...) (...) { 
  2.   ...
  3.   // 加载service informer
  4.   serviceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{ cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
  5.     if kubeDeps.KubeClient != nil { 
  6.         serviceLW := cache.NewListWatchFromClient(kubeDeps.KubeClient.CoreV1().RESTClient(), "services", metav1.NamespaceAll, fields.Everything())
  7.         r := cache.NewReflector(serviceLW, &v1.Service{ }, serviceIndexer, 0)
  8.         go r.Run(wait.NeverStop)
  9.     }
  10.     serviceLister := corelisters.NewServiceLister(serviceIndexer)
  12.   // 加载node informer
  13.     nodeIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{ })
  14.     if kubeDeps.KubeClient != nil { 
  15.         fieldSelector := fields.Set{ api.ObjectNameField: string(nodeName)}.AsSelector()
  16.         nodeLW := cache.NewListWatchFromClient(kubeDeps.KubeClient.CoreV1().RESTClient(), "nodes", metav1.NamespaceAll, fieldSelector)
  17.         r := cache.NewReflector(nodeLW, &v1.Node{ }, nodeIndexer, 0)
  18.         go r.Run(wait.NeverStop)
  19.     }
  20.     nodeInfo := &predicates.CachedNodeInfo{ NodeLister: corelisters.NewNodeLister(nodeIndexer)}
  22.   ...
  24.   // secretManager和configMapManager初始化，因为这两者被使用都是需要往容器内挂载目录的，需要kubelet来参与
  25.   var secretManager secret.Manager
  26.     var configMapManager configmap.Manager
  27.     switch kubeCfg.ConfigMapAndSecretChangeDetectionStrategy { 
  28.     case kubeletconfiginternal.WatchChangeDetectionStrategy:
  29.         secretManager = secret.NewWatchingSecretManager(kubeDeps.KubeClient)
  30.         configMapManager = configmap.NewWatchingConfigMapManager(kubeDeps.KubeClient)
  31.     case kubeletconfiginternal.TTLCacheChangeDetectionStrategy:
  32.         secretManager = secret.NewCachingSecretManager(
  33.             kubeDeps.KubeClient, manager.GetObjectTTLFromNodeFunc(klet.GetNode))
  34.         configMapManager = configmap.NewCachingConfigMapManager(
  35.             kubeDeps.KubeClient, manager.GetObjectTTLFromNodeFunc(klet.GetNode))
  36.     case kubeletconfiginternal.GetChangeDetectionStrategy:
  37.         secretManager = secret.NewSimpleSecretManager(kubeDeps.KubeClient)
  38.         configMapManager = configmap.NewSimpleConfigMapManager(kubeDeps.KubeClient)
  39.     default:
  40.         return nil, fmt.Errorf("unknown configmap and secret manager mode: %v", kubeCfg.ConfigMapAndSecretChangeDetectionStrategy)
  41.     }
  43.     klet.secretManager = secretManager
  44.     klet.configMapManager = configMapManager
  46.   // 初始化存活探针管理器
  47.   klet.livenessManager = proberesults.NewManager()
  49.   //专为dockershim开辟的网络插件集 
  50.     pluginSettings := dockershim.NetworkPluginSettings{ 
  51.         HairpinMode:        kubeletconfiginternal.HairpinMode(kubeCfg.HairpinMode),
  52.         NonMasqueradeCIDR:  nonMasqueradeCIDR,
  53.         PluginName:         crOptions.NetworkPluginName,
  54.         PluginConfDir:      crOptions.CNIConfDir,  // 默认在/etc/cni/net.d/下存放cni配置文件
  55.         PluginBinDirString: crOptions.CNIBinDir,   // 默认在/opt/cni/bin/下存放cni二进制文件，如bridge/tuning/vlan/dhcp/macvlan等等
  56.         MTU:                int(crOptions.NetworkPluginMTU), // 网卡mtu
  57.     }  
  59.   // kubelet相关运行时初始化 
  60.     runtime, err := kuberuntime.NewKubeGenericRuntimeManager(
  61.         kubecontainer.FilterEventRecorder(kubeDeps.Recorder),
  62.         klet.livenessManager,
  63.         seccompProfileRoot,
  64.         containerRefManager,
  65.         machineInfo,
  66.         klet,
  67.         kubeDeps.OSInterface,
  68.         klet,
  69.         httpClient,
  70.         imageBackOff,
  71.         kubeCfg.SerializeImagePulls,
  72.         float32(kubeCfg.RegistryPullQPS),
  73.         int(kubeCfg.RegistryBurst),
  74.         kubeCfg.CPUCFSQuota,
  75.         kubeCfg.CPUCFSQuotaPeriod,
  76.         runtimeService,
  77.         imageService,
  78.         kubeDeps.ContainerManager.InternalContainerLifecycle(),
  79.         legacyLogProvider,
  80.         klet.runtimeClassManager,
  81.     )
  82.     if err != nil { 
  83.         return nil, err
  84.     }
  85.     klet.containerRuntime = runtime
  86.     klet.streamingRuntime = runtime
  87.     klet.runner = runtime
  89.     runtimeCache, err := kubecontainer.NewRuntimeCache(klet.containerRuntime)
  90.     if err != nil { 
  91.         return nil, err
  92.     }
  93.     klet.runtimeCache = runtimeCache  
  95.   // pleg初始化(Pod Lifecycle Event Generator)
  96.   klet.pleg = pleg.NewGenericPLEG(klet.containerRuntime, plegChannelCapacity,      plegRelistPeriod, klet.podCache, clock.RealClock{ })
  98.   // pod workQueue初始化
  99.   klet.workQueue = queue.NewBasicWorkQueue(klet.clock)
  100.   // pod worker初始化，worker从workQueue中取队首，根据指令对pod进行相应的直接操作，另外还有更新pod cache的操作
  101.     klet.podWorkers = newPodWorkers(klet.syncPod, kubeDeps.Recorder, klet.workQueue, klet.resyncInterval, backOffPeriod, klet.podCache)
  103.     klet.backOff = flowcontrol.NewBackOff(backOffPeriod, MaxContainerBackOff)
  104.     klet.podKillingCh = make(chan *kubecontainer.PodPair, podKillingChannelCapacity)
  106.     // 初始化驱逐管理器
  107.     evictionManager, evictionAdmitHandler := eviction.NewManager(klet.resourceAnalyzer, evictionConfig, killPodNow(klet.podWorkers, kubeDeps.Recorder), klet.podManager.GetMirrorPodByPod, klet.imageManager, klet.containerGC, kubeDeps.Recorder, nodeRef, klet.clock)
  109.     klet.evictionManager = evictionManager
  110.     klet.admitHandlers.AddPodAdmitHandler(evictionAdmitHandler)
  112.   ...
  113. }

再次回到主线，进入最后的k.Run()函数循环逻辑：

–> cmd/kubelet/app/server.go:1058

  1. func startKubelet(k kubelet.Bootstrap, podCfg *config.PodConfig, kubeCfg *kubeletconfiginternal.KubeletConfiguration, kubeDeps *kubelet.Dependencies, enableServer bool) { 
  2.     // start the kubelet
  3.   // 循环执行kubelet的工作逻辑k.Run()方法
  4.     go wait.Until(func() { 
  5.         k.Run(podCfg.Updates())
  6.     }, 0, wait.NeverStop)
  8.     // start the kubelet server
  9.   // 提供诸如/metrics /health等api
  10.     if enableServer { 
  11.         go k.ListenAndServe(net.ParseIP(kubeCfg.Address), uint(kubeCfg.Port), kubeDeps.TLSOptions, kubeDeps.Auth, kubeCfg.EnableDebuggingHandlers, kubeCfg.EnableContentionProfiling)
  13.     }
  14.   // 配置查询api
  15.     if kubeCfg.ReadOnlyPort > 0 { 
  16.         go k.ListenAndServeReadOnly(net.ParseIP(kubeCfg.Address), uint(kubeCfg.ReadOnlyPort))
  17.     }
  18.     if utilfeature.DefaultFeatureGate.Enabled(features.KubeletPodResources) { 
  19.         go k.ListenAndServePodResources()
  20.     }
  21. }

wait.Until()循环执行函数前面的文章中已经分析过多次了，不再赘述，来分析一下k.Run(podCfg.Updates())传的实参是什么:

–> pkg/kubelet/config/config.go:105

  1. func (c *PodConfig) Updates() <-chan kubetypes.PodUpdate {
  2.    return c.updates
  3. }

接着看c.updates --> pkg/kubelet/config/config.go:58

  1. type PodConfig struct { 
  2.    pods *podStorage
  3.    mux  *config.Mux
  5.    // the channel of denormalized changes passed to listeners
  6.    updates chan kubetypes.PodUpdate
  8.    // contains the list of all configured sources
  9.    sourcesLock       sync.Mutex
  10.    sources           sets.String
  11.    checkpointManager checkpointmanager.CheckpointManager
  12. }

–> pkg/kubelet/types/pod_update.go:80

  1. type PodUpdate struct { 
  2.    Pods   []*v1.Pod
  3.    Op     PodOperation
  4.    Source string
  5. }

猜测是将pod的写(删查改)请求转换成结构体，放入chan中，然后由k.Run()方法来处理这些写请求，k.Run()的实现在这里``pkg/kubelet/kubelet.go:1382`，留作下回分析，本篇启动流程篇到此结束。

小结

kubelet的源码果真是相当的复杂，一个函数动辄数百行，也难怪，毕竟作为daemon端执行数据平面工作的它要承担着很多职责，先到这吧

发表评论

表情：
评论列表 （有 0 条评论，121人围观）

还没有评论，来说两句吧...

相关阅读

    相关 SpringBoot2.1.1启动流程源码分析

    使用springboot已经有一年多了，但是一直停留的在整合，项目能够正常跑的阶段。年底了。抽点时间研究一下Springboot源码 ，这肯定会对以后的工作有帮助。今天我们从s

    悠悠悠悠/ 0 赞/ 385 阅读