【Kubernetes】Device Plugin

Kubernetes 原生支持对于CPU和内存资源的发现，但是有很多其他的设备 kubelet不能原生处理，比如GPU、FPGA、RDMA、存储设备和其他类似的异构计算资源设备。为了能够使用这些设备资源，我们需要进行各个设备的初始化和设置。按照 Kubernetes 的 OutOfTree 的哲学理念，我们不应该把各个厂商的设备初始化设置相关代码与 Kubernetes 核心代码放在一起。与之相反，我们需要一种机制能够让各个设备厂商向 Kubelet 上报设备资源，而不需要修改 Kubernetes 核心代码。这即是 Device Plugin 这一机制的来源，本文将介绍 Device Plugin 的实现原理，并介绍其使用。

Device 插件原理

Device Plugin 实际上是一个 gPRC server，Device 插件一般推荐使用 DaemonSet 的方式部署，并将 /var/lib/kubelet/device-plugins 以 Volume 的形式挂载到容器中。当然，也可以手动运行的方式来部署，但这样就没有失败自动恢复的功能了。

为了能够使用某个厂商的特定设备，一般有两步：

• kubectl create -f http://vendor.com/device-plugin-daemonset.yaml
• 执行 kubectl describe nodes的时候，相关设备会出现在node status中：vendor-domain/vendor-device

当 Device Plugin 向 kubelet 注册后，kubelet 就通过 RPC 与 Device Plugin 交互：

• ListAndWatch() ：让 kubelet 发现设备资源和对应属性，并且在设备资源发生变动的时候接收通知
• Allocate() ：kubelet 在创建容器前通过 Allocate来申请相关设备资源

Registration

为了向 kubelet 告知 Device Plugin 的存在，Device Plugin 必须向 kubelet 发出注册请求，这之后 kubelet 才会和 Device Plugin 通过 gRPC交互，具体过程如下：

• Device Plugin 向 Kubelet 发送一个 RegisterRequest的请求
• Kubelet 收到 RegisterRequest 请求后，返回一个 RegisterResponse，如果Kubelet碰到任何错误，会把错误附在Response中
• 如果 Device Plugin 没有收到任何错误，则启动他的 gRPC server

插件启动后要持续监控 Kubelet 的状态，并在 Kubelet 重启后重新注册自己。比如，Kubelet 刚启动后会清空 /var/lib/kubelet/device-plugins/ 目录，所以插件作者可以监控自己监听的 unix socket 是否被删除了，并根据此事件重新注册自己

Unix Socket

Device Plugin 和 Kubelet 通过在一个 Unix Socket上使用 gRPC 交互，当启动 gRPC server的时候，Device Plugin 将会在 /var/lib/kubelet/device-plugins/ 这个 HostPath 创建一个 UnixSocket，比如 /var/lib/kubelet/device-plugins/nvidiaGPU.sock。

在实现 Device 插件时需要注意

• 插件启动时，需要通过 /var/lib/kubelet/device-plugins/kubelet.sock 向 Kubelet 注册，同时提供插件的 Unix Socket 名称、API 的版本号和插件名称（格式为 vendor-domain/resource，如 nvidia.com/gpu）。Kubelet 会将这些设备暴露到 Node 状态中，方便后续调度器使用
• 插件启动后向 Kubelet 发送插件列表、按需分配设备并持续监控设备的实时状态

Protocol Overview

API specification

// Registration is the service advertised by the Kubelet
// Only when Kubelet answers with a success code to a Register Request
// may Device Plugins start their service
// Registration may fail when device plugin version is not supported by
// Kubelet or the registered resourceName is already taken by another
// active device plugin. Device plugin is expected to terminate upon registration failure
service Registration {
    rpc Register(RegisterRequest) returns (Empty) {}
}

message DevicePluginOptions {
  // Indicates if PreStartContainer call is required before each container start
    bool pre_start_required = 1;
}

message RegisterRequest {
    // Version of the API the Device Plugin was built against
    string version = 1;
    // Name of the unix socket the device plugin is listening on
    // PATH = path.Join(DevicePluginPath, endpoint)
    string endpoint = 2;
    // Schedulable resource name. As of now it's expected to be a DNS Label
    string resource_name = 3;
    // Options to be communicated with Device Manager
    options = 4;
}

message Empty {
}

// DevicePlugin is the service advertised by Device Plugins
service DevicePlugin {
    // GetDevicePluginOptions returns options to be communicated with Device
    // Manager
    rpc GetDevicePluginOptions(Empty) returns (DevicePluginOptions) {}

    // ListAndWatch returns a stream of List of Devices
    // Whenever a Device state change or a Device disapears, ListAndWatch
    // returns the new list
    rpc ListAndWatch(Empty) returns (stream ListAndWatchResponse) {}

    // Allocate is called during container creation so that the Device
    // Plugin can run device specific operations and instruct Kubelet
    // of the steps to make the Device available in the container
    rpc Allocate(AllocateRequest) returns (AllocateResponse) {}

    // PreStartContainer is called, if indicated by Device Plugin during registeration phase,
    // before each container start. Device plugin can run device specific operations
    // such as reseting the device before making devices available to the container
    rpc PreStartContainer(PreStartContainerRequest) returns (PreStartContainerResponse) {}
}

// ListAndWatch returns a stream of List of Devices
// Whenever a Device state change or a Device disapears, ListAndWatch
// returns the new list
message ListAndWatchResponse {
    repeated Device devices = 1;
}

/* E.g:
* struct Device {
*    ID: "GPU-fef8089b-4820-abfc-e83e-94318197576e",
*    State: "Healthy",
*} */
message Device {
    // A unique ID assigned by the device plugin used
    // to identify devices during the communication
    // Max length of this field is 63 characters
    string ID = 1;
    // Health of the device, can be healthy or unhealthy, see constants.go
    string health = 2;
}

// - PreStartContainer is expected to be called before each container start if indicated by plugin during registration phase.
// - PreStartContainer allows kubelet to pass reinitialized devices to containers.
// - PreStartContainer allows Device Plugin to run device specific operations on
//   the Devices requested
message PreStartContainerRequest {
    repeated string devicesIDs = 1;
}

// PreStartContainerResponse will be send by plugin in response to PreStartContainerRequest
message PreStartContainerResponse {
}

