kubernetes源码分析 kubelet

news2025/2/21 9:41:00

简介

从官方的架构图中很容易就能找到 kubelet

执行 kubelet -h 看到 kubelet 的功能介绍:

  • kubelet 是每个 Node 节点上都运行的主要“节点代理”。使用如下的一个向 apiserver 注册 Node 节点:主机的 hostname;覆盖 host 的参数;或者云提供商指定的逻辑。

  • kubelet 基于 PodSpec 工作。PodSpec 是用 YAML 或者 JSON 对象来描述 Pod。Kubelet 接受通过各种机制(主要是 apiserver)提供的一组 PodSpec,并确保里面描述的容器良好运行。

除了由 apiserver 提供 PodSpec,还可以通过以下方式提供:

  • 文件

  • HTTP 端点

  • HTTP 服务器

kubelet 功能归纳一下就是上报 Node 节点信息,和管理(创建、销毁)Pod。 功能看似简单,实际不然。每一个点拿出来都需要很大的篇幅来讲,比如 Node 节点的计算资源,除了传统的 CPU、内存、硬盘,还提供扩展来支持类似 GPU 等资源;Pod 不仅仅有容器,还有相关的网络、安全策略等。

架构 

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

PLEG

Pod Lifecycle Event Generator,字面意思 Pod 生命周期事件(ContainerStartedContainerDiedContainerRemovedContainerChanged)生成器。

其维护着 Pod 缓存;定期通过 ContainerRuntime 获取 Pod 的信息,与缓存中的信息比较,生成如上的事件;将事件写入其维护的通道(channel)中。

PodWorkers

处理事件中 Pod 的同步。核心方法 managePodLoop() 间接调用 kubelet.syncPod() 完成 Pod 的同步:

  • 如果 Pod 正在被创建,记录其延迟

  • 生成 Pod 的 API Status,即 v1.PodStatus:从运行时的 status 转换成 api status

  • 记录 Pod 从 pendingrunning 的耗时

  • StatusManager 中更新 pod 的状态

  • 杀掉不应该运行的 Pod

  • 如果网络插件未就绪,只启动使用了主机网络(host network)的 Pod

  • 如果 static pod 不存在,为其创建镜像(Mirror)Pod

  • 为 Pod 创建文件系统目录:Pod 目录、卷目录、插件目录

  • 使用 VolumeManager 为 Pod 挂载卷

  • 获取 image pull secrets

  • 调用容器运行时(container runtime)的 #SyncPod() 方法

PodManager

存储 Pod 的期望状态,kubelet 服务的不同渠道的 Pod

StatusProvider

提供节点和容器的统计信息,有 cAdvisorCRI 两种实现。

ContainerRuntime

顾名思义,容器运行时。与遵循 CRI 规范的高级容器运行时进行交互。

生命周期事件生成器PLEG

对于 Pod来说,Kubelet 会从多个数据来源(api、file以及http) watch Pod spec 中的变化。对于容器来说,Kubelet 会定期轮询容器运行时,以获取所有容器的最新状态。随着 Pod 和容器数量的增加,轮询会产生较大开销,带来的周期性大量并发请求会导致较高的 CPU 使用率峰值,降低节点性能,从而降低节点的可靠性。为了降低 Pod 的管理开销,提升 Kubelet 的性能和可扩展性,引入了 PLEG(Pod Lifecycle Event Generator)。改进了之前的工作方式:

  1. 减少空闲期间的不必要工作(例如 Pod 的定义和容器的状态没有发生更改)。

  2. 减少获取容器状态的并发请求数量。

为了进一步降低损耗,社区推出了基于event实现的PLEG,当然也需要CRI运行时支持。

它定期检查节点上 Pod 运行情况,如果发现感兴趣的变化,PLEG 就会把这种变化包 装成 Event 发送给 Kubelet 的主同步机制 syncLoop 去处理。

syncLoop

Kubelet启动后通过syncLoop进入到主循环处理Node上Pod Changes事件,监听来自file,apiserver,http三类的事件并汇聚到kubetypes.PodUpdate Channel(Config Channel)中,由syncLoopIteration不断从kubetypes.PodUpdate Channel中消费。

源码分析

源码版本

origin/release-1.30

启动流程

入口函数

入口位于 cmd/kubelet/app/server.go 。

其中 RunKubelet() 接口下的 startKubelet 即是启动 kubelet 的启动函数。

startKubelet
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, kubeDeps.TracerProvider)
    }
    if kubeCfg.ReadOnlyPort > 0 {
        go k.ListenAndServeReadOnly(netutils.ParseIPSloppy(kubeCfg.Address), uint(kubeCfg.ReadOnlyPort))
    }
    go k.ListenAndServePodResources()
}

跳转到 pkg/kubelet/kubelet.go ,这里就是 kubelet 的启动接口代码所在。

