深入分析kubelet（7）—— 选取GPU挂载

深入浅出kubernetes之device-plugins主要分析device-plugin资源上报部分，本来着重分析下分配过程。

device-plugin

kubelet过于复杂，所以通过device-plugin反推

interface

kubernetes\pkg\kubelet\apis\deviceplugin\v1beta1\api.pb.go

type DevicePluginServer interface {
   GetDevicePluginOptions(context.Context, *Empty) (*DevicePluginOptions, error)

   ListAndWatch(*Empty, DevicePlugin_ListAndWatchServer) error

   Allocate(context.Context, *AllocateRequest) (*AllocateResponse, error)

   PreStartContainer(context.Context, *PreStartContainerRequest) (*PreStartContainerResponse, error)
}

最重要的是ListAndWatch()/Allocate()，因为另外两个方法直接返回结果，没有任何逻辑

ListAndWatch

k8s-device-plugin\server.go

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

   for {
      select {
      case <-m.stop:
         return nil
      case d := <-m.health:
         d.Health = pluginapi.Unhealthy
         s.Send(&pluginapi.ListAndWatchResponse{Devices: m.devs})
      }
   }
}

老朋友了，list所有设备，并长连接http-steaming将变化发到客户端。

// E.g:
// struct Device {
//    ID: "GPU-fef8089b-4820-abfc-e83e-94318197576e",
//    State: "Healthy",
// }
type Device struct {
   ID string `protobuf:"bytes,1,opt,name=ID,json=iD,proto3" json:"ID,omitempty"`
   Health string `protobuf:"bytes,2,opt,name=health,proto3" json:"health,omitempty"`
}

目前设备信息只有设备号和健康状态，没办法扩展，所以也就不知道GPU拓扑=。=，所以说目前也就支持GPU数量。

Allocate

func (m *NvidiaDevicePlugin) Allocate(ctx context.Context, reqs *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
   devs := m.devs
   responses := pluginapi.AllocateResponse{}
   for _, req := range reqs.ContainerRequests {
      response := pluginapi.ContainerAllocateResponse{
         Envs: map[string]string{
            "NVIDIA_VISIBLE_DEVICES": strings.Join(req.DevicesIDs, ","),
         },
      }

      for _, id := range req.DevicesIDs {
         if !deviceExists(devs, id) {
            return nil, fmt.Errorf("invalid allocation request: unknown device: %s", id)
         }
      }

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

   return &responses, nil
}

Allocate做了两件事情，返回NVIDIA_VISIBLE_DEVICES环境变量，以及检查设备是否存在。

Note:

1. 这里其实就已经告诉了我们分配逻辑，即kubelet根据limit选择挂载具体的GPU卡，然后将设备号发送给device-plugin，得到env；
2. 以后想在调度器里面根据GPU拓扑选择GPU卡，是很难实现的，并且调度器本身逻辑只创建bind，赋值node name，要想再把设备号加进去比较困难。

kubelet

从上面我们可以知道最重要的就是Allocate方法，所以我们首先去找kubelet中Allocate方法的调用。

kubernetes\pkg\kubelet\cm\devicemanager\endpoint.go

type endpoint interface {
   run()
   stop()
   allocate(devs []string) (*pluginapi.AllocateResponse, error)
   preStartContainer(devs []string) (*pluginapi.PreStartContainerResponse, error)
   callback(resourceName string, devices []pluginapi.Device)
   isStopped() bool
   stopGracePeriodExpired() bool
}

其中最重要的就是run和allocate，分别会调用device-plugin的ListAndWatch()和Allocate()。

run

func (e *endpointImpl) run() {
   stream, err := e.client.ListAndWatch(context.Background(), &pluginapi.Empty{})

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

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

调用ListAndWatch，再调用callback处理设备

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

func (m *ManagerImpl) genericDeviceUpdateCallback(resourceName string, devices []pluginapi.Device) {
   m.mutex.Lock()
   m.healthyDevices[resourceName] = sets.NewString()
   m.unhealthyDevices[resourceName] = sets.NewString()
   for _, dev := range devices {
      if dev.Health == pluginapi.Healthy {
         m.healthyDevices[resourceName].Insert(dev.ID)
      } else {
         m.unhealthyDevices[resourceName].Insert(dev.ID)
      }
   }
   m.mutex.Unlock()
   m.writeCheckpoint()
}

这里就看到在kubelet.ContainerManager.deviceManager中保存了设备ID，数据结构是map[string]sets.String

allocate

kubernetes\pkg\kubelet\cm\devicemanager\endpoint.go

func (e *endpointImpl) allocate(devs []string) (*pluginapi.AllocateResponse, error) {
   return e.client.Allocate(context.Background(), &pluginapi.AllocateRequest{
      ContainerRequests: []*pluginapi.ContainerAllocateRequest{
         {DevicesIDs: devs},
      },
   })
}

这里就直接发了gRPC请求，看下函数调用处是怎么选择设备ID的。

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

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
    
   for k, v := range container.Resources.Limits {
      resource := string(k)
      needed := int(v.Value())

      allocDevices, err := m.devicesToAllocate(podUID, contName, resource, needed, devicesToReuse[resource])

      startRPCTime := time.Now()
      m.mutex.Lock()
      e, ok := m.endpoints[resource]
      m.mutex.Unlock()

      devs := allocDevices.UnsortedList()
      resp, err := e.allocate(devs)
      
      // 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()
}

1. 通过devicesToAllocate方法获得分配的设备ID
2. 调用allocate方法，获取响应env
3. 更新devicemanager.podDevices数据

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
   devices = sets.NewString()
   
   devicesInUse := m.allocatedDevices[resource]
   available := m.healthyDevices[resource].Difference(devicesInUse)

   allocated := available.UnsortedList()[:needed]

   for _, device := range allocated {
      m.allocatedDevices[resource].Insert(device)
      devices.Insert(device)
   }
   return devices, nil
}

分配资源逻辑

1. 获取容器已分配资源
2. 从cache中获取已使用的设备
3. 比较全部设备与已用设备，得到可用设备
4. 随机从可用设备选出设备ID
5. 更新已用设备cache
6. 返回取得的设备ID

这里就一切真相大白了，kubelet是随机去GPU挂载的。

保存资源分配情况

kubernetes\pkg\kubelet\cm\devicemanager\pod_devices.go

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,
   }
}

这里就保存了每个Pod下每个contrainer的每种资源的使用情况。

// Returns combined container runtime settings to consume the container's allocated devices.
func (pdev podDevices) deviceRunContainerOptions(podUID, contName string) *DeviceRunContainerOptions {}

deviceRunContainerOptions方法返回了创建容器所需的设备信息配置参数。

ps. 一般来说信息不会存两份，所以资源分配情况应该只存在于devicemanager中；只有在需要的时候，返回对应的配置文件就好。