Kubelet Device Plugin 的工作机制

Kubernetes provides adevice plugin frameworkthat you can use to advertise system hardware
resources to theKubelet.
Instead of customizing the code for Kubernetes itself, vendors can implement adevice plugin that 
you deploy either manually or as aDaemonSet.The targeted devices include GPUs, high-performance
NICs, FPGAs, InfiniBand adapters,and other similar computing resources that may require vendor
specific initializationand setup.

type DevicePluginClient interface {
   // 获取device server的管理plugin的参数，如PreStartRequired用来判断container启动使用plugin之前
   // 是否需要执行PreStartContainer()
   GetDevicePluginOptions(ctx context.Context, in *Empty, opts ...grpc.CallOption)
   (*DevicePluginOptions, error)
   // 监听plugin的状态，当plugin的状态发生变化时及时反馈到deviceManager
   ListAndWatch(ctx context.Context, in *Empty, opts ...grpc.CallOption)
   (DevicePlugin_ListAndWatchClient, error)
   // device server分配plugin，返回container使用该plugin的参数
   Allocate(ctx context.Context, in *AllocateRequest, opts ...grpc.CallOption) 
   (*AllocateResponse, error)
   // container启动使用plugin之前需要执行的操作
   PreStartContainer(ctx context.Context, in *PreStartContainerRequest, opts ...grpc.CallOption)
   (*PreStartContainerResponse, error)
}

if event.Name == pluginapi.KubeletSocket && event.Op&fsnotify.Create == fsnotify.Create {
   log.Printf("inotify: %s created, restarting.", pluginapi.KubeletSocket)
   goto restart
}

type ManagerImpl struct {
   // kubelet对外暴露的socket名，即 kubelet.sock
   socketname string
   // device plugins' socket的存放的目录，/var/lib/kubelet/device-plugins/
   socketdir  string
   // map对象，key为Resource Name，value为endpoint接口(包括run，stop，allocate，preStartContainer，
   // getDevices，callback，isStoped，StopGracePeriodExpired)，每个endpoint接口对应一个已注册的device
   // plugin，负责与device plugin的gRPC通信及缓存device plugin反馈的device states。
   endpoints map[string]endpointInfo
   mutex     sync.Mutex
   // Register服务暴露的gRPC Server
   server *grpc.Server
   wg     sync.WaitGroup
   // 用来获取该节点上所有active pods，即non-Terminated状态的Pods, 即：
   // allPods := kl.podManager.GetPods()
   // activePods := kl.filterOutTerminatedPods(allPods)
   activePods ActivePodsFunc
   sourcesReady config.SourcesReady
   // kubelet收到device plugin的ListAndWatch gRCP stream中有devices state变更时的回调函数，
   // 包括有新设备增加、旧设备删除、设备状态变化，所以通过ListAndWatch接口的回调方式，可以实现设备的自动发现
   // 和热插拔。
   callback monitorCallback
   // 缓存所有设备信息，map[设备类型名][设备id]设备详情
   allDevices map[string]map[string]pluginapi.Device
   // key为Resource Name，value为对应的健康的device IDs。
   healthyDevices map[string]sets.String
   // key为Resource Name，value为对应的不健康的device IDs。
   unhealthyDevices map[string]sets.String
   // key为Resource Name，value为已经分配出去的device IDs。
   allocatedDevices map[string]sets.String
   // 记录每个pod中每个容器的device分配情况
   podDevices        podDevices
   // 在/var/lib/kubelet/device-plugins/目录下创建file store类型的key-value存储文件
   // kubelet_internal_checkpoint，用来作为kubelet的device plugin的checkpoint。

   checkpointManager checkpointmanager.CheckpointManager
   numaNodes []int
   // 本地存储设备和pod的对应信息(/var/lib/kubelet/device-plugins/kubelet_internal_checkpoint)，
   // 具体存储了每个Pod分配的Devices信息PodDeviceEntries, 以及已经注册的Resource Name及对应的Devices IDs
   topologyAffinityStore topologymanager.Store
}

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) {
        
...略
if devicePluginEnabled {
   cm.deviceManager, err = devicemanager.NewManagerImpl(numaNodeInfo, cm.topologyManager)
   cm.topologyManager.AddHintProvider(cm.deviceManager)
}

//注册callback为genericDeviceUpdateCallback，用来处理对应devices的add,delete,update事件。
manager.callback = manager.genericDeviceUpdateCallback

// 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)
}