// - Allocate is expected to be called during pod creation since allocation
//   failures for any container would result in pod startup failure.
// - Allocate allows kubelet to exposes additional artifacts in a pod's
//   environment as directed by the plugin.
// - Allocate allows Device Plugin to run device specific operations on
//   the Devices requested
message AllocateRequest {
    repeated ContainerAllocateRequest container_requests = 1;
}

message ContainerAllocateRequest {
    repeated string devicesIDs = 1;
}

// AllocateResponse includes the artifacts that needs to be injected into
// a container for accessing 'deviceIDs' that were mentioned as part of
// 'AllocateRequest'.
// Failure Handling:
// if Kubelet sends an allocation request for dev1 and dev2.
// Allocation on dev1 succeeds but allocation on dev2 fails.
// The Device plugin should send a ListAndWatch update and fail the
// Allocation request
message AllocateResponse {
    repeated ContainerAllocateResponse container_responses = 1;
}

message ContainerAllocateResponse {
    // List of environment variable to be set in the container to access one of more devices.
    map<string, string> envs = 1;
    // Mounts for the container.
    repeated Mount mounts = 2;
    // Devices for the container.
    repeated DeviceSpec devices = 3;
    // Container annotations to pass to the container runtime
    map<string, string> annotations = 4;
}

// Mount specifies a host volume to mount into a container.
// where device library or tools are installed on host and container
message Mount {
    // Path of the mount within the container.
    string container_path = 1;
    // Path of the mount on the host.
    string host_path = 2;
    // If set, the mount is read-only.
    bool read_only = 3;
}

// DeviceSpec specifies a host device to mount into a container.
message DeviceSpec {
  // Path of the device within the container.
  string container_path = 1;
  // Path of the device on the host.
  string host_path = 2;
  // Cgroups permissions of the device, candidates are one or more of
  // * r - allows container to read from the specified device.
  // * w - allows container to write to the specified device.
  // * m - allows container to create device files that do not yet exist.
  string permissions = 3;
}

插件生命周期管理

插件启动时，以grpc的形式通过/var/lib/kubelet/device-plugins/kubelet.sock向Kubelet注册，同时提供插件的监听Unix Socket，API版本号和设备名称（比如nvidia.com/gpu）。Kubelet将会把这些设备暴露到Node状态中，以Extended Resource的要求发送到API server中，后续Scheduler会根据这些信息进行调度。

插件启动后，Kubelet会建立一个到插件的listAndWatch长连接，当插件检测到某个设备不健康的时候，就会主动通知Kubelet。此时如果这个设备处于空闲状态，Kubelet就会将其挪出可分配列表；如果该设备已经被某个pod使用，Kubelet就会将该Pod杀掉

插件启动后可以利用Kubelet的socket持续检查Kubelet的状态，如果Kubelet重启，插件也会相应的重启，并且重新向Kubelet注册自己

NVIDIA Device Plugin

NVIDIA 提供了一个基于 Device Plugins 接口的 GPU 设备插件 NVIDIA/k8s-device-plugin。

部署

kubectl apply -f https://raw.githubusercontent.com/NVIDIA/k8s-device-plugin/master/nvidia-device-plugin.yml

创建 Pod 时请求 GPU 资源

apiVersion: v1
kind: Pod
metadata:
  name: pod1
spec:
  restartPolicy: OnFailure
  containers:
  - image: nvidia/cuda
    name: pod1-ctr
    command: ["sleep"]
    args: ["100000"]

    resources:
      limits:
        nvidia.com/gpu: 1

注意：使用该插件时需要配置 nvidia-docker 2.0，并配置 nvidia 为默认运行时 （即配置 docker daemon 的选项 --default-runtime=nvidia）。nvidia-docker 2.0 的安装方法为（以 Ubuntu Xenial 为例，其他系统的安装方法可以参考 这里）：

整个Kubernetes调度GPU的过程如下：

• GPU Device plugin 部署到GPU节点上，通过 ListAndWatch 接口，上报注册节点的GPU信息和对应的DeviceID。
• 当有声明 nvidia.com/gpu 的GPU Pod创建出现，调度器会综合考虑GPU设备的空闲情况，将Pod调度到有充足GPU设备的节点上。
• 节点上的kubelet 启动Pod时，根据request中的声明调用各个Device plugin 的 allocate接口， 由于容器声明了GPU。 kubelet 根据之前 ListAndWatch 接口收到的Device信息，选取合适的设备，DeviceID 作为参数，调用GPU DevicePlugin的 Allocate 接口
• GPU DevicePlugin ，接收到调用，将DeviceID 转换为 NVIDIA_VISIBLE_DEVICES 环境变量，返回kubelet
• kubelet将环境变量注入到Pod， 启动容器
• 容器启动时， gpu-container-runtime 调用 gpu-containers-runtime-hook
• gpu-containers-runtime-hook 根据容器的 NVIDIA_VISIBLE_DEVICES 环境变量，转换为 --devices 参数，调用 nvidia-container-cli prestart
• nvidia-container-cli 根据 --devices ，将GPU设备映射到容器中。 并且将宿主机的Nvidia Driver Lib 的so文件也映射到容器中。 此时容器可以通过这些so文件，调用宿主机的Nvidia Driver。

在前面 API Specification 中，通过 Protobuf 定义了 DevicePlugin 应该提供的服务，在 Kubelet 中会调用 DevicePluginClient 来使用对应的服务，这里的 DevicePluginClient 即是通过 Protobuf 自动生成的代码。

k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1/api.pb.go

type DevicePluginClient interface {
    GetDevicePluginOptions(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*DevicePluginOptions, error)
    ListAndWatch(ctx context.Context, in *Empty, opts ...grpc.CallOption) (DevicePlugin_ListAndWatchClient, error)
    Allocate(ctx context.Context, in *AllocateRequest, opts ...grpc.CallOption) (*AllocateResponse, error)
    PreStartContainer(ctx context.Context, in *PreStartContainerRequest, opts ...grpc.CallOption) (*PreStartContainerResponse, error)
}

在 NVIDIA/k8s-device-plugin 中，我们可以看到上面不同服务的具体实现：

func (m *NvidiaDevicePlugin) GetDevicePluginOptions(context.Context, *pluginapi.Empty) (*pluginapi.DevicePluginOptions, error)
func (m *NvidiaDevicePlugin) ListAndWatch(e *pluginapi.Empty, s pluginapi.DevicePlugin_ListAndWatchServer) error
func (m *NvidiaDevicePlugin) Allocate(ctx context.Context, reqs *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error)
func (m *NvidiaDevicePlugin) PreStartContainer(context.Context, *pluginapi.PreStartContainerRequest) (*pluginapi.PreStartContainerResponse, error)