kubelet.Run
// Run starts the kubelet reacting to config updates
func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) {
    // ...

    // Start the cloud provider sync manager
    if kl.cloudResourceSyncManager != nil {
        go kl.cloudResourceSyncManager.Run(wait.NeverStop)
    }
    
    // 加载部分模块,如镜像管理, 观察oom, 
    if err := kl.initializeModules(); err != nil {
        kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.KubeletSetupFailed, err.Error())
        klog.ErrorS(err, "Failed to initialize internal modules")
        os.Exit(1)
    }

    // Start volume manager
    // 开启volume 管理
    go kl.volumeManager.Run(kl.sourcesReady, wait.NeverStop)

    //...
    
    // 开启pleg, 即watch方式获取 runtime的event
    // Start the pod lifecycle event generator.
    kl.pleg.Start()

    // Start eventedPLEG only if EventedPLEG feature gate is enabled.
    if utilfeature.DefaultFeatureGate.Enabled(features.EventedPLEG) {
        kl.eventedPleg.Start()
    }

// 启动主进程,接口 kl.syncLoopIteration() 
// 监听来自 file/apiserver/http 事件源发送过来的事件,并对事件做出对应的同步处理。
    kl.syncLoop(ctx, updates, kl)
}

事件watch

syncLoop

// syncLoop is the main loop for processing changes. It watches for changes from
// three channels (file, apiserver, and http) and creates a union of them. For
// any new change seen, will run a sync against desired state and running state. If
// no changes are seen to the configuration, will synchronize the last known desired
// state every sync-frequency seconds. Never returns.
func (kl *Kubelet) syncLoop(ctx context.Context, updates <-chan kubetypes.PodUpdate, handler SyncHandler) {
   //...

    for {
        if err := kl.runtimeState.runtimeErrors(); err != nil {
            klog.ErrorS(err, "Skipping pod synchronization")
            // exponential backoff
            time.Sleep(duration)
            duration = time.Duration(math.Min(float64(max), factor*float64(duration)))
            continue
        }
        // reset backoff if we have a success
        duration = base

        kl.syncLoopMonitor.Store(kl.clock.Now())
        // updatedates为配置聚合通道,handler 为kubelet对象的接口集合
        // plegch 是来自runtime变化事件的通道
        if !kl.syncLoopIteration(ctx, updates, handler, syncTicker.C, \
        housekeepingTicker.C, plegCh) {
            break
        }
        kl.syncLoopMonitor.Store(kl.clock.Now())
    }
}

syncLoop参数中的updates 是来自下面文件的配置聚合通道

// PodConfig is a configuration mux that merges many sources of pod configuration into a single
// consistent structure, and then delivers incremental change notifications to listeners
// in order.
type PodConfig struct {
    pods *podStorage
    mux  *mux

    // the channel of denormalized changes passed to listeners
    // updates通道
    updates chan kubetypes.PodUpdate

    // contains the list of all configured sources
    sourcesLock sync.Mutex
    sources     sets.String
}

handler interface 传入的是实现了下面方法的kubelet对象

syncLoopIteration


func (kl *Kubelet) syncLoopIteration(ctx context.Context, configCh <-chan kubetypes.PodUpdate, handler SyncHandler,
    syncCh <-chan time.Time, housekeepingCh <-chan time.Time, plegCh <-chan *pleg.PodLifecycleEvent) bool {
    select {
    // 获取来自配置通道的事件
    case u, open := <-configCh:
        // Update from a config source; dispatch it to the right handler
        // callback.
        if !open {
            klog.ErrorS(nil, "Update channel is closed, exiting the sync loop")
            return false
        }
        // 下面为对收到pod配置后各个操作的处理
        switch u.Op {
        // 关注ADD操作
        case kubetypes.ADD:
            klog.V(2).InfoS("SyncLoop ADD", "source", u.Source, "pods", klog.KObjSlice(u.Pods))
            // After restarting, kubelet will get all existing pods through
            // ADD as if they are new pods. These pods will then go through the
            // admission process and *may* be rejected. This can be resolved
            // once we have checkpointing.
            handler.HandlePodAdditions(u.Pods)
        case kubetypes.UPDATE:
            klog.V(2).InfoS("SyncLoop UPDATE", "source", u.Source, "pods", klog.KObjSlice(u.Pods))
            handler.HandlePodUpdates(u.Pods)
        case kubetypes.REMOVE:
            klog.V(2).InfoS("SyncLoop REMOVE", "source", u.Source, "pods", klog.KObjSlice(u.Pods))
            handler.HandlePodRemoves(u.Pods)
        case kubetypes.RECONCILE:
            klog.V(4).InfoS("SyncLoop RECONCILE", "source", u.Source, "pods", klog.KObjSlice(u.Pods))
            handler.HandlePodReconcile(u.Pods)
        case kubetypes.DELETE:
            klog.V(2).InfoS("SyncLoop DELETE", "source", u.Source, "pods", klog.KObjSlice(u.Pods))
            // DELETE is treated as a UPDATE because of graceful deletion.
            handler.HandlePodUpdates(u.Pods)
        case kubetypes.SET:
            // TODO: Do we want to support this?
            klog.ErrorS(nil, "Kubelet does not support snapshot update")
        default:
            klog.ErrorS(nil, "Invalid operation type received", "operation", u.Op)
        }

        kl.sourcesReady.AddSource(u.Source)
    
    // 获取来自pleg通道的事件
    case e := <-plegCh:
        if isSyncPodWorthy(e) {
            // PLEG event for a pod; sync it.
            if pod, ok := kl.podManager.GetPodByUID(e.ID); ok {
                klog.V(2).InfoS("SyncLoop (PLEG): event for pod", "pod", klog.KObj(pod), "event", e)
                // 处理 同步自容器运行时的容器最新信息
                handler.HandlePodSyncs([]*v1.Pod{pod})
            } else {
                // If the pod no longer exists, ignore the event.
                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)
            }
        }
        //...
    }
    return true
}