// Detect a kubelet restart by watching for a newly created
// 'pluginapi.KubeletSocket' file. When this occurs, restart this loop,
// restarting all of the plugins in the process.
case event := <-watcher.Events:
   if event.Name == pluginapi.KubeletSocket && event.Op&fsnotify.Create == fsnotify.Create {
      log.Printf("inotify: %s created, restarting.", pluginapi.KubeletSocket)
      goto restart
   }

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

func (e *endpointImpl) run() {
	//建立stream conn
	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)
	}
}

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

   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.cachedDevices})
      }
   }
}

func (cm *containerManagerImpl) GetPluginRegistrationHandler() cache.PluginHandler {
   return cm.deviceManager.GetWatcherHandler()
}
if err := handler.RegisterPlugin(infoResp.Name, infoResp.Endpoint, infoResp.SupportedVersions); err != nil {
   return og.notifyPlugin(client, false, fmt.Sprintf("RegisterPlugin error -- plugin registration failed with err: %v", err))
}

# pwd
/var/lib/kubelet
# tree
.
├── config.yaml   // kubelet的配置文件
├── cpu_manager_state   // cpu manager的checkpoint文件
├── device-plugins      // device manager的工作目录
│   ├── DEPRECATION
│   ├── kubelet_internal_checkpoint // device manager的checkpoint文件，
// 具体存储了每个Pod分配的Devices信息PodDeviceEntries, 以及已经注册的Resource Name及对应的Devices IDs。
│   └── kubelet.sock    // device manager和device server通信的UDS文件
├── kubeadm-flags.env   // kubeadm配置kubelet服务的配置参数
├── plugins             // 存放kubelet plugin manager和plugin server交互的UDS文件的目录
│   ├── kubernetes.io
│   │   └── csi
│   │       └── pv
│   └── xxx.csi.xxx.io
│       └── csi.sock    // csi server的UDS文件
├── plugins_registry
│   └── xxx.csi.xxx.io-reg.sock  // 用户注册csi server的UDS文件
├── pod-resources
│   └── kubelet.sock    // PodResources service的工作目录

func (kl *Kubelet) ListenAndServePodResources() {
   socket, err := util.LocalEndpoint(kl.getPodResourcesDir(), podresources.Socket)
   if err != nil {
      klog.V(2).Infof("Failed to get local endpoint for PodResources endpoint: %v", err)
      return
   }
   server.ListenAndServePodResources(socket, kl.podManager, kl.containerManager)
}

// PodResources is a service provided by the kubelet that provides information about the
// node resources consumed by pods and containers on the node
service PodResources {
    rpc List(ListPodResourcesRequest) returns (ListPodResourcesResponse) {}
}

// List returns information about the resources assigned to pods on the node
func (p *podResourcesServer) List(ctx context.Context, req *v1alpha1.ListPodResourcesRequest) 
                                 (*v1alpha1.ListPodResourcesResponse, error) {
…
for j, container := range pod.Spec.Containers {
   pRes.Containers[j] = &v1alpha1.ContainerResources{
      Name:    container.Name,
      Devices: p.devicesProvider.GetDevices(string(pod.UID), container.Name),
   }
}
...
}

func NewPredicateAdmitHandler(getNodeAnyWayFunc getNodeAnyWayFuncType, 
admissionFailureHandler AdmissionFailureHandler, 
pluginResourceUpdateFunc pluginResourceUpdateFuncType) *predicateAdmitHandler {
   return &predicateAdmitHandler{
      getNodeAnyWayFunc,
      pluginResourceUpdateFunc,
      admissionFailureHandler,
   }
}

func (cm *containerManagerImpl) UpdatePluginResources(node *schedulernodeinfo.NodeInfo, 
attrs *lifecycle.PodAdmitAttributes) error {
   return cm.deviceManager.Allocate(node, attrs)
}

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...)

ctx, cancel := context.WithTimeout(context.Background(), pluginapi.KubeletPreStartContainerRPCTimeoutInSecs*time.Second)
defer cancel()
return e.client.PreStartContainer(ctx, &pluginapi.PreStartContainerRequest{
   DevicesIDs: devs,
})

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