Kubernetes的Device Plugin机制源码解析(1)

本文涉及的产品
容器镜像服务 ACR,镜像仓库100个 不限时长
容器服务 Serverless 版 ACK Serverless,952元额度 多规格
容器服务 Serverless 版 ACK Serverless,317元额度 多规格
简介: Kubernetes 1.8 引入的Device Plugin机制,通过扩展的方式实现支持GPU、FPGA、高性能 NIC、InfiniBand等各种设备的集成。而Device Manager正是Kubelet内负责Device Plugin交互和设备生命周期管理的模块,在了解其基本设计后,本文对Device Manager的源码分析,理解其运作方式。

Kubernetes 1.8 引入的Device Plugin机制,通过扩展的方式实现支持GPU、FPGA、高性能 NIC、InfiniBand等各种设备的集成。而Device Manager正是Kubelet内负责Device Plugin交互和设备生命周期管理的模块,在了解其基本设计后,我们需要通过对Device Manager的源码分析,理解其运作方式。

基本原则

首先明确目标:

并不是搞懂Kubelet的所有实现,而是希望理解Device Manager如何在资源发现,Pod创建,设备健康检查过程中所做的工作以及其如何与Kubelet交互,所以我们会忽略掉与Device Manager无关的操作。

这里是我阅读代码的原则和一些体会:

  • 理解接口,搞清楚和外部模块的交互
  • 理解实现接口的结构体
  • 从用户场景的角度将方法调用和数据结构关联起来,好比将剧情和人物串联起来,了解了人物设定后,就可以更快速切入代码的调用过程;而代码调用的阅读也可以加深对数据结构设计的理解
  • Kubernetes的代码比较复杂,很难一下就搞清楚每一个数据结构定义的目的和用途,这时我们可以把问题和假设记下来,不要过分纠结,可以在后面求证。书读百变其意自现,代码也是一样,当你逐渐熟悉了代码的脉络的时候,有些问题也会迎刃而解
  • 由于Device Manager工作在Kubelet中,对于Kubelet的源码通篇的了解是理解具体模块运作机制的基础

P.S. 本文解析的Kubernetes源码版本是1.9.3

DeviceManager的核心代码都在 pkg/kubelet/cm/deviceplugin

devicePlugin.png

DeviceManager接口定义

所在的文件

pkg/kubelet/cm/deviceplugin/types.go

具体定义:

// Manager manages all the Device Plugins running on a node.
type Manager interface {
    // Start starts device plugin registration service.
    Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady) error

    // Devices is the map of devices that have registered themselves
    // against the manager.
    // The map key is the ResourceName of the device plugins.
    Devices() map[string][]pluginapi.Device

    // Allocate configures and assigns devices to pods. The pods are provided
    // through the pod admission attributes in the attrs argument. From the
    // requested device resources, Allocate will communicate with the owning
    // device plugin to allow setup procedures to take place, and for the
    // device plugin to provide runtime settings to use the device (environment
    // variables, mount points and device files). The node object is provided
    // for the device manager to update the node capacity to reflect the
    // currently available devices.
    Allocate(node *schedulercache.NodeInfo, attrs *lifecycle.PodAdmitAttributes) error

    // Stop stops the manager.
    Stop() error

    // GetDeviceRunContainerOptions checks whether we have cached containerDevices
    // for the passed-in <pod, container> and returns its DeviceRunContainerOptions
    // for the found one. An empty struct is returned in case no cached state is found.
    GetDeviceRunContainerOptions(pod *v1.Pod, container *v1.Container) *DeviceRunContainerOptions

    // GetCapacity returns the amount of available device plugin resource capacity
    // and inactive device plugin resources previously registered on the node.
    GetCapacity() (v1.ResourceList, []string)
}

从注释中可以看到DeviceManager负责管理节点上运行的所有设备插件,这里分别定义了可以和外界交互的6个方法:

  • Start()Stop()分别是启动设备插件注册和停止服务,这其实K8S中的常见套路
  • Devices()以map的形式列出device列表