Pod的创建

容器的创建

HandlePodAdditions

重点关注ADD操作

kubetypes.ADD

kubetypes.ADD 既是 Pod 的创建,它调用了 handler.HandlePodAdditions(u.Pods) 接口来响应 Pod 的创建操作,我们先看看 Pod 具体是怎么样被创建的。

  1. HandlePodAdditions() 接口接收的是 pods 切片,意味着可能会出现多个 pod 同时创建。

  2. pods 按创建时间排序,以此对排好序的 pods 进行循环遍历。

  3. 获取 kubelet 所在节点上现有的所有 pods 。

  4. kl.podManager.AddPod(pod) 把当前 pod 信息缓存进 podManager 的 podByUID 这个 map 里面。

  5. 检查 pod 是否 MirrorPod ,如果是 MirrorPod ,则调用 kl.handleMirrorPod(pod, start) 对 Pod 进行处理 。

  6. 无论是 MirrorPod 还是正常 Pod ,最终都是调用 kl.dispatchWork() 接口进行处理,他们的区别在于传进去的参数的不同。

  7. MirrorPod 是 静态 Pod 在 apiserver 上的一个镜像,kubelet 并不需要在节点上给他创建真实的容器,它的 ADD/UPDATE/DELETE 操作类型都被当做 UPDATE 来处理,更新它在 apiserver 上的状态。

// HandlePodAdditions is the callback in SyncHandler for pods being added from
// a config source.
func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {
    start := kl.clock.Now()
    
    // 按照创建事件排序
    sort.Sort(sliceutils.PodsByCreationTime(pods))
    if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) {
        kl.podResizeMutex.Lock()
        defer kl.podResizeMutex.Unlock()
    }
    // 收到的chungking事件通知是一个pods数组,pods可能有多个
    for _, pod := range pods {
        existingPods := kl.podManager.GetPods()
        // Always add the pod to the pod manager. Kubelet relies on the pod
        // manager as the source of truth for the desired state. If a pod does
        // not exist in the pod manager, it means that it has been deleted in
        // the apiserver and no action (other than cleanup) is required.
        //把当前 pod 信息缓存进 podManager 的 podByUID 这个 map 里面
        kl.podManager.AddPod(pod)

        pod, mirrorPod, wasMirror := kl.podManager.GetPodAndMirrorPod(pod)
        if wasMirror {
            if pod == nil {
                klog.V(2).InfoS("Unable to find pod for mirror pod, skipping", "mirrorPod", klog.KObj(mirrorPod), "mirrorPodUID", mirrorPod.UID)
                continue
            }
            kl.podWorkers.UpdatePod(UpdatePodOptions{
                Pod:        pod,
                MirrorPod:  mirrorPod,
                UpdateType: kubetypes.SyncPodUpdate,
                StartTime:  start,
            })
            continue
        }

        //...
        // 转移到podworkers的Update继续处理pod的创建
        kl.podWorkers.UpdatePod(UpdatePodOptions{
            Pod:        pod,
            MirrorPod:  mirrorPod,
            UpdateType: kubetypes.SyncPodCreate,
            StartTime:  start,
        })
    }
}

podWorkers.UpdatePod

// UpdatePod carries a configuration change or termination state to a pod. A pod is either runnable,
// terminating, or terminated, and will transition to terminating if: deleted on the apiserver,
// discovered to have a terminal phase (Succeeded or Failed), or evicted by the kubelet.
func (p *podWorkers) UpdatePod(options UpdatePodOptions) {
   //...
    
    
    // start the pod worker goroutine if it doesn't exist
    podUpdates, exists := p.podUpdates[uid]
    // 当没有该pod时将会执行创建流程, 是以goroutine 方式启动运行
    if !exists {
        // buffer the channel to avoid blocking this method
        podUpdates = make(chan struct{}, 1)
        p.podUpdates[uid] = podUpdates

        // ensure that static pods start in the order they are received by UpdatePod
        if kubetypes.IsStaticPod(pod) {
            p.waitingToStartStaticPodsByFullname[status.fullname] =
                append(p.waitingToStartStaticPodsByFullname[status.fullname], uid)
        }

        // allow testing of delays in the pod update channel
        var outCh <-chan struct{}
        if p.workerChannelFn != nil {
            outCh = p.workerChannelFn(uid, podUpdates)
        } else {
            outCh = podUpdates
        }

        // 在协程中启动pod
        // spawn a pod worker
        go func() {
            // TODO: this should be a wait.Until with backoff to handle panics, and
            // accept a context for shutdown
            defer runtime.HandleCrash()
            defer klog.V(3).InfoS("Pod worker has stopped", "podUID", uid)
                // 启动pod逻辑
            p.podWorkerLoop(uid, outCh)
        }()
    }

    //...
}