对 NVIDIA/k8s-device-plugin 来说，这里的关键数据结构为 NvidiaDevicePlugin，它实现了 Device Plugin 架构定义的API：

type NvidiaDevicePlugin struct {
    ResourceManager
    resourceName     string
    deviceListEnvvar string
    allocatePolicy   gpuallocator.Policy
    socket           string

    server        *grpc.Server
    cachedDevices []*Device
    health        chan *Device
    stop          chan interface{}
}

下面根据 Device Plugin 的生命周期，依次分析每个部分的实现机制。

NVIDIA DevicePlugin 启动

NVIDIA 的 k8s-device-plugin 启动之后逻辑如下，总的来说干了三件事：

• Serve：启动 gRPC server
• Register：向 Kubelet 注册给定的 resourceName
• CheckHealth：执行设备的健康检查逻辑，当检查到不健康的设备时，写到 unhealthy 的 channel 中

func (m *NvidiaDevicePlugin) Start() error {
    m.initialize()

    err := m.Serve()
  // ...

    err = m.Register()
  // ...
  
  go m.CheckHealth(m.stop, m.cachedDevices, m.health)

    return nil
}

Serve

Serve 监听在/var/lib/kubelet/device-plugins/nvidia-gpu.sock 这 个 Unix Socket，并且启动了 gRPC server，其他的就是启动失败重试的逻辑了。

func (m *NvidiaDevicePlugin) Serve() error {
    os.Remove(m.socket)
    sock, err := net.Listen("unix", m.socket)
    if err != nil {
        return err
    }

    pluginapi.RegisterDevicePluginServer(m.server, m)

    go func() {
        lastCrashTime := time.Now()
        restartCount := 0
        for {
            log.Printf("Starting GRPC server for '%s'", m.resourceName)
            err := m.server.Serve(sock)
            if err == nil {
                break
            }

            log.Printf("GRPC server for '%s' crashed with error: %v", m.resourceName, err)

            // restart if it has not been too often
            // i.e. if server has crashed more than 5 times and it didn't last more than one hour each time
            if restartCount > 5 {
                // quit
                log.Fatalf("GRPC server for '%s' has repeatedly crashed recently. Quitting", m.resourceName)
            }
            timeSinceLastCrash := time.Since(lastCrashTime).Seconds()
            lastCrashTime = time.Now()
            if timeSinceLastCrash > 3600 {
                // it has been one hour since the last crash.. reset the count
                // to reflect on the frequency
                restartCount = 1
            } else {
                restartCount++
            }
        }
    }()

    // Wait for server to start by launching a blocking connexion
    conn, err := m.dial(m.socket, 5*time.Second)
    if err != nil {
        return err
    }
    conn.Close()

    return nil
}

Register

Register 通过和 /var/lib/kubelet/device-plugins/kubelet.sock 这个 Unix Socket 向 Kubelet 注册，传递了 DevicePlugin 的 Unix Socket 的 Endpoint、资源的名称、API的版本号等信息。

func (m *NvidiaDevicePlugin) Register() error {
    conn, err := m.dial(pluginapi.KubeletSocket, 5*time.Second)
    if err != nil {
        return err
    }
    defer conn.Close()

    client := pluginapi.NewRegistrationClient(conn)
    reqt := &pluginapi.RegisterRequest{
        Version:      pluginapi.Version,
        Endpoint:     path.Base(m.socket),
        ResourceName: m.resourceName,
        Options: &pluginapi.DevicePluginOptions{
            GetPreferredAllocationAvailable: (m.allocatePolicy != nil),
        },
    }

    _, err = client.Register(context.Background(), reqt)
    if err != nil {
        return err
    }
    return nil
}

CheckHealth

这里调用了 nvml.NewEventSet 来监听 GPU 是否发生变化的事件，并且将 unhealthy Device 传递给 m.health 这个channel。

func checkHealth(stop <-chan interface{}, devices []*Device, unhealthy chan<- *Device) {
    disableHealthChecks := strings.ToLower(os.Getenv(envDisableHealthChecks))
    if disableHealthChecks == "all" {
        disableHealthChecks = allHealthChecks
    }
    if strings.Contains(disableHealthChecks, "xids") {
        return
    }

    eventSet := nvml.NewEventSet()
    defer nvml.DeleteEventSet(eventSet)

    for _, d := range devices {
        gpu, _, _, err := nvml.ParseMigDeviceUUID(d.ID)
        if err != nil {
            gpu = d.ID
        }

        err = nvml.RegisterEventForDevice(eventSet, nvml.XidCriticalError, gpu)
        if err != nil && strings.HasSuffix(err.Error(), "Not Supported") {
            log.Printf("Warning: %s is too old to support healthchecking: %s. Marking it unhealthy.", d.ID, err)
            unhealthy <- d
            continue
        }
        check(err)
    }

    for {
        select {
        case <-stop:
            return
        default:
        }

        e, err := nvml.WaitForEvent(eventSet, 5000)
        if err != nil && e.Etype != nvml.XidCriticalError {
            continue
        }

        // FIXME: formalize the full list and document it.
        // http://docs.nvidia.com/deploy/xid-errors/index.html#topic_4
        // Application errors: the GPU should still be healthy
        if e.Edata == 31 || e.Edata == 43 || e.Edata == 45 {
            continue
        }

        if e.UUID == nil || len(*e.UUID) == 0 {
            // All devices are unhealthy
            log.Printf("XidCriticalError: Xid=%d, All devices will go unhealthy.", e.Edata)
            for _, d := range devices {
                unhealthy <- d
            }
            continue
        }

        for _, d := range devices {
            // Please see https://github.com/NVIDIA/gpu-monitoring-tools/blob/148415f505c96052cb3b7fdf443b34ac853139ec/bindings/go/nvml/nvml.h#L1424
            // for the rationale why gi and ci can be set as such when the UUID is a full GPU UUID and not a MIG device UUID.
            gpu, gi, ci, err := nvml.ParseMigDeviceUUID(d.ID)
            if err != nil {
                gpu = d.ID
                gi = 0xFFFFFFFF
                ci = 0xFFFFFFFF
            }

            if gpu == *e.UUID && gi == *e.GpuInstanceId && ci == *e.ComputeInstanceId {
                log.Printf("XidCriticalError: Xid=%d on Device=%s, the device will go unhealthy.", e.Edata, d.ID)
                unhealthy <- d
            }
        }
    }
}

