1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

package main

import (
    "net"
    "os"
    "time"
    "golang.org/x/net/context"
    "google.golang.org/grpc"
    pluginapi "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1"
)

// MyDevicePlugin 实现 DevicePluginServer 接口
type MyDevicePlugin struct {
    devices  []*pluginapi.Device
    socket   string
    server   *grpc.Server
}

// ListAndWatch：定期向 kubelet 汇报设备列表
func (p *MyDevicePlugin) ListAndWatch(e *pluginapi.Empty, s pluginapi.DevicePlugin_ListAndWatchServer) error {
    for {
        resp := &pluginapi.ListAndWatchResponse{Devices: p.devices}
        s.Send(resp)
        time.Sleep(10 * time.Second)
    }
}

// Allocate：分配设备给容器
func (p *MyDevicePlugin) Allocate(ctx context.Context, req *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
    var responses []*pluginapi.ContainerAllocateResponse
    for _, containerReq := range req.ContainerRequests {
        resp := &pluginapi.ContainerAllocateResponse{}
        for _, deviceID := range containerReq.DevicesIDs {
            resp.Devices = append(resp.Devices, &pluginapi.DeviceSpec{
                HostPath:      "/dev/" + deviceID,
                ContainerPath: "/dev/" + deviceID,
                Permissions:   "rw",
            })
        }
        responses = append(responses, resp)
    }
    return &pluginapi.AllocateResponse{ContainerResponses: responses}, nil
}

// PreStartContainer：容器启动前勾子
func (p *MyDevicePlugin) PreStartContainer(ctx context.Context, req *pluginapi.PreStartContainerRequest) (*pluginapi.PreStartContainerResponse, error) {
    return &pluginapi.PreStartContainerResponse{}, nil
}

// GetDevicePluginOptions
func (p *MyDevicePlugin) GetDevicePluginOptions(ctx context.Context, e *pluginapi.Empty) (*pluginapi.DevicePluginOptions, error) {
    return &pluginapi.DevicePluginOptions{PreStartRequired: false}, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

func (p *FPGAPlugin) discoverFPGAs() []*pluginapi.Device {
    var devices []*pluginapi.Device
    files, _ := filepath.Glob("/dev/xfpga/*")
    for i, f := range files {
        devices = append(devices, &pluginapi.Device{
            ID:     fmt.Sprintf("fpga-%d-%s", i, filepath.Base(f)),
            Health: pluginapi.Healthy,
        })
    }
    return devices
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

func (p *FPGAPlugin) Allocate(ctx context.Context, req *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
    responses := &pluginapi.AllocateResponse{}
    for _, cr := range req.ContainerRequests {
        resp := &pluginapi.ContainerAllocateResponse{
            Envs: map[string]string{
                "FPGA_DEVICE_IDS": strings.Join(cr.DevicesIDs, ","),
                "FPGA_BITSTREAM":  "/opt/bitstreams/default.aocx",
            },
            Mounts: []*pluginapi.Mount{
                {
                    ContainerPath: "/opt/bitstreams",
                    HostPath:      "/opt/fpga/bitstreams",
                    ReadOnly:      true,
                },
            },
            Devices: []*pluginapi.DeviceSpec{},
        }
        for _, id := range cr.DevicesIDs {
            resp.Devices = append(resp.Devices, &pluginapi.DeviceSpec{
                HostPath:      "/dev/xfpga/" + id,
                ContainerPath: "/dev/xfpga/" + id,
                Permissions:   "rw",
            })
        }
        responses.ContainerResponses = append(responses.ContainerResponses, resp)
    }
    return responses, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

func (p *MyDevicePlugin) register() error {
    conn, err := grpc.Dial("unix:///var/lib/kubelet/device-plugins/kubelet.sock", grpc.WithInsecure())
    if err != nil {
        return err
    }
    client := pluginapi.NewRegistrationClient(conn)
    req := &pluginapi.RegisterRequest{
        Version:      pluginapi.Version,
        Endpoint:     "my-device-plugin.sock",
        ResourceName: "example.com/my-device",
    }
    _, err = client.Register(context.Background(), req)
    return err
}