以下3个方法是比较核心的工作:

  • Allocate()为Pod分配可用的设备,并且调用设备插件进行所需的设备初始化
  • GetDeviceRunContainerOptions()获得为容器配置设备所需要的参数,比如Environment,Volume和Device,这个方法会用于创建容器的过程中
  • GetCapacity()用于节点向API Server上报Extended Resource的数量

当然要更清楚的理解,还需要结合具体场景中的调用链路进行理解。这里DeviceManager接口有两个实现分别是:MangerImplManagerStub, ManagerStub实际上是一个空实现,无需细看。下面简单了解一下 MangerImpl的实现

DeviceManager接口实现

所在的文件

pkg/kubelet/cm/deviceplugin/manager.go

具体定义:

// ManagerImpl is the structure in charge of managing Device Plugins.
type ManagerImpl struct {
    socketname string
    socketdir  string

    endpoints map[string]endpoint // Key is ResourceName
    mutex     sync.Mutex

    server *grpc.Server

    // activePods is a method for listing active pods on the node
    // so the amount of pluginResources requested by existing pods
    // could be counted when updating allocated devices
    activePods ActivePodsFunc

    // sourcesReady provides the readiness of kubelet configuration sources such as apiserver update readiness.
    // We use it to determine when we can purge inactive pods from checkpointed state.
    sourcesReady config.SourcesReady

    // callback is used for updating devices' states in one time call.
    // e.g. a new device is advertised, two old devices are deleted and a running device fails.
    callback monitorCallback

    // allDevices contains all of registered resourceNames and their exported device IDs.
    allDevices map[string]sets.String

    // allocatedDevices contains allocated deviceIds, keyed by resourceName.
    allocatedDevices map[string]sets.String

    // podDevices contains pod to allocated device mapping.
    podDevices podDevices
}

在ManagerImpl的定义和注释中,可以大致猜测它在做三件事:

  • 提供grpc的服务,支持多个Device Plugin的注册
  • 为Device Plugin提供回调函数monitorCallback,当设备的状态发生变化时,可以让Device Manager被通知,从而做一些相应的处理。比如当某个设备无法正常工作时,就需要将节点上可用资源总数减去一个
  • 设备的分配和管理,具体讲就是记录某种设备一共有哪几个,已经分配出去的是哪几个。从这里看,Device Plugin需要为每个设备提供一个UUID, 这个UUID需要在本节点唯一并且不可改变,而Device Manager要做的事情就是维护这个UUID的集合,并且负责设备更新和分配

场景分类

这里主要涉及五个场景:

  • Device Manager的初始化和启动
  • 接收Device Plugin的endpoint注册,并且向Endpoint查询Device ID列表
  • 定时上报节点上的设备信息
  • 创建Pod时,将设备信息与Pod结合,生成创建容器所需要的配置(Environment, Device, Volume)
  • 当设备状态不健康的时候,通知Kubelet更新可用设备的状态

本文首先分析场景一:Device Manager的初始化和启动过程

Device Manager的初始化和启动过程

Kubernetes的代码量巨大,但是细看每个模块的启动流程都有比较相似的套路,以Kubelet为例:

  1. 创建一个 KubeletServer 配置对象,这个对象保存着 kubelet 运行需要的所有配置信息
  2. 解析命令行,根据命令行的参数更新 KubeletServer
  3. 根据 KubeletServer 的配置创建真正的 kubelet 运行时对象
  4. 通过Start()方法启动该 kubelet 运行时对象

而DeviceManger的初始化就是发生在步骤3和步骤4

deviceManagerInit.png

  • app.kubelet对应的是cmd/kubelet/kubelet.go
  • server对应的是cmd/kubelet/app/server.go
  • kubelet对应的是pkg/kubelet/kubelet.go
  • container_manager_linux对应的是pkg/kubelet/cm/container_manager_linux.go
  • device.manager对应的是pkg/kubelet/cm/deviceplugin/manager.go