podWorkerLoop


func (p *podWorkers) podWorkerLoop(podUID types.UID, podUpdates <-chan struct{}) {
    var lastSyncTime time.Time
    for range podUpdates {
        ctx, update, canStart, canEverStart, ok := p.startPodSync(podUID)
            //...

            // Take the appropriate action (illegal phases are prevented by UpdatePod)
            switch {
            case update.WorkType == TerminatedPod:
                err = p.podSyncer.SyncTerminatedPod(ctx, update.Options.Pod, status)

            case update.WorkType == TerminatingPod:
                var gracePeriod *int64
                if opt := update.Options.KillPodOptions; opt != nil {
                    gracePeriod = opt.PodTerminationGracePeriodSecondsOverride
                }
                podStatusFn := p.acknowledgeTerminating(podUID)

                // if we only have a running pod, terminate it directly
                if update.Options.RunningPod != nil {
                    err = p.podSyncer.SyncTerminatingRuntimePod(ctx, update.Options.RunningPod)
                } else {
                    err = p.podSyncer.SyncTerminatingPod(ctx, update.Options.Pod, status, gracePeriod, podStatusFn)
                }

            default:
                // 继续进入pod的创建
                isTerminal, err = p.podSyncer.SyncPod(ctx, update.Options.UpdateType, update.Options.Pod, update.Options.MirrorPod, status)
            }

            lastSyncTime = p.clock.Now()
            return err
        }()

      
    }
}

Kubelet.SyncPod

// This operation writes all events that are dispatched in order to provide
// the most accurate information possible about an error situation to aid debugging.
// Callers should not write an event if this operation returns an error.
func (kl *Kubelet) SyncPod(ctx context.Context, updateType kubetypes.SyncPodType, pod, mirrorPod *v1.Pod, podStatus *kubecontainer.PodStatus) (isTerminal bool, err error) {
    // ...
    sctx := context.WithoutCancel(ctx)
    // 创建pod
    result := kl.containerRuntime.SyncPod(sctx, pod, podStatus, pullSecrets, kl.backOff)
    kl.reasonCache.Update(pod.UID, result)
    // ...

    return false, nil
}

kubeGenericRuntimeManager.SyncPod

这里就是调用容器运行时 SyncPod() 接口的源码所在,很长,我们先总结一下他的步骤。

  1. 计算 sandbox 容器和普通容器的状态有没有变更。

  2. 如果必要,kill 掉 sandbox 容器重建。

  3. 如果容器不该被运行,则 kill 掉 pod 下的所有容器。

  4. 如果需要,创建 sandbox 容器。

  5. 创建临时容器。

  6. 创建初始化容器 initContainers 。

  7. 创建普通容器。

我们接下来,在根据上面的每一个步骤,来分析这些步骤的具体内容。

注意

sanbox 容器到底是什么东西?kubelet 创建 pod 的时候,你通过 docker ps 去看,会发现它使用 pause 镜像也跟着创建一个同名容器。

57f7fc7cf97b 605c77e624dd "/docker-entrypoint.…" 14 hours ago Up 14 hours k8s_nginx_octopus-deployment-549545bf46-x9cqm_default_c799bb7f-d5d9-41bd-ba60-9a968f0fac54_0 4e0c7fb21e49 registry.aliyuncs.com/google_containers/pause:3.2 "/pause" 14 hours ago Up 14 hours k8s_POD_octopus-deployment-549545bf46-x9cqm_default_c799bb7f-d5d9-41bd-ba60-9a968f0fac54_0 

它是用来为 pod 下的一组容器创建各种 namespace 环境和资源隔离用的, pod 下的各个容器就可以在这个隔离的

环境里面共享各种资源。使得容器无需像 kvm 一样,需要创建一个操作系统实例来对资源进行隔离,这样就可以很好地利用到宿主机的各种资源,这也就是 kubernetes 的精髓所在。