Kubelet DeviceManager

DeviceManager 启动

kubernetes/pkg/kubelet/cm/devicemanager/manager.go

type ManagerImpl struct {
    socketname string
    socketdir  string

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

    server *grpc.Server
    wg     sync.WaitGroup

    // 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 is a map by resource name of all the devices currently registered to the device manager
    allDevices map[string]map[string]pluginapi.Device

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

    // unhealthyDevices contains all of the unhealthy devices and their exported device IDs.
    unhealthyDevices 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
    checkpointManager checkpointmanager.CheckpointManager

    // List of NUMA Nodes available on the underlying machine
    numaNodes []int

    // Store of Topology Affinties that the Device Manager can query.
    topologyAffinityStore topologymanager.Store

    // devicesToReuse contains devices that can be reused as they have been allocated to
    // init containers.
    devicesToReuse PodReusableDevices
}

Device Manager 在 kubelet 启动时的 NewContainerManager 中创建,属于 containerManager 的子模块。

func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.Interface, nodeConfig NodeConfig, failSwapOn bool, devicePluginEnabled bool, recorder record.EventRecorder) (ContainerManager, error) {
  // ...
  
    klog.Infof("Creating device plugin manager: %t", devicePluginEnabled)
    if devicePluginEnabled {
        cm.deviceManager, err = devicemanager.NewManagerImpl(numaNodeInfo, cm.topologyManager)
        cm.topologyManager.AddHintProvider(cm.deviceManager)
    } else {
        cm.deviceManager, err = devicemanager.NewManagerStub()
    }

  // ...
}

具体创建 DeviceManager 的代码如下：

func newManagerImpl(socketPath string, numaNodeInfo cputopology.NUMANodeInfo, topologyAffinityStore topologymanager.Store) (*ManagerImpl, error) {
    klog.V(2).Infof("Creating Device Plugin manager at %s", socketPath)

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

    var numaNodes []int
    for node := range numaNodeInfo {
        numaNodes = append(numaNodes, node)
    }

    dir, file := filepath.Split(socketPath)
    manager := &ManagerImpl{
        endpoints: make(map[string]endpointInfo),

        socketname:            file,
        socketdir:             dir,
        allDevices:            make(map[string]map[string]pluginapi.Device),
        healthyDevices:        make(map[string]sets.String),
        unhealthyDevices:      make(map[string]sets.String),
        allocatedDevices:      make(map[string]sets.String),
        podDevices:            make(podDevices),
        numaNodes:             numaNodes,
        topologyAffinityStore: topologyAffinityStore,
        devicesToReuse:        make(PodReusableDevices),
    }
    manager.callback = manager.genericDeviceUpdateCallback

    // The following structures 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{}
    checkpointManager, err := checkpointmanager.NewCheckpointManager(dir)
    if err != nil {
        return nil, fmt.Errorf("failed to initialize checkpoint manager: %v", err)
    }
    manager.checkpointManager = checkpointManager

    return manager, nil
}

其中除了构建 DeviceManager 相关的结构之外，另外做的一个事情就是注册了一个 callback，用来处理对应 devices 的add，delete，update 事件。

func (m *ManagerImpl) genericDeviceUpdateCallback(resourceName string, devices []pluginapi.Device) {
    m.mutex.Lock()
    m.healthyDevices[resourceName] = sets.NewString()
    m.unhealthyDevices[resourceName] = sets.NewString()
    m.allDevices[resourceName] = make(map[string]pluginapi.Device)
    for _, dev := range devices {
        m.allDevices[resourceName][dev.ID] = dev
        if dev.Health == pluginapi.Healthy {
            m.healthyDevices[resourceName].Insert(dev.ID)
        } else {
            m.unhealthyDevices[resourceName].Insert(dev.ID)
        }
    }
    m.mutex.Unlock()
    if err := m.writeCheckpoint(); err != nil {
        klog.Errorf("writing checkpoint encountered %v", err)
    }
}

接下来到了 DeviceManager 启动的方法，它读取了 checkpoint file 中的数据，恢复 ManagerImpl中的相关数据，包括：

• podDevices
• allocatedDevices
• healthyDevices
• unhealthyDevices
• endpoints

然后将 /var/lib/kubelet/device-plugins/ 下面的除了 checkpiont文件 的所有文件清空，也就是清空所有的socket文件，包括自己的 kubelet.sock，以及其他所有之前的 DevicePlugin 的socket文件。最后创建 kubelet.sock 并启动 gRPC Server对外提供gRPC服务，其中 Register()用于 DevicePlugin 调用进行插件注册。

func (m *ManagerImpl) Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady) error {
    klog.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 {
        klog.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)
    if err = os.MkdirAll(m.socketdir, 0750); err != nil {
        return err
    }
    if selinux.SELinuxEnabled() {
        if err := selinux.SetFileLabel(m.socketdir, config.KubeletPluginsDirSELinuxLabel); err != nil {
            klog.Warningf("Unprivileged containerized plugins might not work. Could not set selinux context on %s: %v", m.socketdir, err)
        }
    }

    // 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 {
        klog.Errorf("Fail to clean up stale contents under %s: %v", m.socketdir, err)
    }

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

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

    pluginapi.RegisterRegistrationServer(m.server, m)
    go func() {
        defer m.wg.Done()
        m.server.Serve(s)
    }()

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

    return nil
}

DeviceManager 注册

DeviceManager 接收到 DevicePlugin的 RegisterRequest请求，其结构体如下

type RegisterRequest struct {
   Version string
   Endpoint string 
   ResourceName string 
   Options   *DevicePluginOptions 
}

检查注册的device Name、version是否符合 Extended Resource 的规则，Name不能属于kubernetes.i o，得有自己的domain，比如nvidia.com

根据 endpoint 信息创建 EndpointImpl 对象，即根据 endpoint 建立 socket 连接：

func (m *ManagerImpl) RegisterPlugin(pluginName string, endpoint string, versions []string) error {
    klog.V(2).Infof("Registering Plugin %s at endpoint %s", pluginName, endpoint)

    e, err := newEndpointImpl(endpoint, pluginName, m.callback)
    if err != nil {
        return fmt.Errorf("failed to dial device plugin with socketPath %s: %v", endpoint, err)
    }

    options, err := e.client.GetDevicePluginOptions(context.Background(), &pluginapi.Empty{})
    if err != nil {
        return fmt.Errorf("failed to get device plugin options: %v", err)
    }

    m.registerEndpoint(pluginName, options, e)
    go m.runEndpoint(pluginName, e)

    return nil
}

