深入kubelet：Kubernetes节点代理源码剖析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

// Kubelet 是运行在每个节点上的主要kubelet实现
// 文件路径: pkg/kubelet/kubelet.go
type Kubelet struct {
    // 基础配置
    hostname                      string
    nodeName                      types.NodeName
    runtimeState                  *runtimeState
    kubeClient                    clientset.Interface
    heartbeatClient              clientset.Interface
    
    // 运行时管理
    containerRuntime              kubecontainer.Runtime
    streamingRuntime             kubecontainer.StreamingRuntime
    runner                       kubecontainer.CommandRunner
    
    // HTTP服务器
    httpServer                   *httpserver.Server
    
    // 管理器组件
    podManager                   kubepod.Manager
    statusManager               status.Manager
    probeManager                prober.Manager
    volumeManager               volumemanager.VolumeManager
    evictionManager             eviction.Manager
    
    // 监控组件
    cadvisor                    cadvisor.Interface
    
    // 配置和策略
    kubeletConfiguration        *kubeletconfiginternal.KubeletConfiguration
    nodeStatusReportFrequency   time.Duration
    nodeStatusUpdateFrequency   time.Duration
    
    // 同步和控制
    syncLoopMonitor             atomic.Value
    podWorkers                  PodWorkers
    workQueue                   queue.WorkQueue
    
    // 网络和存储
    networkPlugin               network.NetworkPlugin
    volumePluginMgr            *volume.VolumePluginMgr
    
    // 资源管理
    containerGCPolicy          kubecontainer.GCPolicy
    imageGCPolicy              images.ImageGCPolicy
    diskSpacePolicy            DiskSpacePolicy
    
    // 安全和认证
    tlsOptions                 *server.TLSOptions
    auth                       server.AuthInterface
    
    // 状态和缓存
    lastObservedNodeAddressesMux sync.RWMutex
    lastObservedNodeAddresses   []v1.NodeAddress
    
    // 驱逐相关
    evictionAdmitHandler        lifecycle.PodAdmitHandler
    
    // 生命周期管理
    shutdownManager             shutdown.Manager
    
    // 插件管理
    pluginManager              pluginmanager.PluginManager
    
    // 设备管理
    deviceManager              devicemanager.Manager
    
    // CPU管理
    cpuManager                 cpumanager.Manager
    
    // 内存管理
    memoryManager              memorymanager.Manager
    
    // 拓扑管理
    topologyManager            topologymanager.Manager
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

// podWorkers 管理所有Pod的工作协程
// 文件路径: pkg/kubelet/pod_workers.go
type podWorkers struct {
    // 每个Pod的工作器
    podUpdates map[types.UID]chan podWork
    
    // 工作器状态
    isWorking map[types.UID]bool
    
    // 最后已知的Pod状态
    lastUndeliveredWorkUpdate map[types.UID]podWork
    
    // 同步锁
    podLock sync.Mutex
    
    // kubelet实例引用
    kubelet *Kubelet
    
    // 工作队列
    workQueue queue.WorkQueue
    
    // 工作器状态跟踪
    podSyncStatuses map[types.UID]podSyncStatus
}

// podWork 表示需要对Pod执行的工作
type podWork struct {
    // Pod对象
    pod *v1.Pod
    
    // 镜像Pod（用于静态Pod）
    mirrorPod *v1.Pod
    
    // 更新类型
    updateType kubetypes.SyncPodType
    
    // 工作完成回调
    onCompleteFunc func()
    
    // 工作开始时间
    workType podWorkType
}

// podSyncStatus Pod同步状态
type podSyncStatus struct {
    // 同步状态
    syncedAt time.Time
    
    // 是否正在同步
    working bool
    
    // 是否需要终止
    terminatingAt time.Time
    
    // 是否已删除
    deleted bool
    
    // 最后同步结果
    lastSyncResult error
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// main kubelet主函数入口
// 文件路径: cmd/kubelet/kubelet.go
func main() {
    // 创建kubelet命令对象
    // 包含所有启动参数和配置选项
    command := app.NewKubeletCommand(context.Background())
    
    // 使用component-base的CLI运行器执行命令
    // 提供统一的命令行处理、信号处理和优雅关闭机制
    code := cli.Run(command)
    
    // 以返回的退出码结束程序
    os.Exit(code)
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104

// NewKubeletCommand 创建kubelet命令
// 文件路径: cmd/kubelet/app/server.go
func NewKubeletCommand(ctx context.Context) *cobra.Command {
    // 清理函数列表，用于程序退出时的清理工作
    cleanFlagSet := pflag.NewFlagSet("clean", pflag.ContinueOnError)
    cleanFlagSet.SetNormalizeFunc(cliflag.WordSepNormalizeFunc)
    
    // kubelet配置选项
    kubeletFlags := options.NewKubeletFlags()
    kubeletConfig, err := options.NewKubeletConfiguration()
    if err != nil {
        klog.ErrorS(err, "Failed to create a new kubelet configuration")
        os.Exit(1)
    }

    cmd := &cobra.Command{
        Use: componentKubelet,
        Long: `The kubelet is the primary "node agent" that runs on each
node. It can register the node with the apiserver using one of: the hostname; a flag to
override the hostname; or specific logic for a cloud provider.

The kubelet works in terms of a PodSpec. A PodSpec is a YAML or JSON object
that describes a pod. The kubelet takes a set of PodSpecs that are provided through
various mechanisms (primarily through the apiserver) and ensures that the containers
described in those PodSpecs are running and healthy. The kubelet doesn't manage
containers which were not created by Kubernetes.

Other than from an PodSpec from the apiserver, there are two ways that a container
manifest can be provided to the Kubelet.

File: Path passed as a flag on the command line. Files under this path will be monitored
periodically for updates. The monitoring period is 20s by default and is configurable
via a flag.

HTTP endpoint: HTTP endpoint passed as a parameter on the command line. This endpoint
is checked every 20 seconds (also configurable with a flag).`,
        
        // 禁用在命令错误时打印使用说明
        SilenceUsage: true,
        
        // 主运行函数
        RunE: func(cmd *cobra.Command, args []string) error {
            // 验证初始标志
            if err := cleanFlagSet.Parse(args); err != nil {
                return fmt.Errorf("failed to parse clean flag: %w", err)
            }
            
            // 短路清理标志
            if cleanFlagSet.NArg() > 0 {
                return fmt.Errorf("%q does not take any arguments, got %q", cmd.CommandPath(), cleanFlagSet.Args())
            }

            // 设置特性门控
            if err := utilfeature.DefaultMutableFeatureGate.SetFromMap(kubeletConfig.FeatureGates); err != nil {
                return fmt.Errorf("failed to set feature gates from initial flags-based config: %w", err)
            }

            // 验证初始kubelet配置
            if err := kubeletconfigvalidation.ValidateKubeletConfiguration(kubeletConfig, utilfeature.DefaultFeatureGate); err != nil {
                return fmt.Errorf("failed to validate kubelet configuration, error: %w, path: %s", err, kubeletConfig)
            }

            // 加载kubelet配置文件
            if configFile := kubeletFlags.KubeletConfigFile; len(configFile) > 0 {
                kubeletConfig, err = loadConfigFile(configFile)
                if err != nil {
                    return fmt.Errorf("failed to load kubelet config file, error: %w, path: %s", err, configFile)
                }
                
                // 重新设置特性门控
                if err := utilfeature.DefaultMutableFeatureGate.SetFromMap(kubeletConfig.FeatureGates); err != nil {
                    return fmt.Errorf("failed to set feature gates from config file: %w", err)
                }
            }

            // 构造kubelet服务器配置
            kubeletServer := &options.KubeletServer{
                KubeletFlags:         *kubeletFlags,
                KubeletConfiguration: *kubeletConfig,
            }

            // 使用动态kubelet配置，如果启用的话
            if dynamicConfigDir := kubeletFlags.DynamicConfigDir.Value(); len(dynamicConfigDir) > 0 {
                kubeletServer, err = getDynamicKubeletConfig(kubeletServer.KubeletConfiguration, dynamicConfigDir, kubeletServer.KubeletFlags)
                if err != nil {
                    return fmt.Errorf("failed to get dynamic kubelet config: %w", err)
                }
            }

            // 运行kubelet
            return Run(ctx, kubeletServer)
        },
    }

    // 添加kubelet标志
    kubeletFlags.AddFlags(cleanFlagSet)
    options.AddKubeletConfigFlags(cleanFlagSet, kubeletConfig)
    options.AddGlobalFlags(cleanFlagSet)
    cleanFlagSet.BoolP("help", "h", false, fmt.Sprintf("help for %s", cmd.Name()))

    // 将清理标志集合添加到命令中
    cmd.Flags().AddFlagSet(cleanFlagSet)
    return cmd
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

// Run 运行kubelet的主函数
// 文件路径: cmd/kubelet/app/server.go
func Run(ctx context.Context, s *options.KubeletServer) error {
    // 构造kubelet依赖项
    kubeletDeps, err := UnsecuredDependencies(s, utilfeature.DefaultFeatureGate)
    if err != nil {
        return err
    }

    // 添加认证/授权
    if s.KubeletConfiguration.Authentication.Anonymous.Enabled || 
       (len(s.KubeletConfiguration.Authentication.X509.ClientCAFile) > 0) ||
       s.KubeletConfiguration.Authentication.Webhook.Enabled {
        kubeletDeps.Auth, err = BuildAuth(s.KubeletConfiguration.Authentication, s.KubeletConfiguration.Authorization, s.KubeletFlags.ExperimentalMounts)
        if err != nil {
            return err
        }
    }

    // 运行kubelet
    return RunKubelet(s, kubeletDeps, s.RunOnce)
}

// RunKubelet 运行kubelet实例
func RunKubelet(kubeServer *options.KubeletServer, kubeletDeps *kubelet.Dependencies, runOnce bool) error {
    // 获取主机名
    hostname, err := nodeutil.GetHostname(kubeServer.HostnameOverride)
    if err != nil {
        return err
    }
    
    // 设置节点名称
    nodeName, err := getNodeName(kubeletDeps.Cloud, hostname)
    if err != nil {
        return err
    }
    
    // 检查权限
    if err := checkPermissions(); err != nil {
        klog.ErrorS(err, "kubelet running with insufficient permissions")
    }

    // 设置资源限制
    if kubeServer.KubeletConfiguration.KernelMemcgNotification {
        if err := watchForLowMemory(); err != nil {
            klog.ErrorS(err, "Failed to start watching for low memory")
        }
    }

    // 创建并运行kubelet
    k, err := createAndInitKubelet(kubeServer, kubeletDeps, hostname, nodeName)
    if err != nil {
        return fmt.Errorf("failed to create kubelet: %w", err)
    }

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

// createAndInitKubelet 创建并初始化kubelet
func createAndInitKubelet(kubeServer *options.KubeletServer, kubeletDeps *kubelet.Dependencies, 
    hostname string, nodeName types.NodeName) (k kubelet.Bootstrap, err error) {
    
    // 创建kubelet实例
    k, err = kubelet.NewMainKubelet(
        &kubeServer.KubeletConfiguration,
        kubeletDeps,
        &kubeServer.ContainerRuntimeOptions,
        hostname,
        nodeName,
        kubeServer.NodeIP,
        kubeServer.ProviderID,
        kubeServer.CloudProvider,
        kubeServer.CertDirectory,
        kubeServer.RootDirectory,
        kubeServer.ImageCredentialProviderConfigFile,
        kubeServer.ImageCredentialProviderBinDir,
        kubeServer.RegisterNode,
        kubeServer.RegisterWithTaints,
        kubeServer.AllowedUnsafeSysctls,
        kubeServer.ExperimentalMounterPath,
        kubeServer.KernelMemcgNotification,
        kubeServer.ExperimentalCheckNodeCapabilitiesBeforeMount,
        kubeServer.ExperimentalNodeAllocatableIgnoreEvictionThreshold,
        kubeServer.MinimumGCAge,
        kubeServer.MaxPerPodContainerCount,
        kubeServer.MaxContainerCount,
        kubeServer.MasterServiceNamespace,
        kubeServer.RegisterSchedulable,
        kubeServer.KeepTerminatedPodVolumes,
        kubeServer.NodeLabels,
        kubeServer.SeccompDefault || kubeServer.KubeletConfiguration.SeccompDefault,
        kubeServer.NodeStatusMaxImages,
        kubeServer.KubeletConfiguration.LocalStorageCapacityIsolation,
    )
    if err != nil {
        return nil, err
    }

    // 初始化kubelet
    k.BirthCry()

    // 启动垃圾回收
    go k.StartGarbageCollection()

    return k, nil
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

// syncLoop 是kubelet的主要同步循环
// 文件路径: pkg/kubelet/kubelet.go
func (kl *Kubelet) syncLoop(ctx context.Context, updates <-chan kubetypes.PodUpdate, handler SyncHandler) {
    klog.InfoS("Starting kubelet main sync loop")
    
    // 同步循环监控器，用于检测循环是否正常运行
    syncTicker := time.NewTicker(time.Second)
    defer syncTicker.Stop()
    
    // 清理Pod定时器
    housekeepingTicker := time.NewTicker(housekeepingPeriod)
    defer housekeepingTicker.Stop()
    
    // Pod生命周期事件生成器定时器
    plegCh := kl.pleg.Watch()
    const (
        base   = 100 * time.Millisecond
        max    = 5 * time.Second
        factor = 2
    )
    duration := base
    
    // 无限循环处理各种事件
    for {
        select {
        case u := <-updates:
            // 处理Pod更新事件
            // 来源包括：API服务器、文件、HTTP端点等
            klog.V(2).InfoS("SyncLoop (UPDATE)", "source", u.Source, "pods", klog.KObjSlice(u.Pods))
            handler.HandlePodUpdates(u.Pods)
            
        case e := <-plegCh:
            // 处理Pod生命周期事件
            // PLEG (Pod Lifecycle Event Generator) 生成的事件
            if e.Type == pleg.ContainerStarted {
                // 更新最新的容器启动时间，用于驱逐管理器
                kl.lastContainerStartedTime.Add(e.ID, time.Now())
            }
            if isSyncPodWorthy(e) {
                // 如果事件值得同步，则触发Pod同步
                if pod, ok := kl.podManager.GetPodByUID(e.ID); ok {
                    klog.V(2).InfoS("SyncLoop (PLEG)", "pod", klog.KObj(pod), "event", e)
                    handler.HandlePodSyncs([]*v1.Pod{pod})
                } else {
                    // 如果找不到Pod，记录警告
                    klog.V(4).InfoS("SyncLoop (PLEG): pod does not exist, ignore irrelevant event", "event", e)
                }
            }
            
            if e.Type == pleg.ContainerDied {
                // 容器死亡事件，检查是否需要重启
                if containerID, ok := e.Data.(string); ok {
                    kl.cleanUpContainersInPod(e.ID, containerID)
                }
            }
            
        case <-syncTicker.C:
            // 定期同步所有Pod
            // 这是一个保底机制，确保即使没有事件也会定期检查Pod状态
            klog.V(6).InfoS("SyncLoop (SYNC)")
            handler.HandlePodSyncs(kl.getPodsToSync())
            
        case update := <-kl.livenessManager.Updates():
            // 处理存活性探针结果
            if update.Result == proberesults.Failure {
                handleProbeSync(kl, update, handler, "liveness", "unhealthy")
            }
            
        case update := <-kl.readinessManager.Updates():
            // 处理就绪性探针结果  
            ready := update.Result == proberesults.Success
            kl.statusManager.SetContainerReadiness(update.PodUID, update.ContainerID, ready)
            status := "ready"
            if !ready {
                status = "unready"
            }
            klog.V(1).InfoS("SyncLoop (probe)", "probe", "readiness", "status", status, "pod", klog.KRef(update.PodNamespace, update.PodName))
            
        case update := <-kl.startupManager.Updates():
            // 处理启动探针结果
            started := update.Result == proberesults.Success
            kl.statusManager.SetContainerStartup(update.PodUID, update.ContainerID, started)
            status := "started"
            if !started {
                status = "unhealthy"
            }
            klog.V(1).InfoS("SyncLoop (probe)", "probe", "startup", "status", status, "pod", klog.KRef(update.PodNamespace, update.PodName))
            
        case <-housekeepingTicker.C:
            // 定期清理工作
            // 包括：清理已终止的Pod、垃圾回收等
            klog.V(2).InfoS("SyncLoop (housekeeping)")
            if !kl.sourcesReady.AllReady() {
                // 如果Pod源还没有准备好，跳过清理工作
                klog.V(4).InfoS("SyncLoop (housekeeping, skipped): sources aren't ready yet")
            } else {
                start := time.Now()
                if err := handler.HandlePodCleanups(ctx); err != nil {
                    klog.ErrorS(err, "Failed cleaning pods")
                }
                duration := time.Since(start)
                if duration > housekeepingWarningDuration {
                    klog.ErrorS(nil, "Housekeeping took too long", "duration", duration, "threshold", housekeepingWarningDuration)
                }
            }
            
        case <-ctx.Done():
            // 上下文取消，退出循环
            klog.InfoS("SyncLoop (context cancelled)")
            return
        }
    }
}

// isSyncPodWorthy 判断PLEG事件是否值得同步Pod
func isSyncPodWorthy(e *pleg.PodLifecycleEvent) bool {
    // 只有以下事件类型才值得同步：
    // - 容器启动
    // - 容器死亡  
    // - 网络设置完成
    // - 网络拆除完成
    return e.Type == pleg.ContainerStarted ||
           e.Type == pleg.ContainerDied ||
           e.Type == pleg.NetworkSetupCompleted ||
           e.Type == pleg.NetworkTeardownCompleted
}

// getPodsToSync 获取需要同步的Pod列表
func (kl *Kubelet) getPodsToSync() []*v1.Pod {
    allPods := kl.podManager.GetPods()
    podUIDs := make(map[types.UID]empty, len(allPods))
    for _, pod := range allPods {
        podUIDs[pod.UID] = empty{}
    }
    
    // 添加正在工作的Pod
    kl.podWorkers.ForEachWorker(func(podUID types.UID, syncStatus *podSyncStatus) {
        if syncStatus.working {
            podUIDs[podUID] = empty{}
        }
    })
    
    // 转换为Pod切片
    var pods []*v1.Pod
    for _, pod := range allPods {
        if _, exists := podUIDs[pod.UID]; exists {
            pods = append(pods, pod)
        }
    }
    return pods
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

// HandlePodUpdates 处理Pod更新
// 文件路径: pkg/kubelet/kubelet.go
func (kl *Kubelet) HandlePodUpdates(pods []*v1.Pod) {
    start := kl.clock.Now()
    for _, pod := range pods {
        // 更新Pod管理器中的Pod信息
        existing, found := kl.podManager.GetPodByUID(pod.UID)
        kl.podManager.UpdatePod(pod)
        
        if (found && !podutil.IsMirrorPod(existing) && podutil.IsMirrorPod(pod)) ||
           (found && podutil.IsMirrorPod(existing) && !podutil.IsMirrorPod(pod)) {
            // 镜像Pod状态变更，需要特殊处理
            kl.handleMirrorPod(pod, start)
        }
    }
}

// HandlePodSyncs 处理Pod同步请求
func (kl *Kubelet) HandlePodSyncs(pods []*v1.Pod) {
    start := kl.clock.Now()
    for _, pod := range pods {
        // 为每个Pod分派同步工作
        kl.podWorkers.UpdatePod(UpdatePodOptions{
            Pod:        pod,
            MirrorPod:  kl.podManager.GetMirrorPodByPod(pod),
            UpdateType: kubetypes.SyncPodSync,
            StartTime:  start,
        })
    }
}

// syncPod 是同步单个Pod的核心函数
func (kl *Kubelet) syncPod(ctx context.Context, updateType kubetypes.SyncPodType, pod *v1.Pod, mirrorPod *v1.Pod, podStatus *kubecontainer.PodStatus) (isTerminal bool, err error) {
    // 记录同步开始时间，用于性能监控
    start := time.Now()
    defer func() {
        metrics.PodSyncDuration.Observe(metrics.SinceInSeconds(start))
        metrics.DeprecatedPodSyncDuration.Observe(metrics.SinceInMicroseconds(start))
    }()
    
    // 生成Pod状态
    apiPodStatus := kl.generateAPIPodStatus(pod, podStatus)
    
    // 检查Pod是否可以在此节点上运行
    runnable := kl.canRunPod(pod)
    if !runnable.Admit {
        // 如果不能运行，拒绝Pod并返回
        kl.rejectPod(pod, runnable.Reason, runnable.Message)
        return true, fmt.Errorf("pod rejected: %s", runnable.Message)
    }
    
    // 更新Pod状态到API服务器
    kl.statusManager.SetPodStatus(pod, apiPodStatus)
    
    // 检查Pod是否应该被终止
    if !kl.podIsTerminated(pod) {
        // Pod还在运行，检查是否需要杀死
        if apiPodStatus.Phase == v1.PodFailed || apiPodStatus.Phase == v1.PodSucceeded || (pod.DeletionTimestamp != nil && notRunning(apiPodStatus.ContainerStatuses)) {
            klog.V(3).InfoS("Pod is terminated, but some containers are still running", "pod", klog.KObj(pod))
            return false, kl.killPod(pod, nil, podStatus, killPodReasonTerminated)
        }
    }
    
    // 如果Pod没有被删除且不是终止状态，继续同步
    if pod.DeletionTimestamp == nil {
        // 确保数据目录存在
        if err := kl.makePodDataDirs(pod); err != nil {
            klog.ErrorS(err, "Unable to make pod data directories for pod", "pod", klog.KObj(pod))
            return false, err
        }
        
        // 等待卷附加和挂载
        if !kl.podWorkers.IsPodTerminationRequested(pod.UID) {
            // 等待卷挂载完成
            if err := kl.volumeManager.WaitForAttachAndMount(pod); err != nil {
                kl.recorder.Eventf(pod, v1.EventTypeWarning, events.FailedMountVolume, "Unable to attach or mount volumes: %v", err)
                klog.ErrorS(err, "Unable to attach or mount volumes for pod; skipping pod", "pod", klog.KObj(pod))
                return false, err
            }
        }
        
        // 获取镜像拉取密钥
        pullSecrets := kl.getPullSecretsForPod(pod)
        
        // 调用容器运行时同步Pod
        result := kl.containerRuntime.SyncPod(ctx, pod, podStatus, pullSecrets, kl.backOff)
        kl.reasonCache.Update(pod.UID, result)
        if result.Error() != nil {
            // 如果同步失败，记录错误但不返回，因为状态管理器会处理
            klog.ErrorS(result.Error(), "Error syncing pod, skipping", "pod", klog.KObj(pod))
            return false, result.Error()
        }
    }
    
    return false, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// Runtime 容器运行时接口定义
// 文件路径: pkg/kubelet/container/runtime.go
type Runtime interface {
    // Type 返回容器运行时类型
    Type() string
    
    // Version 返回容器运行时版本信息
    Version(ctx context.Context) (Version, error)
    
    // APIVersion 返回容器运行时API版本
    APIVersion() (Version, error)
    
    // Status 返回容器运行时状态
    Status(ctx context.Context) (*RuntimeStatus, error)
    
    // GetPods 返回所有Pod列表
    GetPods(ctx context.Context, all bool) ([]*Pod, error)
    
    // SyncPod 同步Pod状态，这是最重要的方法
    SyncPod(ctx context.Context, pod *v1.Pod, podStatus *PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) PodSyncResult
    
    // KillPod 杀死Pod中的所有容器
    KillPod(ctx context.Context, pod *v1.Pod, runningPod Pod, gracePeriodOverride *int64) error
    
    // GetPodStatus 获取Pod状态
    GetPodStatus(ctx context.Context, uid types.UID, name, namespace string) (*PodStatus, error)
    
    // PullImage 拉取容器镜像
    PullImage(ctx context.Context, image ImageSpec, pullSecrets []v1.Secret, podSandboxConfig *runtimeapi.PodSandboxConfig) (string, error)
    
    // GetImageRef 获取镜像引用
    GetImageRef(ctx context.Context, image ImageSpec) (string, error)
    
    // ListImages 列出所有镜像
    ListImages(ctx context.Context) ([]Image, error)
    
    // RemoveImage 删除镜像
    RemoveImage(ctx context.Context, image ImageSpec) error
    
    // ImageStats 获取镜像统计信息
    ImageStats(ctx context.Context) (*ImageStats, error)
    
    // DeleteContainer 删除容器
    DeleteContainer(ctx context.Context, containerID ContainerID) error
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223

// kubeGenericRuntimeManager CRI运行时管理器
// 文件路径: pkg/kubelet/kuberuntime/kuberuntime_manager.go
type kubeGenericRuntimeManager struct {
    // CRI服务客户端
    runtimeService internalapi.RuntimeService
    imageService   internalapi.ImageService
    
    // 记录器
    recorder record.EventRecorder
    
    // 操作系统接口
    osInterface kubecontainer.OSInterface
    
    // 运行时助手
    runtimeHelper kubecontainer.RuntimeHelper
    
    // HTTP客户端（用于流式操作）
    httpClient types.HTTPClient
    
    // 流式运行时服务
    streamingRuntime kubecontainer.StreamingRuntime
    
    // 镜像拉取QPS限制
    imageBackOff *flowcontrol.Backoff
    
    // 序列化镜像拉取
    serializedImagePulls bool
    
    // 镜像拉取信号量
    imagePullSemaphore semaphore.Weighted
    
    // CPU CFS配额
    cpuCFSQuota bool
    
    // CPU CFS配额周期
    cpuCFSQuotaPeriod metav1.Duration
    
    // seccomp配置文件根目录
    seccompProfileRoot string
    
    // 内部生命周期事件
    internalLifecycle cm.InternalContainerLifecycle
    
    // 日志管理器
    logManager logs.ContainerLogManager
    
    // 运行时类管理器
    runtimeClassManager *runtimeclass.Manager
    
    // 内存交换行为
    memorySwapBehavior string
    
    // 获取容器和镜像文件系统信息
    getNodeAllocatableAbsolute func() v1.ResourceList
    
    // 内存QoS
    memoryQoSEnforced bool
}

// SyncPod 同步Pod，这是CRI运行时管理器的核心方法
func (m *kubeGenericRuntimeManager) SyncPod(ctx context.Context, pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult) {
    // 创建结果对象
    result = kubecontainer.NewSyncResult(kubecontainer.SyncPod, pod.Name)
    result.StartTime = time.Now()
    defer func() {
        result.EndTime = time.Now()
    }()
    
    // 第1步：计算沙箱和容器变更
    podContainerChanges := m.computePodActions(ctx, pod, podStatus)
    klog.V(3).InfoS("computePodActions got for pod", "podActions", podContainerChanges, "pod", klog.KObj(pod))
    
    if podContainerChanges.CreateSandbox {
        ref, err := ref.GetReference(legacyscheme.Scheme, pod)
        if err != nil {
            klog.ErrorS(err, "Couldn't make a ref to pod", "pod", klog.KObj(pod))
        }
        
        if podContainerChanges.SandboxID != "" {
            // 需要杀死现有的沙箱
            m.recorder.Eventf(ref, v1.EventTypeNormal, events.SandboxChanged, "Pod sandbox changed, it will be killed and re-created.")
        } else {
            // 创建新的沙箱
            klog.V(4).InfoS("Creating PodSandbox for pod", "pod", klog.KObj(pod))
            m.recorder.Eventf(ref, v1.EventTypeNormal, events.SandboxReceived, "Pod sandbox received")
        }
    }
    
    // 第2步：如果需要，杀死Pod沙箱
    if podContainerChanges.KillPod {
        if podContainerChanges.CreateSandbox {
            klog.V(4).InfoS("Stopping PodSandbox for pod, will start new one", "pod", klog.KObj(pod))
        } else {
            klog.V(4).InfoS("Stopping PodSandbox for pod", "pod", klog.KObj(pod))
        }
        
        // 杀死Pod中的所有容器
        if err := m.killPodWithSyncResult(ctx, pod, kubecontainer.ConvertPodStatusToRunningPod(m.runtimeName, podStatus), &result); err != nil {
            klog.ErrorS(err, "killPodWithSyncResult failed", "pod", klog.KObj(pod))
            return result
        }
    } else {
        // 第3步：杀死不需要的容器
        for containerID, containerInfo := range podContainerChanges.ContainersToKill {
            klog.V(3).InfoS("Killing unwanted container for pod", "containerName", containerInfo.name, "containerID", containerID, "pod", klog.KObj(pod))
            if err := m.killContainer(ctx, pod, containerID, containerInfo.name, containerInfo.message, containerInfo.reason, nil); err != nil {
                klog.ErrorS(err, "killContainer for pod failed", "containerName", containerInfo.name, "containerID", containerID, "pod", klog.KObj(pod))
                result.Fail(kubecontainer.ErrKillContainer, err.Error())
                return result
            }
        }
    }
    
    // 记录Pod沙箱状态变更的事件
    m.recordPodSandboxChange(pod, podContainerChanges)
    
    // 第4步：如果需要，创建沙箱
    podSandboxID := podContainerChanges.SandboxID
    if podContainerChanges.CreateSandbox {
        var msg string
        var err error
        
        klog.V(4).InfoS("Creating PodSandbox for pod", "pod", klog.KObj(pod))
        metrics.StartedPodsTotal.Inc()
        createSandboxResult := kubecontainer.NewSyncResult(kubecontainer.CreatePodSandbox, format.Pod(pod))
        result.AddSyncResult(createSandboxResult)
        
        // 创建Pod沙箱配置
        podSandboxConfig, err := m.generatePodSandboxConfig(pod, podContainerChanges.Attempt)
        if err != nil {
            message := fmt.Sprintf("Failed to generate sandbox config for pod %q: %v", format.Pod(pod), err)
            klog.ErrorS(err, "Failed to generate sandbox config for pod", "pod", klog.KObj(pod))
            createSandboxResult.Fail(kubecontainer.ErrConfigPodSandbox, message)
            m.recorder.Eventf(pod, v1.EventTypeWarning, events.FailedCreatePodSandBox, "Failed to create pod sandbox: %v", err)
            return result
        }
        
        // 创建Pod沙箱
        podSandboxID, err = m.runtimeService.RunPodSandbox(ctx, podSandboxConfig, runtimeHandler)
        if err != nil {
            message := fmt.Sprintf("Failed to create sandbox for pod %q: %v", format.Pod(pod), err)
            klog.ErrorS(err, "Failed to create sandbox for pod", "pod", klog.KObj(pod))
            createSandboxResult.Fail(kubecontainer.ErrCreatePodSandbox, message)
            m.recorder.Eventf(pod, v1.EventTypeWarning, events.FailedCreatePodSandBox, "Failed to create pod sandbox: %v", err)
            return result
        }
        klog.V(4).InfoS("Created PodSandbox for pod", "podSandboxID", podSandboxID, "pod", klog.KObj(pod))
        
        podSandboxStatus, err := m.runtimeService.PodSandboxStatus(ctx, podSandboxID, false)
        if err != nil {
            ref, referr := ref.GetReference(legacyscheme.Scheme, pod)
            if referr != nil {
                klog.ErrorS(referr, "Couldn't make a ref to pod", "pod", klog.KObj(pod))
            }
            klog.ErrorS(err, "Failed to get pod sandbox status; Skipping pod", "pod", klog.KObj(pod))
            result.Fail(kubecontainer.ErrPodSandboxStatus, err.Error())
            m.recorder.Eventf(ref, v1.EventTypeWarning, events.FailedStatusPodSandBox, "Unable to get pod sandbox status: %v", err)
            return result
        }
        
        // 设置Pod网络
        if !kubecontainer.IsHostNetworkPod(pod) {
            // 为Pod设置网络
            if err := m.runtimeHelper.SetUpPodNetwork(ctx, pod, podSandboxStatus.GetNetwork().GetNamespace(), podSandboxStatus.GetMetadata().GetUid(), podSandboxID); err != nil {
                message := fmt.Sprintf("Failed to setup network for pod %q using network plugins %q: %v", format.Pod(pod), m.runtimeHelper.GetPodNetworkPlugins(), err)
                
                klog.ErrorS(err, "Failed to setup network for pod using network plugins", "pod", klog.KObj(pod), "networkPlugins", m.runtimeHelper.GetPodNetworkPlugins())
                
                killResult := kubecontainer.NewSyncResult(kubecontainer.KillPodSandbox, format.Pod(pod))
                result.AddSyncResult(killResult)
                if killErr := m.runtimeService.StopPodSandbox(ctx, podSandboxID); killErr != nil {
                    killResult.Fail(kubecontainer.ErrKillPodSandbox, killErr.Error())
                    klog.ErrorS(killErr, "Failed to stop sandbox", "podSandboxID", podSandboxID)
                }
                
                createSandboxResult.Fail(kubecontainer.ErrSetupNetwork, message)
                m.recorder.Eventf(pod, v1.EventTypeWarning, events.FailedSetupNetwork, "Failed to setup network for pod using network plugins %q: %v", m.runtimeHelper.GetPodNetworkPlugins(), err)
                
                return result
            }
        }
    }
    
    // 获取Pod沙箱配置
    configPodSandboxResult := kubecontainer.NewSyncResult(kubecontainer.ConfigPodSandbox, podSandboxID)
    result.AddSyncResult(configPodSandboxResult)
    podSandboxConfig, err := m.generatePodSandboxConfig(pod, podContainerChanges.Attempt)
    if err != nil {
        message := fmt.Sprintf("Failed to generate sandbox config for pod %q: %v", format.Pod(pod), err)
        klog.ErrorS(err, "Failed to generate sandbox config for pod", "pod", klog.KObj(pod))
        configPodSandboxResult.Fail(kubecontainer.ErrConfigPodSandbox, message)
        return result
    }
    
    // 第5步：获取Pod IP
    podIPs := m.determinePodSandboxIPs(pod.Namespace, pod.Name, podSandboxStatus)
    
    // 第6步：启动init容器
    if container := podContainerChanges.NextInitContainerToStart; container != nil {
        // 启动下一个init容器
        if err := m.startContainer(ctx, podSandboxID, podSandboxConfig, container.container, pod, podStatus, pullSecrets, podIPs, kubecontainer.ContainerTypeInit); err != nil {
            klog.ErrorS(err, "Container start failed", "pod", klog.KObj(pod), "podSandboxID", podSandboxID, "containerName", container.container.Name)
            result.Fail(err)
            return result
        }
        
        // 成功启动init容器后，返回，等待下次同步
        klog.V(4).InfoS("Completed init container for pod", "containerName", container.container.Name, "pod", klog.KObj(pod))
        return result
    }
    
    // 第7步：启动常规容器
    for _, idx := range podContainerChanges.ContainersToStart {
        container := &pod.Spec.Containers[idx]
        if err := m.startContainer(ctx, podSandboxID, podSandboxConfig, container, pod, podStatus, pullSecrets, podIPs, kubecontainer.ContainerTypeRegular); err != nil {
            klog.ErrorS(err, "Container start failed", "pod", klog.KObj(pod), "podSandboxID", podSandboxID, "containerName", container.Name)
            result.Fail(err)
            continue
        }
    }
    
    return result
}