// SyncPod syncs the running pod into the desired pod by executing following steps:
//
//  1. Compute sandbox and container changes.
//  2. Kill pod sandbox if necessary.
//  3. Kill any containers that should not be running.
//  4. Create sandbox if necessary.
//  5. Create ephemeral containers.
//  6. Create init containers.
//  7. Resize running containers (if InPlacePodVerticalScaling==true)
//  8. Create normal containers.
func (m *kubeGenericRuntimeManager) SyncPod(ctx context.Context, pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult) {
    // Step 1: Compute sandbox and container changes.
   
    // Step 2: Kill the pod if the sandbox has changed.

    // Step 3: kill any running containers in this pod which are not to keep.

    // Step 4: Create a sandbox for the pod if necessary.
    podSandboxID := podContainerChanges.SandboxID
    if podContainerChanges.CreateSandbox {
        // ...
        podSandboxID, msg, err = m.createPodSandbox(ctx, pod, podContainerChanges.Attempt)
        
    }

    //...
    // pod 中container 启动方法
    // Helper containing boilerplate common to starting all types of containers.
    // typeName is a description used to describe this type of container in log messages,
    // currently: "container", "init container" or "ephemeral container"
    // metricLabel is the label used to describe this type of container in monitoring metrics.
    // currently: "container", "init_container" or "ephemeral_container"
    start := func(ctx context.Context, typeName, metricLabel string, spec *startSpec) error {
        startContainerResult := kubecontainer.NewSyncResult(kubecontainer.StartContainer, spec.container.Name)
        result.AddSyncResult(startContainerResult)

        isInBackOff, msg, err := m.doBackOff(pod, spec.container, podStatus, backOff)
        if isInBackOff {
            startContainerResult.Fail(err, msg)
            klog.V(4).InfoS("Backing Off restarting container in pod", "containerType", typeName, "container", spec.container, "pod", klog.KObj(pod))
            return err
        }

        metrics.StartedContainersTotal.WithLabelValues(metricLabel).Inc()
        if sc.HasWindowsHostProcessRequest(pod, spec.container) {
            metrics.StartedHostProcessContainersTotal.WithLabelValues(metricLabel).Inc()
        }
        klog.V(4).InfoS("Creating container in pod", "containerType", typeName, "container", spec.container, "pod", klog.KObj(pod))
        // NOTE (aramase) podIPs are populated for single stack and dual stack clusters. Send only podIPs.
        if msg, err := m.startContainer(ctx, podSandboxID, podSandboxConfig, spec, pod, podStatus, pullSecrets, podIP, podIPs); err != nil {
            // startContainer() returns well-defined error codes that have reasonable cardinality for metrics and are
            // useful to cluster administrators to distinguish "server errors" from "user errors".
            metrics.StartedContainersErrorsTotal.WithLabelValues(metricLabel, err.Error()).Inc()
            if sc.HasWindowsHostProcessRequest(pod, spec.container) {
                metrics.StartedHostProcessContainersErrorsTotal.WithLabelValues(metricLabel, err.Error()).Inc()
            }
            startContainerResult.Fail(err, msg)
            // known errors that are logged in other places are logged at higher levels here to avoid
            // repetitive log spam
            switch {
            case err == images.ErrImagePullBackOff:
                klog.V(3).InfoS("Container start failed in pod", "containerType", typeName, "container", spec.container, "pod", klog.KObj(pod), "containerMessage", msg, "err", err)
            default:
                utilruntime.HandleError(fmt.Errorf("%v %+v start failed in pod %v: %v: %s", typeName, spec.container, format.Pod(pod), err, msg))
            }
            return err
        }

        return nil
    }

    // Step 5: start ephemeral containers
    // These are started "prior" to init containers to allow running ephemeral containers even when there
    // are errors starting an init container. In practice init containers will start first since ephemeral
    // containers cannot be specified on pod creation.
    // 创建临时容器, 主要用于需要在运行的 Pod 内执行一些临时任务,
    // 例如调试、日志收集、文件系统浏览等。
    for _, idx := range podContainerChanges.EphemeralContainersToStart {
        start(ctx, "ephemeral container", metrics.EphemeralContainer, ephemeralContainerStartSpec(&pod.Spec.EphemeralContainers[idx]))
    }

    if !utilfeature.DefaultFeatureGate.Enabled(features.SidecarContainers) {
        // Step 6: start the init container.
        if container := podContainerChanges.NextInitContainerToStart; container != nil {
            // Start the next init container.
            if err := start(ctx, "init container", metrics.InitContainer, containerStartSpec(container)); err != nil {
                return
            }

            // Successfully started the container; clear the entry in the failure
            klog.V(4).InfoS("Completed init container for pod", "containerName", container.Name, "pod", klog.KObj(pod))
        }
    } else {
        //  启动init container  
        // 可以帮助确保在主应用容器启动之前完成必要的预处理任务,以确保应用程序的正确运行和可靠性。
        // Step 6: start init containers.
        for _, idx := range podContainerChanges.InitContainersToStart {
            container := &pod.Spec.InitContainers[idx]
            // Start the next init container.
            if err := start(ctx, "init container", metrics.InitContainer, containerStartSpec(container)); err != nil {
                if types.IsRestartableInitContainer(container) {
                    klog.V(4).InfoS("Failed to start the restartable init container for the pod, skipping", "initContainerName", container.Name, "pod", klog.KObj(pod))
                    continue
                }
                klog.V(4).InfoS("Failed to initialize the pod, as the init container failed to start, aborting", "initContainerName", container.Name, "pod", klog.KObj(pod))
                return
            }

            // Successfully started the container; clear the entry in the failure
            klog.V(4).InfoS("Completed init container for pod", "containerName", container.Name, "pod", klog.KObj(pod))
        }
    }

    // Step 7: For containers in podContainerChanges.ContainersToUpdate[CPU,Memory] list, invoke UpdateContainerResources
    if isInPlacePodVerticalScalingAllowed(pod) {
        if len(podContainerChanges.ContainersToUpdate) > 0 || podContainerChanges.UpdatePodResources {
            m.doPodResizeAction(pod, podStatus, podContainerChanges, result)
        }
    }

    // 启动业务container
    // Step 8: start containers in podContainerChanges.ContainersToStart.
    for _, idx := range podContainerChanges.ContainersToStart {
        start(ctx, "container", metrics.Container, containerStartSpec(&pod.Spec.Containers[idx]))
    }

    return
}