下面是 endPointsImpl 的具体实现：

type endpointImpl struct {
    client     pluginapi.DevicePluginClient
    clientConn *grpc.ClientConn

    socketPath   string
    resourceName string
    stopTime     time.Time

    mutex sync.Mutex
    cb    monitorCallback
}

func newEndpointImpl(socketPath, resourceName string, callback monitorCallback) (*endpointImpl, error) {
    client, c, err := dial(socketPath)
    if err != nil {
        klog.Errorf("Can't create new endpoint with path %s err %v", socketPath, err)
        return nil, err
    }

    return &endpointImpl{
        client:     client,
        clientConn: c,

        socketPath:   socketPath,
        resourceName: resourceName,

        cb: callback,
    }, nil
}

执行 EndpointImpl 对象的 run()，在 run方法中:

func (e *endpointImpl) run() {
    stream, err := e.client.ListAndWatch(context.Background(), &pluginapi.Empty{})
    if err != nil {
        klog.Errorf(errListAndWatch, e.resourceName, err)

        return
    }

    for {
        response, err := stream.Recv()
        if err != nil {
            klog.Errorf(errListAndWatch, e.resourceName, err)
            return
        }

        devs := response.Devices
        klog.V(2).Infof("State pushed for device plugin %s", e.resourceName)

        var newDevs []pluginapi.Device
        for _, d := range devs {
            newDevs = append(newDevs, *d)
        }

        e.callback(e.resourceName, newDevs)
    }
}

• 调用 DevicePlugin 的ListAndWatch gRPC 接口，通过长连接持续获取 ListAndWatch gRPC stream
• 从 stream 流中获取的devices详情列表然后调用Endpoint的 callback，也就是 ManagerImpl 注册的callback方法genericDeviceUpdateCallback进行Device Manager的缓存更新并写到checkpoint文件中
• run()是通过协程启动的，持续获取device server的ListAndWatch结果，持续更新device状态
• 当获取异常时，deviceManager断开连接，将device设置为不健康的状态。

ListAndWatch

看一下 DevicePlugin 实现的 ListAndWatch，先是立马返回device详情列表，然后开启协程，一旦感知device的健康状态发生变化了，更新 device 详情列表再次返回给 deviceManager。回想起健康检查，DevicePlugin 的 CheckHealth 就就会将设备的健康状态传递给 m.health 这个 channel。

func (m *NvidiaDevicePlugin) ListAndWatch(e *pluginapi.Empty, s pluginapi.DevicePlugin_ListAndWatchServer) error {
    s.Send(&pluginapi.ListAndWatchResponse{Devices: m.apiDevices()})

    for {
        select {
        case <-m.stop:
            return nil
        case d := <-m.health:
            // FIXME: there is no way to recover from the Unhealthy state.
            d.Health = pluginapi.Unhealthy
            log.Printf("'%s' device marked unhealthy: %s", m.resourceName, d.ID)
            s.Send(&pluginapi.ListAndWatchResponse{Devices: m.apiDevices()})
        }
    }
}

那么问题来了，DevicePlugin 是如何知道有多少 Device 的呢？我们看看 apiDevices 的实现：

func (m *NvidiaDevicePlugin) apiDevices() []*pluginapi.Device {
    var pdevs []*pluginapi.Device
    for _, d := range m.cachedDevices {
        pdevs = append(pdevs, &d.Device)
    }
    return pdevs
}

这里的 cachedDevices 是通过 ResourceManager 获得的 Device 信息，其具体通过 GpuDeviceManager 结构来实现，可以看到它们是调用了 nvml 库而实现的。这里还有一个 MigDeviceManager 本质上相同，不再概述。

func (g *GpuDeviceManager) Devices() []*Device {
    n, err := nvml.GetDeviceCount()
    check(err)

    var devs []*Device
    for i := uint(0); i < n; i++ {
        d, err := nvml.NewDeviceLite(i)
        check(err)

        migEnabled, err := d.IsMigEnabled()
        check(err)

        if migEnabled && g.skipMigEnabledGPUs {
            continue
        }

        devs = append(devs, buildDevice(d))
    }

    return devs
}

Allocation

kubelet 接收到被调度到本节点的pods后

HandlePodAdditions

当 Node 上的 Kubelet 监听到有新的 Pod 创建时，会调用 HandlerPodAdditions 来处理 Pod 创建的事件。

func (kl *Kubelet) syncLoopIteration(configCh <-chan kubetypes.PodUpdate, handler SyncHandler,
    syncCh <-chan time.Time, housekeepingCh <-chan time.Time, plegCh <-chan *pleg.PodLifecycleEvent) bool {
    select {
    case u, open := <-configCh:
        switch u.Op {
        case kubetypes.ADD:
            klog.V(2).Infof("SyncLoop (ADD, %q): %q", u.Source, format.Pods(u.Pods))
            handler.HandlePodAdditions(u.Pods)
        case kubetypes.UPDATE:
            klog.V(2).Infof("SyncLoop (UPDATE, %q): %q", u.Source, format.PodsWithDeletionTimestamps(u.Pods))
            handler.HandlePodUpdates(u.Pods)
    // ...
        }
    case e := <-plegCh:
    // ...
  }
  return true
}

接下来进一步看下 HandlerPodAdditions 的实现，对于传入的每一个 Pod ，如果它没有被 terminate，则通过 canAdmitPod 检查是否可以允许该 Pod 创建。

func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {
    start := kl.clock.Now()
    sort.Sort(sliceutils.PodsByCreationTime(pods))
    for _, pod := range pods {
        existingPods := kl.podManager.GetPods()
        kl.podManager.AddPod(pod)
   
    // ...
        if !kl.podIsTerminated(pod) {
            activePods := kl.filterOutTerminatedPods(existingPods)
            if ok, reason, message := kl.canAdmitPod(activePods, pod); !ok {
                kl.rejectPod(pod, reason, message)
                continue
            }
        }
        // ...
    }
}

canAdmitPod 里面，Kubelet 将会依次执行每一个 admit handler 来看 Pod 能否通过。

// "pod" is new pod, while "pods" are all admitted pods
func (kl *Kubelet) canAdmitPod(pods []*v1.Pod, pod *v1.Pod) (bool, string, string) {
    attrs := &lifecycle.PodAdmitAttributes{Pod: pod, OtherPods: pods}
    for _, podAdmitHandler := range kl.admitHandlers {
        if result := podAdmitHandler.Admit(attrs); !result.Admit {
            return false, result.Reason, result.Message
        }
    }

    return true, "", ""
}