以上时序图就是Kubelet如何初始化和启动DeviceManager的流程(为了方便理解,这里会忽略和DeviceManager无关的方法)

可以看到serverrun()方法做两件事情:NewMainKubeletstartKubelet,而Device Manager的初始化与启动也是在这两个步骤中完成,同时启动grpc注册服务,这时Device Plugin就可以注册进来。

  1. DeviceManger的初始化是在创建ContainerManager对象时完成的,而ContainerManager对象作为NewMainKubelet创建Kubelet运行时对象的参数,

实际定义在:pkg/kubelet/cm/container_manager_linux.go

func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.Interface, nodeConfig NodeConfig, failSwapOn bool, devicePluginEnabled bool, recorder record.EventRecorder) (ContainerManager, error) {
...

glog.Infof("Creating device plugin manager: %t", devicePluginEnabled)
    if devicePluginEnabled {
        cm.devicePluginManager, err = deviceplugin.NewManagerImpl()
    } else {
        cm.devicePluginManager, err = deviceplugin.NewManagerStub()
    }

...
}

由于这个功能目前还比较新,需要通过feature gate打开, 即配置 --feature-gates=DevicePlugins=true,默认该功能是关闭的。当该功能打开时会调用deviceplugin.NewManagerImpl(),否则会有stub实现,不作任何事情。

deviceplugin.NewManagerImpl()定义在pkg/kubelet/cm/deviceplugin/manager.go内,

// NewManagerImpl creates a new manager.
func NewManagerImpl() (*ManagerImpl, error) {
    return newManagerImpl(pluginapi.KubeletSocket)
}

实际上真正做初始的工作都是在下列方法完成的

func newManagerImpl(socketPath string) (*ManagerImpl, error) {
    glog.V(2).Infof("Creating Device Plugin manager at %s", socketPath)

    if socketPath == "" || !filepath.IsAbs(socketPath) {
        return nil, fmt.Errorf(errBadSocket+" %v", socketPath)
    }

    dir, file := filepath.Split(socketPath)
    manager := &ManagerImpl{
        endpoints:        make(map[string]endpoint),
        socketname:       file,
        socketdir:        dir,
        allDevices:       make(map[string]sets.String),
        allocatedDevices: make(map[string]sets.String),
        podDevices:       make(podDevices),
    }
    manager.callback = manager.genericDeviceUpdateCallback

    // The following structs are populated with real implementations in manager.Start()
    // Before that, initializes them to perform no-op operations.
    manager.activePods = func() []*v1.Pod { return []*v1.Pod{} }
    manager.sourcesReady = &sourcesReadyStub{}

    return manager, nil
}

这里只是做ManagerImpl的初始化,有意义的工作只有两个

  • 设置DeviceManager内置grpc服务的监听文件 socketPath, 由于DeviceManager和Device Plugin部署在同一个节点,所以只需要利用Unix Socket的模式通信
  • 设置设备状态的回调函数 genericDeviceUpdateCallback