Init containers(初始化容器)是 Kubernetes 中的一个概念,它是一种特殊类型的容器,用于在 Pod 中运行在主应用容器启动之前执行的初始化任务。

当创建一个 Pod 时,除了主应用容器之外,你可以定义一个或多个 init containers。这些 init containers 会按照定义的顺序依次执行,并且每个 init container 必须成功完成(即退出状态码为 0)才能继续下一个 init container 的执行,以及主应用容器的启动。

Init containers 可以用于在主应用容器启动之前执行各种初始化任务,例如:

数据库初始化:在主应用容器启动之前,可以使用 init container 创建数据库并执行必要的初始化脚本。

依赖检查:在主应用容器启动之前,可以使用 init container 检查依赖的服务或资源是否可用。

文件下载:在主应用容器启动之前,可以使用 init container 下载配置文件或其他必要的资源。

安全设置:在主应用容器启动之前,可以使用 init container 配置安全相关的设置。

每个 init container 都可以有自己的镜像和配置,它们共享 Pod 的网络命名空间和存储卷。一旦所有的 init containers 成功完成,主应用容器将启动并加入到 Pod 中。

使用 init containers 可以帮助确保在主应用容器启动之前完成必要的预处理任务,以确保应用程序的正确运行和可靠性。

请注意,init containers 是顺序执行的,因此如果某个 init container 失败,则整个 Pod 的启动过程将被阻塞,直到该 init container 成功或手动重启 Pod。

pause容器的注入

containerd.RunPodSandbox

使用的pasue 是在containerd创建sandbox 时。在配置中指定的

// RunPodSandbox creates and starts a pod-level sandbox. Runtimes should ensure
// the sandbox is in ready state.
func (c *criService) RunPodSandbox(ctx context.Context, r *runtime.RunPodSandboxRequest) (_ *runtime.RunPodSandboxResponse, retErr error) {
    ...

    // Ensure sandbox container image snapshot.
    // 使用的sandbox镜像,一般就是pause
    image, err := c.ensureImageExists(ctx, c.config.SandboxImage, config)
    if err != nil {
        return nil, fmt.Errorf("failed to get sandbox image %q: %w", c.config.SandboxImage, err)
    }
    containerdImage, err := c.toContainerdImage(ctx, *image)
    if err != nil {
        return nil, fmt.Errorf("failed to get image from containerd %q: %w", image.ID, err)
    }

    ociRuntime, err := c.getSandboxRuntime(config, r.GetRuntimeHandler())
    if err != nil {
        return nil, fmt.Errorf("failed to get sandbox runtime: %w", err)
    }
    log.G(ctx).WithField("podsandboxid", id).Debugf("use OCI runtime %+v", ociRuntime)

    runtimeStart := time.Now()
    // Create sandbox container.
    // NOTE: sandboxContainerSpec SHOULD NOT have side
    // effect, e.g. accessing/creating files, so that we can test
    // it safely.
    // NOTE: the network namespace path will be created later and update through updateNetNamespacePath function
    spec, err := c.sandboxContainerSpec(id, config, &image.ImageSpec.Config, "", ociRuntime.PodAnnotations)
    if err != nil {
        return nil, fmt.Errorf("failed to generate sandbox container spec: %w", err)
    }
    log.G(ctx).WithField("podsandboxid", id).Debugf("sandbox container spec: %#+v", spew.NewFormatter(spec))
    sandbox.ProcessLabel = spec.Process.SelinuxLabel
    defer func() {
        if retErr != nil {
            selinux.ReleaseLabel(sandbox.ProcessLabel)
        }
    }()

    // ...

    container, err := c.client.NewContainer(ctx, id, opts...)
    if err != nil {
        return nil, fmt.Errorf("failed to create containerd container: %w", err)
    }

    // Add container into sandbox store in INIT state.
    sandbox.Container = container

    defer func() {
        // Put the sandbox into sandbox store when the some resource fails to be cleaned.
        if retErr != nil && cleanupErr != nil {
            log.G(ctx).WithError(cleanupErr).Errorf("encountered an error cleaning up failed sandbox %q, marking sandbox state as SANDBOX_UNKNOWN", id)
            if err := c.sandboxStore.Add(sandbox); err != nil {
                log.G(ctx).WithError(err).Errorf("failed to add sandbox %+v into store", sandbox)
            }
        }
    }()

   ...

Pause 容器

Pause容器的功能可参考 pause.c文件

pause功能只是截获 SIGINT, SIGTERM, SIGCHLD 信号,其中SIGCHLD只是收集退出的子进程,避免成为孤儿进程。

puase容器主要是为了:

  1. 保持有一个容器在sandbox中运行,维持sandbox环境。

  2. 其他容器停止时需要在同空间下有主进程进行回收,避免成为孤儿进程。

/*
Copyright 2016 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>

#define STRINGIFY(x) #x
#define VERSION_STRING(x) STRINGIFY(x)

#ifndef VERSION
#define VERSION HEAD
#endif

static void sigdown(int signo) {
  psignal(signo, "Shutting down, got signal");
  exit(0);
}

static void sigreap(int signo) {
  while (waitpid(-1, NULL, WNOHANG) > 0)
    ;
}

int main(int argc, char **argv) {
  int i;
  for (i = 1; i < argc; ++i) {
    if (!strcasecmp(argv[i], "-v")) {
      printf("pause.c %s\n", VERSION_STRING(VERSION));
      return 0;
    }
  }

  if (getpid() != 1)
    /* Not an error because pause sees use outside of infra containers. */
    fprintf(stderr, "Warning: pause should be the first process\n");
    // 收集退出的子进程
  if (sigaction(SIGINT, &(struct sigaction){.sa_handler = sigdown}, NULL) < 0)
    return 1;
  if (sigaction(SIGTERM, &(struct sigaction){.sa_handler = sigdown}, NULL) < 0)
    return 2;
  if (sigaction(SIGCHLD, &(struct sigaction){.sa_handler = sigreap,
                                             .sa_flags = SA_NOCLDSTOP},
                NULL) < 0)
    return 3;

  for (;;)
    pause();
  fprintf(stderr, "Error: infinite loop terminated\n");
  return 42;
}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/2300861.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