admitHandlers 是一个 PodAdmitHandler 的切片，其接口如下：

type PodAdmitHandler interface {
    // Admit evaluates if a pod can be admitted.
    Admit(attrs *PodAdmitAttributes) PodAdmitResult
}

Kubelet 在创建的时候会添加一系列的 PodAdmitHandler 用于检查，对pod的资源做一些准入判断，比如：

• evictionAdmitHandler :当节点有内存压力时，拒绝创建best effort的pod，还有其它条件先略过
• TopologyPodAdmitHandler：拒绝创建因为Topology locality冲突而无法分配资源的pod

  klet.admitHandlers.AddPodAdmitHandler(evictionAdmitHandler)
  klet.admitHandlers.AddPodAdmitHandler(klet.containerManager.GetAllocateResourcesPodAdmitHandler())
// ...

与我们 DevicePlugin 相关的则是 containerManager 的 resourceAllocator，这里会分别调用 DeviceManager 和 CpuManager 的 Allocate 函数，看是否能够申请到相关的资源。这里会对 Pod 的每一个 InitContainer 和 Container检查，看能否申请到。

func (m *resourceAllocator) Admit(attrs *lifecycle.PodAdmitAttributes) lifecycle.PodAdmitResult {
    pod := attrs.Pod

    for _, container := range append(pod.Spec.InitContainers, pod.Spec.Containers...) {
        err := m.deviceManager.Allocate(pod, &container)
        if err != nil {
            return lifecycle.PodAdmitResult{
                Message: fmt.Sprintf("Allocate failed due to %v, which is unexpected", err),
                Reason:  "UnexpectedAdmissionError",
                Admit:   false,
            }
        }

        if m.cpuManager != nil {
            err = m.cpuManager.Allocate(pod, &container)
            if err != nil {
                return lifecycle.PodAdmitResult{
                    Message: fmt.Sprintf("Allocate failed due to %v, which is unexpected", err),
                    Reason:  "UnexpectedAdmissionError",
                    Admit:   false,
                }
            }
        }
    }

    return lifecycle.PodAdmitResult{Admit: true}
}

接下来我们看 ManagerImpl 的 Allocate 函数实现。

ManagerImpl.Allocate

• allocateContainerResources为Pod中的init container分配devices，并更新deviceManager中PodDevices缓存；
• allocateContainerResources为 Pod中的regular container分配devices，并更新deviceManager中PodDevices缓存
  • 每次在为Pod分配devices之前，都去检查一下此时的active pods，并与podDevices缓存中的pods进行比对，将已经terminated的Pods的devices从podDevices中删除，即进行了devices的GC操作。
  • 从 healthyDevices 中随机分配对应数量的devices给该Pod，并注意更新allocatedDevices，否则会导致一个device被分配给多个Pod。
  • 拿到devices后，就通过Grpc调用 DevicePlugin 的 Allocate方法，DevicePlugin 返回 ContainerAllocateResponse (包括注入的环境变量、挂载信息、Annotations)，deviceManager
  • 根据 pod uuid 和 container name 将返回的信息存入 podDevices 缓存，更新 podDevices 缓存信息，并将deviceManager 中缓存数据更新到 checkpoint 文件中。

func (m *ManagerImpl) Allocate(pod *v1.Pod, container *v1.Container) error {
    if _, ok := m.devicesToReuse[string(pod.UID)]; !ok {
        m.devicesToReuse[string(pod.UID)] = make(map[string]sets.String)
    }
    // If pod entries to m.devicesToReuse other than the current pod exist, delete them.
    for podUID := range m.devicesToReuse {
        if podUID != string(pod.UID) {
            delete(m.devicesToReuse, podUID)
        }
    }
    // Allocate resources for init containers first as we know the caller always loops
    // through init containers before looping through app containers. Should the caller
    // ever change those semantics, this logic will need to be amended.
    for _, initContainer := range pod.Spec.InitContainers {
        if container.Name == initContainer.Name {
            if err := m.allocateContainerResources(pod, container, m.devicesToReuse[string(pod.UID)]); err != nil {
                return err
            }
            m.podDevices.addContainerAllocatedResources(string(pod.UID), container.Name, m.devicesToReuse[string(pod.UID)])
            return nil
        }
    }
    if err := m.allocateContainerResources(pod, container, m.devicesToReuse[string(pod.UID)]); err != nil {
        return err
    }
    m.podDevices.removeContainerAllocatedResources(string(pod.UID), container.Name, m.devicesToReuse[string(pod.UID)])
    return nil

}

接下来我们看 allocateContainerResource 的实现，因为扩展资源是DevicePlugin 所发现的，而扩展资源不允许过量提交，因此要求容器中的 Request 与 Limits 相等，并且 DevicePlugin 会遍历所有的 Limits 保证资源是充足的。