就像注释中提到 The following structs are populated with real implementations in manager.Start()的一样,实际上在初始化阶段,并没有

  1. DeviceMangerStart()是在启动Kubelet运行时initializeModules调用的,具体还是ContainerManager`启动的一部分。
func (cm *containerManagerImpl) Start(node *v1.Node,
    activePods ActivePodsFunc,
    sourcesReady config.SourcesReady,
    podStatusProvider status.PodStatusProvider,
    runtimeService internalapi.RuntimeService) error {

...

// Starts device plugin manager.
    if err := cm.devicePluginManager.Start(deviceplugin.ActivePodsFunc(activePods), sourcesReady); err != nil {
        return err
    }
    return nil

}

这里会把活跃的pod列表以及pod元数据的来源(FILE, URL, api-server)作为输入用来启动DeviceManager, 这两个参数在启动的时候并没有用到

func (m *ManagerImpl) Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady) error {
    glog.V(2).Infof("Starting Device Plugin manager")

    m.activePods = activePods
    m.sourcesReady = sourcesReady

    // Loads in allocatedDevices information from disk.
    err := m.readCheckpoint()
    if err != nil {
        glog.Warningf("Continue after failing to read checkpoint file. Device allocation info may NOT be up-to-date. Err: %v", err)
    }

    socketPath := filepath.Join(m.socketdir, m.socketname)
    os.MkdirAll(m.socketdir, 0755)

    // Removes all stale sockets in m.socketdir. Device plugins can monitor
    // this and use it as a signal to re-register with the new Kubelet.
    if err := m.removeContents(m.socketdir); err != nil {
        glog.Errorf("Fail to clean up stale contents under %s: %+v", m.socketdir, err)
    }

    s, err := net.Listen("unix", socketPath)
    if err != nil {
        glog.Errorf(errListenSocket+" %+v", err)
        return err
    }

    m.server = grpc.NewServer([]grpc.ServerOption{}...)

    pluginapi.RegisterRegistrationServer(m.server, m)
    go m.server.Serve(s)

    glog.V(2).Infof("Serving device plugin registration server on %q", socketPath)

    return nil
}

Start主要核心做两件事情:

  • m.readCheckpoint() 负责从本地checkpoint(/var/lib/kubelet/device-plugins/kubelet_internal_checkpoint)中获取已经注册和分配了的设备信息,为什么要这样做呢?这主要是因为Kubelet负责设备的分配和管理工作, 这些信息只存在于Kubelet的内存中。一旦Kubelet重启了之后,哪些设备已经分配了出去,以及这些分配出去的设备具体和哪个Pod关联

DeviceManager在每次分配设备给Pod后会将Pod和设备的映射关系以json格式记录到本地的一个文件

  • go m.server.Serve(s) 以后台grouting的方式启动grpc服务,这样就可以完成Device Plugin的注册,我们会在后面介绍grpc开放的服务如何与Device Plugin进行交互。

小结:

阅读开源源代码可以帮助我们提升技术水平, 不但能深入技术底层原理,快速理解技术架构;同样也可以帮助我们学习优秀的代码风格和设计模式。本文这里只是抛砖引玉,对Device Manager初始化场景进行了分析,后续我们也会对其他场景继续研究,加深对Kubernetes的Device Plugin机制的理解。

相关实践学习
通过Ingress进行灰度发布
本场景您将运行一个简单的应用,部署一个新的应用用于新的发布,并通过Ingress能力实现灰度发布。
容器应用与集群管理
欢迎来到《容器应用与集群管理》课程,本课程是“云原生容器Clouder认证“系列中的第二阶段。课程将向您介绍与容器集群相关的概念和技术,这些概念和技术可以帮助您了解阿里云容器服务ACK/ACK Serverless的使用。同时,本课程也会向您介绍可以采取的工具、方法和可操作步骤,以帮助您了解如何基于容器服务ACK Serverless构建和管理企业级应用。 学习完本课程后,您将能够: 掌握容器集群、容器编排的基本概念 掌握Kubernetes的基础概念及核心思想 掌握阿里云容器服务ACK/ACK Serverless概念及使用方法 基于容器服务ACK Serverless搭建和管理企业级网站应用
目录
相关文章
|
5天前
|
监控 Java 应用服务中间件
高级java面试---spring.factories文件的解析源码API机制
【11月更文挑战第20天】Spring Boot是一个用于快速构建基于Spring框架的应用程序的开源框架。它通过自动配置、起步依赖和内嵌服务器等特性,极大地简化了Spring应用的开发和部署过程。本文将深入探讨Spring Boot的背景历史、业务场景、功能点以及底层原理,并通过Java代码手写模拟Spring Boot的启动过程,特别是spring.factories文件的解析源码API机制。
16 2
|
1月前
|
存储 缓存 算法
分布式锁服务深度解析:以Apache Flink的Checkpointing机制为例
【10月更文挑战第7天】在分布式系统中,多个进程或节点可能需要同时访问和操作共享资源。为了确保数据的一致性和系统的稳定性,我们需要一种机制来协调这些进程或节点的访问,避免并发冲突和竞态条件。分布式锁服务正是为此而生的一种解决方案。它通过在网络环境中实现锁机制,确保同一时间只有一个进程或节点能够访问和操作共享资源。
68 3
|
5天前
|
存储 安全 Linux
Golang的GMP调度模型与源码解析
【11月更文挑战第11天】GMP 调度模型是 Go 语言运行时系统的核心部分,用于高效管理和调度大量协程(goroutine)。它通过少量的操作系统线程(M)和逻辑处理器(P)来调度大量的轻量级协程(G),从而实现高性能的并发处理。GMP 模型通过本地队列和全局队列来减少锁竞争,提高调度效率。在 Go 源码中,`runtime.h` 文件定义了关键数据结构,`schedule()` 和 `findrunnable()` 函数实现了核心调度逻辑。通过深入研究 GMP 模型,可以更好地理解 Go 语言的并发机制。
|
7天前
|
存储 消息中间件 算法
深入探索操作系统的心脏——内核机制解析
本文旨在揭示操作系统核心——内核的工作原理,通过剖析其关键组件与机制,为读者提供一个清晰的内核结构图景。不同于常规摘要的概述性内容,本文摘要将直接聚焦于内核的核心概念、主要功能以及其在系统管理中扮演的角色,旨在激发读者对操作系统深层次运作原理的兴趣与理解。
|
11天前
|
运维 Kubernetes Cloud Native
Kubernetes云原生架构深度解析与实践指南####
本文深入探讨了Kubernetes作为领先的云原生应用编排平台,其设计理念、核心组件及高级特性。通过剖析Kubernetes的工作原理,结合具体案例分析,为读者呈现如何在实际项目中高效部署、管理和扩展容器化应用的策略与技巧。文章还涵盖了服务发现、负载均衡、配置管理、自动化伸缩等关键议题,旨在帮助开发者和运维人员掌握利用Kubernetes构建健壮、可伸缩的云原生生态系统的能力。 ####
|
18天前
|
消息中间件 缓存 安全
Future与FutureTask源码解析,接口阻塞问题及解决方案
【11月更文挑战第5天】在Java开发中,多线程编程是提高系统并发性能和资源利用率的重要手段。然而,多线程编程也带来了诸如线程安全、死锁、接口阻塞等一系列复杂问题。本文将深度剖析多线程优化技巧、Future与FutureTask的源码、接口阻塞问题及解决方案,并通过具体业务场景和Java代码示例进行实战演示。
38 3
|
19天前
|
存储 缓存 安全
🌟Java零基础:深入解析Java序列化机制
【10月更文挑战第20天】本文收录于「滚雪球学Java」专栏,专业攻坚指数级提升,希望能够助你一臂之力,帮你早日登顶实现财富自由🚀;同时,欢迎大家关注&&收藏&&订阅!持续更新中,up!up!up!!
22 3
|
24天前
|
Java 开发者 UED
Java编程中的异常处理机制解析
在Java的世界里,异常处理是确保程序稳定性和可靠性的关键。本文将深入探讨Java的异常处理机制,包括异常的类型、如何捕获和处理异常以及自定义异常的创建和使用。通过理解这些概念,开发者可以编写更加健壮和易于维护的代码。
|
9天前
|
存储 Kubernetes 调度
深度解析Kubernetes中的Pod生命周期管理
深度解析Kubernetes中的Pod生命周期管理
|
1月前
|
存储
让星星⭐月亮告诉你,HashMap的put方法源码解析及其中两种会触发扩容的场景(足够详尽,有问题欢迎指正~)
`HashMap`的`put`方法通过调用`putVal`实现,主要涉及两个场景下的扩容操作:1. 初始化时,链表数组的初始容量设为16,阈值设为12;2. 当存储的元素个数超过阈值时,链表数组的容量和阈值均翻倍。`putVal`方法处理键值对的插入,包括链表和红黑树的转换,确保高效的数据存取。
53 5

相关产品

  • 容器服务Kubernetes版
  • 推荐镜像

    更多