Golang学习笔记_33——桥接模式

Golang学习笔记_30——建造者模式 Golang学习笔记_31——原型模式 Golang学习笔记_32——适配器模式 文章目录 桥接模式详解一、桥接模式核心概念1. 定义2. 解决的问题3. 核心角色4. 类图 二、桥接模式的特点三、适用场景1. 多维度变化2. 跨平台开发3. 动态切换实现 四、与其他…

【js逆向_入门】图灵爬虫练习平台 第四题

(base64解码&#xff09;地址&#xff1a;aHR0cHM6Ly9zdHUudHVsaW5ncHl0b24uY24vcHJvYmxlbS1kZXRhaWwvNC8 请求接口带有加密参数&#xff1a; 全局搜索Sign,找到参数生成位置 一目了然&#xff0c;知道参数是怎么构造生成的 调试代码 测试验证思路是否正确 时间&#xff1a; …

Mybatis后端数据库查询多对多查询解决方案

问题场景&#xff1a; 我开发的是一个论文选择系统。 后端用一个论文表paper来存储论文信息。 论文信息中&#xff0c;包含前置课程&#xff0c;也就是你需要修过这些课程才能选择这个论文。 而一个论文对应的课程有很多个。 这样就造成了一个数据库存储的问题。一个paper…

【MySQL排错 】mysql: command not found 数据库安装后无法加载的解决办法

【MySQL排错 】mysql: command not found 数据库安装后无法加载的解决办法 A Solution to Solve Error - mysql: command not found After The Installation of MySQL Community Server By JacksonML 本文简要介绍如何在macOS安装完毕MySQL数据库服务器后&#xff0c;针对无…

分享一款AI绘画图片展示和分享的小程序

&#x1f3a8;奇绘图册 【开源】一款帮AI绘画爱好者维护绘图作品的小程序 查看Demo 反馈 github 文章目录 前言一、奇绘图册是什么&#xff1f;二、项目全景三、预览体验3.1 截图示例3.2 在线体验 四、功能介绍4.1 小程序4.2 服务端 五、安装部署5.1 快速开始~~5.2 手动部…

大模型知识蒸馏技术(4)——离线蒸馏

版权声明 本文原创作者:谷哥的小弟作者博客地址:http://blog.csdn.net/lfdfhl离线蒸馏概述 离线蒸馏是知识蒸馏中最早被提出且最为常见的实现方式,其核心在于教师模型和学生模型的训练是分阶段进行的。具体而言,教师模型首先在训练集上进行充分训练,直至收敛,然后利用教…

解决DeepSeek服务器繁忙的有效方法

全球42%的企业遭遇过AI工具服务器过载导致内容生产中断&#xff08;数据来源&#xff1a;Gartner 2025&#xff09;。当竞品在凌晨3点自动发布「智能家居安装指南」时&#xff0c;你的团队可能正因DeepSeek服务器繁忙错失「净水器保养教程」的流量黄金期⏳。147SEO智能调度系统…

BT401双模音频蓝牙模块如何开启ble的透传,有什么注意事项

BT401音频蓝牙模块如何开启ble的透传&#xff1f; 首先BT401的蓝牙音频模块&#xff0c;分为两个版本&#xff0c;dac版本和iis数字音频版本 DAC版本&#xff1a;就是BT401蓝牙模块【9和10脚】直接输出模拟音频信号&#xff0c;也就是说&#xff0c;直接推动耳机可以听到声音 …

基于SSM框架的宠物之家系统(有源码+论文!!!)

这个系统可以帮助大家去做设计或者学习,大家可以管我要word版论文🥰这里具体论文内照片、e-r图等等加载不进来, 大家如果想要源码+论文+制定+调试,可以私信我!!(可改别的系统,例如调查问卷系统等等) 目录 第1章 绪论 1.1开发背景 1.2开发工具及语言 第2章 宠物之家系…

网工项目理论1.7 设备选型

本专栏持续更新&#xff0c;整一个专栏为一个大型复杂网络工程项目。阅读本文章之前务必先看《本专栏必读》。 一.交换机选型要点 制式:盒式交换机/框式交换机。功能:二层交换机/三层交换机。端口密度:每交换机可以提供的端口数量。端口速率:百兆/千兆/万兆。交换容量:交换矩阵…

Gateway中的Filter机制

Gateway中的Filter机制 文章目录 Gateway中的Filter机制Gateway中的Filter机制Gateway Filter 机制的概述核心思想与设计Filter 的两种类型过滤器的生命周期Gateway Filter 的特点Gateway Filter 的意义Gateway Filter 的工作原理核心架构与执行流程执行流程解析过滤器类型的角…

顺序表常用操作和笔试题

1、顺序表的常用操作 1.1 顺序表的创建 如下代码所示&#xff1a;创建了一个默认空间为10的整型顺序表&#xff0c;如果空间不足则会以1.5倍扩容。 List<Integer> list new ArrayList<>(); 创建一个空间为15的整型顺序表 List<Integer> list2 new ArrayL…

二.数据治理流程架构

1、数据治理流程架构核心思想&#xff1a; 该图描绘了一个以数据标准规范体系为核心&#xff0c;大数据生命周期管理为主线&#xff0c;数据资源中心为依托&#xff0c;并辅以数据质量管理和大数据安全与隐私管理的数据治理流程架构。它旨在通过规范化的流程和技术手段&#x…

解锁机器学习核心算法 | 线性回归:机器学习的基石

在机器学习的众多算法中&#xff0c;线性回归宛如一块基石&#xff0c;看似质朴无华&#xff0c;却稳稳支撑起诸多复杂模型的架构。它是我们初涉机器学习领域时便会邂逅的算法之一&#xff0c;其原理与应用广泛渗透于各个领域。无论是预测房价走势、剖析股票市场波动&#xff0…

CSS 底部颗粒磨砂特效

参考&#xff1a;element-plus的顶部效果 background-image: radial-gradient(transparent 1px, var(--bg-color) 1px); background-size: 4px 4px; backdrop-filter: saturate(50%) blur(4px); 注意点&#xff1a;var(--bg-color) 改为跟你背景色相同的即可。 其他自己看情况…

QT自定义扫描控件,支持任意方位拖动和拖动扫描范围。

部分代码&#xff1a; void FishControlForm::paintEvent(QPaintEvent *event) {QPainter p(this);p.setRenderHints(QPainter::Antialiasing|QPainter::SmoothPixmapTransform);p.translate(m_centerPoint);//bgp.setPen(Qt::white);p.drawEllipse(-m_radius,-m_radius,m_rad…

Ubuntu18.04安装rvm、ruby2.6.5和rails5.2.6

系统环境&#xff1a;Ubuntu 18.04 一、安装前准备 1. sudo apt update 2. sudo apt upgrade 如果提示abort&#xff0c;忽略。 3. sudo apt install sqlite3 gnupg curl git libpq-dev 二、安装rvm ruby版本管理器 1.切换管理员模式 sudo su 2.安装软件签名公钥 gpg…

燧光 XimmerseMR SDK接入Unity

官网SDK文档连接&#xff1a; RhinoX Unity XR SDK 一&#xff1a;下载SDK 下载链接&#xff1a;RhinoX Unity XR SDK 二&#xff1a;打开Unity项目&#xff0c;添加Package 1、先添加XR Core Utilties包和XR Interaction Toolkit包 2、导 2、再导入下载好的燧光SDK 三&…

罗德与施瓦茨ZNB20,矢量网络分析仪9KHz-20GHz

罗德与施瓦茨ZNB20矢量网络分析仪9KHz-20GHz R&SZNB20矢量网络分析仪 产品型号: ZNB20 产品品牌&#xff1a;罗德与施瓦茨 R&S 产品名称: 矢量网络分析仪 频率范围&#xff1a;9kHz - 20GHz R&S ZNB 矢量网络分析仪 良好的测量速度、动态范围和操作方便性&am…

突破平面限制:低空经济激活城市土地“立体生长力”

蓄锐行&#xff1a;我深耕于低空经济领域&#xff0c;同时对IOT技术有着深入的理解&#xff0c;而在AI的探索性研究上&#xff0c;我也已迈出坚实的步伐&#xff0c;虽起步不久&#xff0c;却满怀热情与憧憬。平日里&#xff0c;我热衷于撰写关于低空经济的文章&#xff0c;分享…