func (m *ManagerImpl) allocateContainerResources(pod *v1.Pod, container *v1.Container, devicesToReuse map[string]sets.String) error {
    podUID := string(pod.UID)
    contName := container.Name
    allocatedDevicesUpdated := false
    // Extended resources are not allowed to be overcommitted.
    // Since device plugin advertises extended resources,
    // therefore Requests must be equal to Limits and iterating
    // over the Limits should be sufficient.
    for k, v := range container.Resources.Limits {
        resource := string(k)
        needed := int(v.Value())
        klog.V(3).Infof("needs %d %s", needed, resource)
        if !m.isDevicePluginResource(resource) {
            continue
        }
        // Updates allocatedDevices to garbage collect any stranded resources
        // before doing the device plugin allocation.
        if !allocatedDevicesUpdated {
            m.UpdateAllocatedDevices()
            allocatedDevicesUpdated = true
        }
        allocDevices, err := m.devicesToAllocate(podUID, contName, resource, needed, devicesToReuse[resource])
        if err != nil {
            return err
        }
        if allocDevices == nil || len(allocDevices) <= 0 {
            continue
        }

        startRPCTime := time.Now()
        // Manager.Allocate involves RPC calls to device plugin, which
        // could be heavy-weight. Therefore we want to perform this operation outside
        // mutex lock. Note if Allocate call fails, we may leave container resources
        // partially allocated for the failed container. We rely on UpdateAllocatedDevices()
        // to garbage collect these resources later. Another side effect is that if
        // we have X resource A and Y resource B in total, and two containers, container1
        // and container2 both require X resource A and Y resource B. Both allocation
        // requests may fail if we serve them in mixed order.
        // TODO: may revisit this part later if we see inefficient resource allocation
        // in real use as the result of this. Should also consider to parallelize device
        // plugin Allocate grpc calls if it becomes common that a container may require
        // resources from multiple device plugins.
        m.mutex.Lock()
        eI, ok := m.endpoints[resource]
        m.mutex.Unlock()
        if !ok {
            m.mutex.Lock()
            m.allocatedDevices = m.podDevices.devices()
            m.mutex.Unlock()
            return fmt.Errorf("unknown Device Plugin %s", resource)
        }

        devs := allocDevices.UnsortedList()
        // TODO: refactor this part of code to just append a ContainerAllocationRequest
        // in a passed in AllocateRequest pointer, and issues a single Allocate call per pod.
        klog.V(3).Infof("Making allocation request for devices %v for device plugin %s", devs, resource)
        resp, err := eI.e.allocate(devs)
        metrics.DevicePluginAllocationDuration.WithLabelValues(resource).Observe(metrics.SinceInSeconds(startRPCTime))
        if err != nil {
            // In case of allocation failure, we want to restore m.allocatedDevices
            // to the actual allocated state from m.podDevices.
            m.mutex.Lock()
            m.allocatedDevices = m.podDevices.devices()
            m.mutex.Unlock()
            return err
        }

        if len(resp.ContainerResponses) == 0 {
            return fmt.Errorf("no containers return in allocation response %v", resp)
        }

        // Update internal cached podDevices state.
        m.mutex.Lock()
        m.podDevices.insert(podUID, contName, resource, allocDevices, resp.ContainerResponses[0])
        m.mutex.Unlock()
    }

    // Checkpoints device to container allocation information.
    return m.writeCheckpoint()
}

我们看到，这里通过 resp, err := eI.e.allocate(devs) 执行 RPC 调用，进入到了 DevicePlugin 的逻辑。这里有一个问题，RPC 远程调用中的 deviceIDs 参数是怎么来的呢？我们看到这里有一个 devicesToAllocate的调用。这里的主要逻辑如下：

• 拿到对应Pod的对应容器已经申请的资源的设备列表，检查是否只申请了部分，如果只有一部分，那么报错
• 然后从 resuableDevices 结构中拿到可以使用的设备列表，如果可用的足够则返回，否则继续从 healthyDevices 中找
• 从 healthyDevices 去掉已经在使用的设备，然后检查是否足够，如果不够则报错
• 如果足够的话，根据是否有满足拓扑亲和性去拿到足够的设备列表

func (m *ManagerImpl) devicesToAllocate(podUID, contName, resource string, required int, reusableDevices sets.String) (sets.String, error) {
    m.mutex.Lock()
    defer m.mutex.Unlock()
    needed := required
    // Gets list of devices that have already been allocated.
    // This can happen if a container restarts for example.
    devices := m.podDevices.containerDevices(podUID, contName, resource)
    if devices != nil {
        klog.V(3).Infof("Found pre-allocated devices for resource %s container %q in Pod %q: %v", resource, contName, podUID, devices.List())
        needed = needed - devices.Len()
        // A pod's resource is not expected to change once admitted by the API server,
        // so just fail loudly here. We can revisit this part if this no longer holds.
        if needed != 0 {
            return nil, fmt.Errorf("pod %q container %q changed request for resource %q from %d to %d", podUID, contName, resource, devices.Len(), required)
        }
    }
    if needed == 0 {
        // No change, no work.
        return nil, nil
    }
    klog.V(3).Infof("Needs to allocate %d %q for pod %q container %q", needed, resource, podUID, contName)
    // Needs to allocate additional devices.
    if _, ok := m.healthyDevices[resource]; !ok {
        return nil, fmt.Errorf("can't allocate unregistered device %s", resource)
    }
    devices = sets.NewString()
    // Allocates from reusableDevices list first.
    for device := range reusableDevices {
        devices.Insert(device)
        needed--
        if needed == 0 {
            return devices, nil
        }
    }
    // Needs to allocate additional devices.
    if m.allocatedDevices[resource] == nil {
        m.allocatedDevices[resource] = sets.NewString()
    }
    // Gets Devices in use.
    devicesInUse := m.allocatedDevices[resource]
    // Gets a list of available devices.
    available := m.healthyDevices[resource].Difference(devicesInUse)
    if available.Len() < needed {
        return nil, fmt.Errorf("requested number of devices unavailable for %s. Requested: %d, Available: %d", resource, needed, available.Len())
    }
    // By default, pull devices from the unsorted list of available devices.
    allocated := available.UnsortedList()[:needed]
    // If topology alignment is desired, update allocated to the set of devices
    // with the best alignment.
    hint := m.topologyAffinityStore.GetAffinity(podUID, contName)
    if m.deviceHasTopologyAlignment(resource) && hint.NUMANodeAffinity != nil {
        allocated = m.takeByTopology(resource, available, hint.NUMANodeAffinity, needed)
    }
    // Updates m.allocatedDevices with allocated devices to prevent them
    // from being allocated to other pods/containers, given that we are
    // not holding lock during the rpc call.
    for _, device := range allocated {
        m.allocatedDevices[resource].Insert(device)
        devices.Insert(device)
    }
    return devices, nil
}

RPC 调用成功后，会将对应的 Response 记录到 m.podDevices 中。

func (pdev podDevices) insert(podUID, contName, resource string, devices sets.String, resp *pluginapi.ContainerAllocateResponse) {
    if _, podExists := pdev[podUID]; !podExists {
        pdev[podUID] = make(containerDevices)
    }
    if _, contExists := pdev[podUID][contName]; !contExists {
        pdev[podUID][contName] = make(resourceAllocateInfo)
    }
    pdev[podUID][contName][resource] = deviceAllocateInfo{
        deviceIds: devices,
        allocResp: resp,
    }
}

DevicePlugin.Allocate

Allocate 接口给容器加上 NVIDIA_VISIBLE_DEVICES 环境变量，设置了相关的 DeviceSpec参数，将 Response 返回给 Kubelet。

