开发者社区> 必嘫> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

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

简介: 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机制的理解。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
【kubernetes】二进制文件方式安装 Kubernetes 集群(四)
二进制文件方式安装 Kubernetes 集群(四)
40 0
【kubernetes】二进制文件方式安装 Kubernetes 集群(一)
【kubernetes】二进制文件方式安装 Kubernetes 集群(一)
75 0
【DockerCon2017最新技术解读】如何在阿里云一键部署高可用的Kubernetes集群
在云栖TechDay34期:DockerCon2017最新的技术解读中,阿里云的容器服务团队技术专家谢瑶瑶做了题为《Kubernetes on Alibaba Cloud》的分享,主要介绍了Kubernetes的基本概念,如何在阿里云快速一键部署高可用安全的Kubernetes集群以及阿里云的Cloud Provider所提供的能力。
8531 0
+关注
必嘫
阿里云技术专家,在应用性能监控和软件交付方面有丰富的实践经验,目前专注于容器服务,微服务以及机器学习等领域。
43
文章
0
问答
来源圈子
更多
容器服务 Kubernetes 版(简称 ACK)提供高性能可伸缩的容器应用管理能力,支持企业级 Kubernetes 容器化应用的全生命周期管理。容器服务 Kubernetes 版简化集群的搭建和扩容等工作,整合阿里云虚拟化、存储、网络和安全能力,打造云端最佳的 Kubernetes 容器化应用运行环境。
+ 订阅
文章排行榜
最热
最新
相关电子书
更多
低代码开发师(初级)实战教程
立即下载
阿里巴巴DevOps 最佳实践手册
立即下载
冬季实战营第三期:MySQL数据库进阶实战
立即下载