func (m *NvidiaDevicePlugin) Allocate(ctx context.Context, reqs *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
    responses := pluginapi.AllocateResponse{}
    for _, req := range reqs.ContainerRequests {
        for _, id := range req.DevicesIDs {
            if !m.deviceExists(id) {
                return nil, fmt.Errorf("invalid allocation request for '%s': unknown device: %s", m.resourceName, id)
            }
        }

        response := pluginapi.ContainerAllocateResponse{}

        if *deviceListStrategyFlag == DeviceListStrategyEnvvar {
            response.Envs = m.apiEnvs(m.deviceListEnvvar, req.DevicesIDs)
        }
        if *deviceListStrategyFlag == DeviceListStrategyVolumeMounts {
            response.Envs = m.apiEnvs(m.deviceListEnvvar, []string{deviceListAsVolumeMountsContainerPathRoot})
            response.Mounts = m.apiMounts(req.DevicesIDs)
        }
        if *passDeviceSpecs {
            response.Devices = m.apiDeviceSpecs(req.DevicesIDs)
        }

        responses.ContainerResponses = append(responses.ContainerResponses, &response)
    }

    return &responses, nil
}

前面我们提到， Nvidia的 gpu-container-runtime 根据容器的 NVIDIA_VISIBLE_DEVICES 环境变量，会决定这个容器是否为GPU容器，并且可以使用哪些GPU设备。 而Nvidia GPU device plugin做的事情，就是根据kubelet 请求中的GPU DeviceId， 转换为 NVIDIA_VISIBLE_DEVICES 环境变量返回给kubelet， kubelet收到返回内容后，会自动将返回的环境变量注入到容器中。当容器中包含环境变量，启动时 gpu-container-runtime 会根据 NVIDIA_VISIBLE_DEVICES 里声明的设备信息，将设备映射到容器中，并将对应的Nvidia Driver Lib 也映射到容器中。

Device 的使用

在kubelet的 GetResource 中，会调用 DeviceManager 的 GetDeviceRunContainerOptions，并将这些 options添加到kubecontainer.RunContainerOptions 中。RunContainerOptions 包括 Envs、Mounts、Devices、PortMappings、Annotations等信息。kubelet调用 GetResources() 为启动container获取启动参数 runtimeapi.ContainerConfig{Args...}

kubernetes/pkg/kubelet/cm/container_manager_linux.go

func (cm *containerManagerImpl) GetResources(pod *v1.Pod, container *v1.Container) (*kubecontainer.RunContainerOptions, error) {
    opts := &kubecontainer.RunContainerOptions{}
    // Allocate should already be called during predicateAdmitHandler.Admit(),
    // just try to fetch device runtime information from cached state here
    devOpts, err := cm.deviceManager.GetDeviceRunContainerOptions(pod, container)
    if err != nil {
        return nil, err
    } else if devOpts == nil {
        return opts, nil
    }
    opts.Devices = append(opts.Devices, devOpts.Devices...)
    opts.Mounts = append(opts.Mounts, devOpts.Mounts...)
    opts.Envs = append(opts.Envs, devOpts.Envs...)
    opts.Annotations = append(opts.Annotations, devOpts.Annotations...)
    return opts, nil
}

GetDeviceRunContainerOptions() 根据 pod uuid 和 container name 从 podDevices 缓存（device的分配过程中会设置缓存数据）中取出Envs、Mounts、Devices、PortMappings、Annotations等信息，另外对于一些PreStartRequired为true的 DevicePlugin，deviceManager需要在启动container之前调用 DevicePlugin的 PreStartContainergrpc接口，做一些device的初始化工作，超时时间限制为30秒。

func (m *ManagerImpl) GetDeviceRunContainerOptions(pod *v1.Pod, container *v1.Container) (*DeviceRunContainerOptions, error) {
    podUID := string(pod.UID)
    contName := container.Name
    needsReAllocate := false
    for k := range container.Resources.Limits {
        resource := string(k)
        if !m.isDevicePluginResource(resource) {
            continue
        }
        err := m.callPreStartContainerIfNeeded(podUID, contName, resource)
        if err != nil {
            return nil, err
        }
        // This is a device plugin resource yet we don't have cached
        // resource state. This is likely due to a race during node
        // restart. We re-issue allocate request to cover this race.
        if m.podDevices.containerDevices(podUID, contName, resource) == nil {
            needsReAllocate = true
        }
    }
    if needsReAllocate {
        klog.V(2).Infof("needs re-allocate device plugin resources for pod %s, container %s", podUID, container.Name)
        if err := m.Allocate(pod, container); err != nil {
            return nil, err
        }
    }
    m.mutex.Lock()
    defer m.mutex.Unlock()
    return m.podDevices.deviceRunContainerOptions(string(pod.UID), container.Name), nil
}

Device 的状态管理

device的状态管理涉及到以下3个部分：

• node上的device状态管理当kubelet更新node status时会调用GetCapacity更新device plugins对应的Resource信息。

kubelet_node_status.go调用deviceManager的GetCapacity()获取device的状态，将device状态添加到node info并通过kube-apiserver存入etcd，GetCapacity()返回device server含有的所有device、已经分配给pod使用的device、pod不能使用的device即no-active的device kubelet_node_status.go根据返回的数据更新node info

• kubelet deviceManager服务的device状态管理其实在device的注册、device分配中都有讲解，即使用checkpoint机制默认是将podDevices以 PodDevicesEntry的格式存入/var/lib/kubelet/device-plugins/kubelet_internal_checkpoint 文件

type PodDevicesEntry struct {
   PodUID        string
   ContainerName string
   ResourceName  string
   DeviceIDs     []string
   AllocResp     []byte     //包含启动container时使用的Envs、Mounts、Devices、PortMappings、Annotations等信息
}

只要device的状态发生了变化（如注册新device、device被分配、device的健康状态发生变化、device被删除），就要将podDevices存入kubelet_internal_checkpoint 文件。kubelet在启动或重启时，都需要读取kubelet_internal_checkpoint 文件里的数据，并以podDevices格式存入podDevices缓存。

• DevicePlugin 上报device状态在device的注册部分已经讲解过，归纳为
  • deviceManager 注册完 DevicePlugin 后，会跟 DevicePlugin 建立长连接，持续获取 DevicePlugin 的ListAndWatch结果，持续更新device状态；
  • 当获取异常时，deviceManager断开连接，将device设置为不健康的状态；
  • DevicePlugin 默认会重启重新注册，重新上报device的状态

参考资料

• Kubernetes device plugin design proposal
• Kubernetes plugin watcher design proposal
• Nvidia Device Plugin
• https://cloud.tencent.com/developer/article/1592800