db9927fa-6ef4-4b9f-b267-16b157867172

Committed 26 Apr 2026 07:37PM UTC coverage: 71.762% (+0.006%) from 71.756%

Build # db9927fa-6ef4-4b9f-b267-16b157867172

Build Type

push

prow

Committed by

web-flow

Commit Message

Merge pull request #17606 from akalenyu/revert-cross-ns-mig-dq

e2e quarantine: revert accidental cross ns migration dequarantine

Coverage Stats

77616 of 108158 relevant lines covered (71.76%)

572.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

19.81

/pkg/virt-handler/device-manager/usb_device.go

/*
 * This file is part of the KubeVirt project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * Copyright The KubeVirt Authors.
 *
 */

package device_manager

import (
        "bufio"
        "context"
        "fmt"
        "net"
        "os"
        "path"
        "path/filepath"
        "strconv"
        "strings"
        "sync"
        "time"

        "github.com/fsnotify/fsnotify"
        "google.golang.org/grpc"
        "k8s.io/apimachinery/pkg/util/rand"
        v1 "kubevirt.io/api/core/v1"
        "kubevirt.io/client-go/log"

        "kubevirt.io/kubevirt/pkg/safepath"
        "kubevirt.io/kubevirt/pkg/util"
        pluginapi "kubevirt.io/kubevirt/pkg/virt-handler/device-manager/deviceplugin/v1beta1"
)

var (
        pathToUSBDevices = "/sys/bus/usb/devices"
)

var discoverLocalUSBDevicesFunc = discoverPluggedUSBDevices

// The sysfs metadata wrapper for the USB devices
type USBDevice struct {
        Name         string
        Manufacturer string
        Vendor       int
        Product      int
        BCD          int
        Bus          int
        DeviceNumber int
        Serial       string
        DevicePath   string
}

// The uniqueness in the system comes from bus and device number but having the vendor:product
// information can help a lot. Not all usb devices provide or export a serial number.
func (dev *USBDevice) GetID() string {
        return fmt.Sprintf("%04x:%04x-%02d:%02d", dev.Vendor, dev.Product, dev.Bus, dev.DeviceNumber)
}

// The actual plugin
type USBDevicePlugin struct {
        *DevicePluginBase
        update  chan struct{}
        devices []*PluginDevices
        logger  *log.FilteredLogger
}

type PluginDevices struct {
        ID        string
        isHealthy bool
        Devices   []*USBDevice
}

func newPluginDevices(resourceName string, index int, usbdevs []*USBDevice) *PluginDevices {
        return &PluginDevices{
                ID:        fmt.Sprintf("%s-%s-%d", resourceName, rand.String(4), index),
                isHealthy: true,
                Devices:   usbdevs,
        }
}

func (plugin *USBDevicePlugin) Start(stop <-chan struct{}) error {
        plugin.stop = stop

        err := plugin.cleanup()
        if err != nil {
                return fmt.Errorf("error on cleanup: %v", err)
        }

        sock, err := net.Listen("unix", plugin.socketPath)
        if err != nil {
                return fmt.Errorf("error creating GRPC server socket: %v", err)
        }

        plugin.server = grpc.NewServer([]grpc.ServerOption{}...)
        defer plugin.stopDevicePlugin()

        pluginapi.RegisterDevicePluginServer(plugin.server, plugin)

        errChan := make(chan error, 2)

        go func() {
                errChan <- plugin.server.Serve(sock)
        }()

        err = waitForGRPCServer(plugin.socketPath, 5*time.Second)
        if err != nil {
                return fmt.Errorf("error starting the GRPC server: %v", err)
        }

        err = plugin.register()
        if err != nil {
                return fmt.Errorf("error registering with device plugin manager: %v", err)
        }

        go func() {
                errChan <- plugin.healthCheck()
        }()

        plugin.setInitialized(true)
        plugin.logger.Infof("%s device plugin started", plugin.resourceName)
        err = <-errChan

        return err
}

func (pd *PluginDevices) toKubeVirtDevicePlugin() *pluginapi.Device {
        healthStr := pluginapi.Healthy
        if !pd.isHealthy {
                healthStr = pluginapi.Unhealthy
        }
        return &pluginapi.Device{
                ID:       pd.ID,
                Health:   healthStr,
                Topology: nil,
        }
}

func (plugin *USBDevicePlugin) FindDevice(pluginDeviceID string) *PluginDevices {
        for _, pd := range plugin.devices {
                if pd.ID == pluginDeviceID {
                        return pd
                }
        }
        return nil
}

func (plugin *USBDevicePlugin) FindDeviceByUSBID(usbID string) *PluginDevices {
        for _, pd := range plugin.devices {
                for _, usb := range pd.Devices {
                        if usb.GetID() == usbID {
                                return pd
                        }
                }
        }
        return nil
}

func (plugin *USBDevicePlugin) setDeviceHealth(usbID string, isHealthy bool) {
        pd := plugin.FindDeviceByUSBID(usbID)
        isDifferent := pd.isHealthy != isHealthy
        pd.isHealthy = isHealthy
        if isDifferent {
                plugin.update <- struct{}{}
        }
}

func (plugin *USBDevicePlugin) devicesToKubeVirtDevicePlugin() []*pluginapi.Device {
        devices := make([]*pluginapi.Device, 0, len(plugin.devices))
        for _, pluginDevices := range plugin.devices {
                devices = append(devices, pluginDevices.toKubeVirtDevicePlugin())
        }
        return devices
}

func (plugin *USBDevicePlugin) GetInitialized() bool {
        plugin.lock.Lock()
        defer plugin.lock.Unlock()
        return plugin.initialized
}

func (plugin *USBDevicePlugin) setInitialized(initialized bool) {
        plugin.lock.Lock()
        plugin.initialized = initialized
        plugin.lock.Unlock()
}

func (plugin *USBDevicePlugin) GetDeviceName() string {
        return plugin.resourceName
}

func (plugin *USBDevicePlugin) stopDevicePlugin() error {
        defer func() {
                select {
                case <-plugin.done:
                        return
                default:
                        close(plugin.done)
                }
        }()

        // Give the device plugin one second to properly deregister
        ticker := time.NewTicker(1 * time.Second)
        defer ticker.Stop()
        select {
        case <-plugin.deregistered:
        case <-ticker.C:
        }

        plugin.server.Stop()
        plugin.setInitialized(false)
        return plugin.cleanup()
}

func (plugin *USBDevicePlugin) healthCheck() error {
        monitoredDevices := make(map[string]string)
        watcher, err := fsnotify.NewWatcher()
        if err != nil {
                return fmt.Errorf("failed to creating a fsnotify watcher: %v", err)
        }
        defer watcher.Close()

        watchedDirs := make(map[string]struct{})
        for _, pd := range plugin.devices {
                for _, usb := range pd.Devices {
                        usbDevicePath := filepath.Join(util.HostRootMount, usb.DevicePath)
                        usbDeviceDirPath := filepath.Dir(usbDevicePath)
                        if _, exists := watchedDirs[usbDeviceDirPath]; !exists {
                                if err := watcher.Add(usbDeviceDirPath); err != nil {
                                        return fmt.Errorf("failed to watch device %s parent directory: %s", usbDevicePath, err)
                                }
                                watchedDirs[usbDeviceDirPath] = struct{}{}
                        }

                        if err := watcher.Add(usbDevicePath); err != nil {
                                return fmt.Errorf("failed to add the device %s to the watcher: %s", usbDevicePath, err)
                        } else if _, err := os.Stat(usbDevicePath); err != nil {
                                return fmt.Errorf("failed to validate device %s: %s", usbDevicePath, err)
                        }
                        monitoredDevices[usbDevicePath] = usb.GetID()
                }
        }

        dirName := filepath.Dir(plugin.socketPath)
        if err := watcher.Add(dirName); err != nil {
                return fmt.Errorf("failed to add the device-plugin kubelet path to the watcher: %v", err)
        } else if _, err = os.Stat(plugin.socketPath); err != nil {
                return fmt.Errorf("failed to stat the device-plugin socket: %v", err)
        }

        for {
                select {
                case <-plugin.stop:
                        return nil
                case err := <-watcher.Errors:
                        plugin.logger.Reason(err).Errorf("error watching devices and device plugin directory")
                case event := <-watcher.Events:
                        plugin.logger.V(2).Infof("health Event: %v", event)
                        if id, exist := monitoredDevices[event.Name]; exist {
                                // Health in this case is if the device path actually exists
                                if event.Op == fsnotify.Create {
                                        plugin.logger.Infof("monitored device %s appeared", plugin.resourceName)
                                        plugin.setDeviceHealth(id, true)
                                } else if (event.Op == fsnotify.Remove) || (event.Op == fsnotify.Rename) {
                                        plugin.logger.Infof("monitored device %s disappeared", plugin.resourceName)
                                        plugin.setDeviceHealth(id, false)
                                }
                        } else if event.Name == plugin.socketPath && event.Op == fsnotify.Remove {
                                plugin.logger.Infof("device socket file for device %s was removed, kubelet probably restarted.", plugin.resourceName)
                                return nil
                        }
                }
        }
}

func (plugin *USBDevicePlugin) register() error {
        conn, err := grpc.Dial(pluginapi.KubeletSocket,
                grpc.WithInsecure(),
                grpc.WithBlock(),
                grpc.WithTimeout(5*time.Second),
                grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
                        return net.DialTimeout("unix", addr, timeout)
                }),
        )
        if err != nil {
                return err
        }
        defer conn.Close()

        client := pluginapi.NewRegistrationClient(conn)
        reqt := &pluginapi.RegisterRequest{
                Version:      pluginapi.Version,
                Endpoint:     path.Base(plugin.socketPath),
                ResourceName: plugin.GetDeviceName(),
        }

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

// Interface to expose Devices: IDs, health and Topology
func (plugin *USBDevicePlugin) ListAndWatch(_ *pluginapi.Empty, lws pluginapi.DevicePlugin_ListAndWatchServer) error {
        sendUpdate := func(devices []*pluginapi.Device) error {
                response := pluginapi.ListAndWatchResponse{
                        Devices: devices,
                }
                err := lws.Send(&response)
                if err != nil {
                        plugin.logger.Reason(err).Warningf("Failed to send device plugin %s",
                                plugin.resourceName)
                }
                return err
        }

        if err := sendUpdate(plugin.devicesToKubeVirtDevicePlugin()); err != nil {
                return err
        }
        done := false
        for !done {
                select {
                case <-plugin.update:
                        if err := sendUpdate(plugin.devicesToKubeVirtDevicePlugin()); err != nil {
                                return err
                        }
                case <-plugin.stop:
                        done = true
                }
        }

        if err := sendUpdate([]*pluginapi.Device{}); err != nil {
                plugin.logger.Reason(err).Warningf("Failed to deregister device plugin %s",
                        plugin.resourceName)
        }
        close(plugin.deregistered)
        return nil
}

// Interface to allocate requested Device, exported by ListAndWatch
func (plugin *USBDevicePlugin) Allocate(_ context.Context, allocRequest *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
        allocResponse := new(pluginapi.AllocateResponse)
        env := make(map[string]string)
        for _, request := range allocRequest.ContainerRequests {
                containerResponse := &pluginapi.ContainerAllocateResponse{}
                for _, id := range request.DevicesIDs {
                        plugin.logger.V(2).Infof("usb device id: %s", id)

                        pluginDevices := plugin.FindDevice(id)
                        if pluginDevices == nil {
                                plugin.logger.V(2).Infof("usb disappeared: %s", id)
                                continue
                        }

                        deviceSpecs := []*pluginapi.DeviceSpec{}
                        for _, dev := range pluginDevices.Devices {
                                spath, err := safepath.JoinAndResolveWithRelativeRoot(util.HostRootMount, dev.DevicePath)
                                if err != nil {
                                        return nil, fmt.Errorf("error opening the socket %s: %v", dev.DevicePath, err)
                                }

                                err = safepath.ChownAtNoFollow(spath, util.NonRootUID, util.NonRootUID)
                                if err != nil {
                                        return nil, fmt.Errorf("error setting the permission the socket %s: %v", dev.DevicePath, err)
                                }

                                // We might have more than one USB device per resource name
                                key := util.ResourceNameToEnvVar(v1.USBResourcePrefix, plugin.resourceName)
                                value := fmt.Sprintf("%d:%d", dev.Bus, dev.DeviceNumber)
                                if previous, exist := env[key]; exist {
                                        env[key] = fmt.Sprintf("%s,%s", previous, value)
                                } else {
                                        env[key] = value
                                }

                                deviceSpecs = append(deviceSpecs, &pluginapi.DeviceSpec{
                                        ContainerPath: dev.DevicePath,
                                        HostPath:      dev.DevicePath,
                                        Permissions:   "mrw",
                                })
                        }
                        containerResponse.Envs = env
                        containerResponse.Devices = append(containerResponse.Devices, deviceSpecs...)
                }
                allocResponse.ContainerResponses = append(allocResponse.ContainerResponses, containerResponse)
        }

        return allocResponse, nil
}

func parseSysUeventFile(path string) *USBDevice {
        // Grab all details we are interested from uevent
        file, err := os.Open(filepath.Join(path, "uevent"))
        if err != nil {
                log.Log.Reason(err).Infof("Unable to access %s/%s", path, "uevent")
                return nil
        }
        defer file.Close()

        u := USBDevice{}
        scanner := bufio.NewScanner(file)
        for scanner.Scan() {
                line := scanner.Text()
                values := strings.Split(line, "=")
                if len(values) != 2 {
                        log.Log.Infof("Skipping %s due not being key=value", line)
                        continue
                }
                switch values[0] {
                case "BUSNUM":
                        val, err := strconv.ParseInt(values[1], 10, 32)
                        if err != nil {
                                return nil
                        }
                        u.Bus = int(val)
                case "DEVNUM":
                        val, err := strconv.ParseInt(values[1], 10, 32)
                        if err != nil {
                                return nil
                        }
                        u.DeviceNumber = int(val)
                case "PRODUCT":
                        products := strings.Split(values[1], "/")
                        if len(products) != 3 {
                                return nil
                        }

                        val, err := strconv.ParseInt(products[0], 16, 32)
                        if err != nil {
                                return nil
                        }
                        u.Vendor = int(val)

                        val, err = strconv.ParseInt(products[1], 16, 32)
                        if err != nil {
                                return nil
                        }
                        u.Product = int(val)

                        val, err = strconv.ParseInt(products[2], 16, 32)
                        if err != nil {
                                return nil
                        }
                        u.BCD = int(val)
                case "DEVNAME":
                        u.DevicePath = filepath.Join("/dev", values[1])
                default:
                        log.Log.V(5).Infof("Skipping unhandled line: %s", line)
                }
        }
        return &u
}

type LocalDevices struct {
        // For quicker indexing, map devices based on vendor string
        devices map[int][]*USBDevice
}

// finds by vendor and product
func (l *LocalDevices) find(vendor, product int) *USBDevice {
        if devices, exist := l.devices[vendor]; exist {
                for _, local := range devices {
                        if local.Product == product {
                                return local
                        }
                }
        }
        return nil
}

// remove all cached elements
func (l *LocalDevices) remove(usbdevs []*USBDevice) {
        for _, dev := range usbdevs {
                devices, exists := l.devices[dev.Vendor]
                if !exists {
                        continue
                }

                for i, usb := range devices {
                        if usb.GetID() == dev.GetID() {
                                devices = append(devices[:i], devices[i+1:]...)
                                break
                        }
                }

                l.devices[dev.Vendor] = devices
                if len(devices) == 0 {
                        delete(l.devices, dev.Vendor)
                }
        }
}

// return a list of USBDevices while removing it from the list of local devices
func (l *LocalDevices) fetch(selectors []v1.USBSelector) ([]*USBDevice, bool) {
        usbdevs := []*USBDevice{}

        // we have to find all devices under this resource name
        for _, selector := range selectors {
                vendor, product, err := parseSelector(&selector)
                if err != nil {
                        log.Log.Reason(err).Warningf("Failed to convert selector: %+v", selector)
                        return nil, false
                }

                local := l.find(vendor, product)
                if local == nil {
                        return nil, false
                }

                usbdevs = append(usbdevs, local)
        }

        // To avoid mapping the same usb device to different k8s plugins
        l.remove(usbdevs)
        return usbdevs, true
}

func discoverPluggedUSBDevices() *LocalDevices {
        usbDevices := make(map[int][]*USBDevice)
        err := filepath.Walk(pathToUSBDevices, func(path string, info os.FileInfo, err error) error {
                if err != nil {
                        return err
                }
                // Ignore named usb controllers
                if strings.HasPrefix(info.Name(), "usb") {
                        return nil
                }
                // We are interested in actual USB devices information that
                // contains idVendor and idProduct. We can skip all others.
                if _, err := os.Stat(filepath.Join(path, "idVendor")); err != nil {
                        return nil
                }

                // Get device information
                if device := parseSysUeventFile(path); device != nil {
                        usbDevices[device.Vendor] = append(usbDevices[device.Vendor], device)
                }
                return nil
        })

        if err != nil {
                log.Log.Reason(err).Error("Failed when walking usb devices tree")
        }
        return &LocalDevices{devices: usbDevices}
}

func parseSelector(s *v1.USBSelector) (int, int, error) {
        val, err := strconv.ParseInt(s.Vendor, 16, 32)
        if err != nil {
                return -1, -1, err
        }
        vendor := int(val)

        val, err = strconv.ParseInt(s.Product, 16, 32)
        if err != nil {
                return -1, -1, err
        }
        product := int(val)

        return vendor, product, nil
}

func discoverAllowedUSBDevices(usbs []v1.USBHostDevice) map[string][]*PluginDevices {
        // The return value: USB Device Plugins found and permitted to be exposed
        plugins := make(map[string][]*PluginDevices)
        // All USB devices found plugged in the Node
        localDevices := discoverLocalUSBDevicesFunc()
        for _, usbConfig := range usbs {
                resourceName := usbConfig.ResourceName
                if usbConfig.ExternalResourceProvider {
                        log.Log.V(6).Infof("Skipping discovery of %s. To be handled by external device-plugin",
                                resourceName)
                        continue
                }
                index := 0
                usbdevs, foundAll := localDevices.fetch(usbConfig.Selectors)
                for foundAll {
                        // Create new USB Device Plugin with found USB Devices for this resource name
                        pluginDevices := newPluginDevices(resourceName, index, usbdevs)
                        plugins[resourceName] = append(plugins[resourceName], pluginDevices)
                        index++
                        usbdevs, foundAll = localDevices.fetch(usbConfig.Selectors)
                }
        }
        return plugins
}

func NewUSBDevicePlugin(resourceName string, pluginDevices []*PluginDevices) *USBDevicePlugin {
        s := strings.Split(resourceName, "/")
        resourceID := s[0]
        if len(s) > 1 {
                resourceID = s[1]
        }
        resourceID = fmt.Sprintf("usb-%s", resourceID)
        usb := &USBDevicePlugin{
                DevicePluginBase: &DevicePluginBase{
                        socketPath:   SocketPath(resourceID),
                        resourceName: resourceName,
                        initialized:  false,
                        lock:         &sync.Mutex{},
                        health:       make(chan deviceHealth),
                        done:         make(chan struct{}),
                        deregistered: make(chan struct{}),
                },
                devices: pluginDevices,
                logger:  log.Log.With("subcomponent", resourceID),
        }
        return usb
}

1	/*
2	* This file is part of the KubeVirt project
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*
16	* Copyright The KubeVirt Authors.
17	*
18	*/
19
20	package device_manager
21
22	import (
23	"bufio"
24	"context"
25	"fmt"
26	"net"
27	"os"
28	"path"
29	"path/filepath"
30	"strconv"
31	"strings"
32	"sync"
33	"time"
34
35	"github.com/fsnotify/fsnotify"
36	"google.golang.org/grpc"
37	"k8s.io/apimachinery/pkg/util/rand"
38	v1 "kubevirt.io/api/core/v1"
39	"kubevirt.io/client-go/log"
40
41	"kubevirt.io/kubevirt/pkg/safepath"
42	"kubevirt.io/kubevirt/pkg/util"
43	pluginapi "kubevirt.io/kubevirt/pkg/virt-handler/device-manager/deviceplugin/v1beta1"
44	)
45
46	var (
47	pathToUSBDevices = "/sys/bus/usb/devices"
48	)
49
50	var discoverLocalUSBDevicesFunc = discoverPluggedUSBDevices
51
52	// The sysfs metadata wrapper for the USB devices
53	type USBDevice struct {
54	Name string
55	Manufacturer string
56	Vendor int
57	Product int
58	BCD int
59	Bus int
60	DeviceNumber int
61	Serial string
62	DevicePath string
63	}
64
65	// The uniqueness in the system comes from bus and device number but having the vendor:product
66	// information can help a lot. Not all usb devices provide or export a serial number.
67	func (dev *USBDevice) GetID() string {	20✔
68	return fmt.Sprintf("%04x:%04x-%02d:%02d", dev.Vendor, dev.Product, dev.Bus, dev.DeviceNumber)	20✔
69	}	20✔
70
71	// The actual plugin
72	type USBDevicePlugin struct {
73	*DevicePluginBase
74	update chan struct{}
75	devices []*PluginDevices
76	logger *log.FilteredLogger
77	}
78
79	type PluginDevices struct {
80	ID string
81	isHealthy bool
82	Devices []*USBDevice
83	}
84
85	func newPluginDevices(resourceName string, index int, usbdevs []USBDevice) PluginDevices {	18✔
86	return &PluginDevices{	18✔
87	ID: fmt.Sprintf("%s-%s-%d", resourceName, rand.String(4), index),	18✔
88	isHealthy: true,	18✔
89	Devices: usbdevs,	18✔
90	}	18✔
91	}	18✔
92
93	func (plugin *USBDevicePlugin) Start(stop <-chan struct{}) error {	×
94	plugin.stop = stop	×
95		×
96	err := plugin.cleanup()	×
97	if err != nil {	×
98	return fmt.Errorf("error on cleanup: %v", err)	×
99	}	×
100
101	sock, err := net.Listen("unix", plugin.socketPath)	×
102	if err != nil {	×
103	return fmt.Errorf("error creating GRPC server socket: %v", err)	×
104	}	×
105
106	plugin.server = grpc.NewServer([]grpc.ServerOption{}...)	×
107	defer plugin.stopDevicePlugin()	×
108		×
109	pluginapi.RegisterDevicePluginServer(plugin.server, plugin)	×
110		×
111	errChan := make(chan error, 2)	×
112		×
113	go func() {	×
114	errChan <- plugin.server.Serve(sock)	×
115	}()	×
116
117	err = waitForGRPCServer(plugin.socketPath, 5*time.Second)	×
118	if err != nil {	×
119	return fmt.Errorf("error starting the GRPC server: %v", err)	×
120	}	×
121
122	err = plugin.register()	×
123	if err != nil {	×
124	return fmt.Errorf("error registering with device plugin manager: %v", err)	×
125	}	×
126
127	go func() {	×
128	errChan <- plugin.healthCheck()	×
129	}()	×
130
131	plugin.setInitialized(true)	×
132	plugin.logger.Infof("%s device plugin started", plugin.resourceName)	×
133	err = <-errChan	×
134		×
135	return err	×
136	}
137
138	func (pd PluginDevices) toKubeVirtDevicePlugin() pluginapi.Device {	×
139	healthStr := pluginapi.Healthy	×
140	if !pd.isHealthy {	×
141	healthStr = pluginapi.Unhealthy	×
142	}	×
143	return &pluginapi.Device{	×
144	ID: pd.ID,	×
145	Health: healthStr,	×
146	Topology: nil,	×
147	}	×
148	}
149
150	func (plugin USBDevicePlugin) FindDevice(pluginDeviceID string) PluginDevices {	×
151	for _, pd := range plugin.devices {	×
152	if pd.ID == pluginDeviceID {	×
153	return pd	×
154	}	×
155	}
156	return nil	×
157	}
158
159	func (plugin USBDevicePlugin) FindDeviceByUSBID(usbID string) PluginDevices {	×
160	for _, pd := range plugin.devices {	×
161	for _, usb := range pd.Devices {	×
162	if usb.GetID() == usbID {	×
163	return pd	×
164	}	×
165	}
166	}
167	return nil	×
168	}
169
170	func (plugin *USBDevicePlugin) setDeviceHealth(usbID string, isHealthy bool) {	×
171	pd := plugin.FindDeviceByUSBID(usbID)	×
172	isDifferent := pd.isHealthy != isHealthy	×
173	pd.isHealthy = isHealthy	×
174	if isDifferent {	×
175	plugin.update <- struct{}{}	×
176	}	×
177	}
178
179	func (plugin USBDevicePlugin) devicesToKubeVirtDevicePlugin() []pluginapi.Device {	×
180	devices := make([]*pluginapi.Device, 0, len(plugin.devices))	×
181	for _, pluginDevices := range plugin.devices {	×
182	devices = append(devices, pluginDevices.toKubeVirtDevicePlugin())	×
183	}	×
184	return devices	×
185	}
186
187	func (plugin *USBDevicePlugin) GetInitialized() bool {	×
188	plugin.lock.Lock()	×
189	defer plugin.lock.Unlock()	×
190	return plugin.initialized	×
191	}	×
192
193	func (plugin *USBDevicePlugin) setInitialized(initialized bool) {	×
194	plugin.lock.Lock()	×
195	plugin.initialized = initialized	×
196	plugin.lock.Unlock()	×
197	}	×
198
199	func (plugin *USBDevicePlugin) GetDeviceName() string {	×
200	return plugin.resourceName	×
201	}	×
202
203	func (plugin *USBDevicePlugin) stopDevicePlugin() error {	×
204	defer func() {	×
205	select {	×
206	case <-plugin.done:	×
207	return	×
208	default:	×
209	close(plugin.done)	×
210	}
211	}()
212
213	// Give the device plugin one second to properly deregister
214	ticker := time.NewTicker(1 * time.Second)	×
215	defer ticker.Stop()	×
216	select {	×
217	case <-plugin.deregistered:	×
218	case <-ticker.C:	×
219	}
220
221	plugin.server.Stop()	×
222	plugin.setInitialized(false)	×
223	return plugin.cleanup()	×
224	}
225
226	func (plugin *USBDevicePlugin) healthCheck() error {	×
227	monitoredDevices := make(map[string]string)	×
228	watcher, err := fsnotify.NewWatcher()	×
229	if err != nil {	×
230	return fmt.Errorf("failed to creating a fsnotify watcher: %v", err)	×
231	}	×
232	defer watcher.Close()	×
233		×
234	watchedDirs := make(map[string]struct{})	×
235	for _, pd := range plugin.devices {	×
236	for _, usb := range pd.Devices {	×
237	usbDevicePath := filepath.Join(util.HostRootMount, usb.DevicePath)	×
238	usbDeviceDirPath := filepath.Dir(usbDevicePath)	×
239	if _, exists := watchedDirs[usbDeviceDirPath]; !exists {	×
240	if err := watcher.Add(usbDeviceDirPath); err != nil {	×
241	return fmt.Errorf("failed to watch device %s parent directory: %s", usbDevicePath, err)	×
242	}	×
243	watchedDirs[usbDeviceDirPath] = struct{}{}	×
244	}
245
246	if err := watcher.Add(usbDevicePath); err != nil {	×
247	return fmt.Errorf("failed to add the device %s to the watcher: %s", usbDevicePath, err)	×
248	} else if _, err := os.Stat(usbDevicePath); err != nil {	×
249	return fmt.Errorf("failed to validate device %s: %s", usbDevicePath, err)	×
250	}	×
251	monitoredDevices[usbDevicePath] = usb.GetID()	×
252	}
253	}
254
255	dirName := filepath.Dir(plugin.socketPath)	×
256	if err := watcher.Add(dirName); err != nil {	×
257	return fmt.Errorf("failed to add the device-plugin kubelet path to the watcher: %v", err)	×
258	} else if _, err = os.Stat(plugin.socketPath); err != nil {	×
259	return fmt.Errorf("failed to stat the device-plugin socket: %v", err)	×
260	}	×
261
262	for {	×
263	select {	×
264	case <-plugin.stop:	×
265	return nil	×
266	case err := <-watcher.Errors:	×
267	plugin.logger.Reason(err).Errorf("error watching devices and device plugin directory")	×
268	case event := <-watcher.Events:	×
269	plugin.logger.V(2).Infof("health Event: %v", event)	×
270	if id, exist := monitoredDevices[event.Name]; exist {	×
271	// Health in this case is if the device path actually exists	×
272	if event.Op == fsnotify.Create {	×
273	plugin.logger.Infof("monitored device %s appeared", plugin.resourceName)	×
274	plugin.setDeviceHealth(id, true)	×
275	} else if (event.Op == fsnotify.Remove) \|\| (event.Op == fsnotify.Rename) {	×
276	plugin.logger.Infof("monitored device %s disappeared", plugin.resourceName)	×
277	plugin.setDeviceHealth(id, false)	×
278	}	×
279	} else if event.Name == plugin.socketPath && event.Op == fsnotify.Remove {	×
280	plugin.logger.Infof("device socket file for device %s was removed, kubelet probably restarted.", plugin.resourceName)	×
281	return nil	×
282	}	×
283	}
284	}
285	}
286
287	func (plugin *USBDevicePlugin) register() error {	×
288	conn, err := grpc.Dial(pluginapi.KubeletSocket,	×
289	grpc.WithInsecure(),	×
290	grpc.WithBlock(),	×
291	grpc.WithTimeout(5*time.Second),	×
292	grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {	×
293	return net.DialTimeout("unix", addr, timeout)	×
294	}),	×
295	)
296	if err != nil {	×
297	return err	×
298	}	×
299	defer conn.Close()	×
300		×
301	client := pluginapi.NewRegistrationClient(conn)	×
302	reqt := &pluginapi.RegisterRequest{	×
303	Version: pluginapi.Version,	×
304	Endpoint: path.Base(plugin.socketPath),	×
305	ResourceName: plugin.GetDeviceName(),	×
306	}	×
307		×
308	_, err = client.Register(context.Background(), reqt)	×
309	if err != nil {	×
310	return err	×
311	}	×
312	return nil	×
313	}
314
315	// Interface to expose Devices: IDs, health and Topology
316	func (plugin USBDevicePlugin) ListAndWatch(_ pluginapi.Empty, lws pluginapi.DevicePlugin_ListAndWatchServer) error {	×
317	sendUpdate := func(devices []*pluginapi.Device) error {	×
318	response := pluginapi.ListAndWatchResponse{	×
319	Devices: devices,	×
320	}	×
321	err := lws.Send(&response)	×
322	if err != nil {	×
323	plugin.logger.Reason(err).Warningf("Failed to send device plugin %s",	×
324	plugin.resourceName)	×
325	}	×
326	return err	×
327	}
328
329	if err := sendUpdate(plugin.devicesToKubeVirtDevicePlugin()); err != nil {	×
330	return err	×
331	}	×
332	done := false	×
333	for !done {	×
334	select {	×
335	case <-plugin.update:	×
336	if err := sendUpdate(plugin.devicesToKubeVirtDevicePlugin()); err != nil {	×
337	return err	×
338	}	×
339	case <-plugin.stop:	×
340	done = true	×
341	}
342	}
343
344	if err := sendUpdate([]*pluginapi.Device{}); err != nil {	×
345	plugin.logger.Reason(err).Warningf("Failed to deregister device plugin %s",	×
346	plugin.resourceName)	×
347	}	×
348	close(plugin.deregistered)	×
349	return nil	×
350	}
351
352	// Interface to allocate requested Device, exported by ListAndWatch
353	func (plugin USBDevicePlugin) Allocate(_ context.Context, allocRequest pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {	×
354	allocResponse := new(pluginapi.AllocateResponse)	×
355	env := make(map[string]string)	×
356	for _, request := range allocRequest.ContainerRequests {	×
357	containerResponse := &pluginapi.ContainerAllocateResponse{}	×
358	for _, id := range request.DevicesIDs {	×
359	plugin.logger.V(2).Infof("usb device id: %s", id)	×
360		×
361	pluginDevices := plugin.FindDevice(id)	×
362	if pluginDevices == nil {	×
363	plugin.logger.V(2).Infof("usb disappeared: %s", id)	×
364	continue	×
365	}
366
367	deviceSpecs := []*pluginapi.DeviceSpec{}	×
368	for _, dev := range pluginDevices.Devices {	×
369	spath, err := safepath.JoinAndResolveWithRelativeRoot(util.HostRootMount, dev.DevicePath)	×
370	if err != nil {	×
371	return nil, fmt.Errorf("error opening the socket %s: %v", dev.DevicePath, err)	×
372	}	×
373
374	err = safepath.ChownAtNoFollow(spath, util.NonRootUID, util.NonRootUID)	×
375	if err != nil {	×
376	return nil, fmt.Errorf("error setting the permission the socket %s: %v", dev.DevicePath, err)	×
377	}	×
378
379	// We might have more than one USB device per resource name
380	key := util.ResourceNameToEnvVar(v1.USBResourcePrefix, plugin.resourceName)	×
381	value := fmt.Sprintf("%d:%d", dev.Bus, dev.DeviceNumber)	×
382	if previous, exist := env[key]; exist {	×
383	env[key] = fmt.Sprintf("%s,%s", previous, value)	×
384	} else {	×
385	env[key] = value	×
386	}	×
387
388	deviceSpecs = append(deviceSpecs, &pluginapi.DeviceSpec{	×
389	ContainerPath: dev.DevicePath,	×
390	HostPath: dev.DevicePath,	×
391	Permissions: "mrw",	×
392	})	×
393	}
394	containerResponse.Envs = env	×
395	containerResponse.Devices = append(containerResponse.Devices, deviceSpecs...)	×
396	}
397	allocResponse.ContainerResponses = append(allocResponse.ContainerResponses, containerResponse)	×
398	}
399
400	return allocResponse, nil	×
401	}
402
403	func parseSysUeventFile(path string) *USBDevice {	×
404	// Grab all details we are interested from uevent	×
405	file, err := os.Open(filepath.Join(path, "uevent"))	×
406	if err != nil {	×
407	log.Log.Reason(err).Infof("Unable to access %s/%s", path, "uevent")	×
408	return nil	×
409	}	×
410	defer file.Close()	×
411		×
412	u := USBDevice{}	×
413	scanner := bufio.NewScanner(file)	×
414	for scanner.Scan() {	×
415	line := scanner.Text()	×
416	values := strings.Split(line, "=")	×
417	if len(values) != 2 {	×
418	log.Log.Infof("Skipping %s due not being key=value", line)	×
419	continue	×
420	}
421	switch values[0] {	×
422	case "BUSNUM":	×
423	val, err := strconv.ParseInt(values[1], 10, 32)	×
424	if err != nil {	×
425	return nil	×
426	}	×
427	u.Bus = int(val)	×
428	case "DEVNUM":	×
429	val, err := strconv.ParseInt(values[1], 10, 32)	×
430	if err != nil {	×
431	return nil	×
432	}	×
433	u.DeviceNumber = int(val)	×
434	case "PRODUCT":	×
435	products := strings.Split(values[1], "/")	×
436	if len(products) != 3 {	×
437	return nil	×
438	}	×
439
440	val, err := strconv.ParseInt(products[0], 16, 32)	×
441	if err != nil {	×
442	return nil	×
443	}	×
444	u.Vendor = int(val)	×
445		×
446	val, err = strconv.ParseInt(products[1], 16, 32)	×
447	if err != nil {	×
448	return nil	×
449	}	×
450	u.Product = int(val)	×
451		×
452	val, err = strconv.ParseInt(products[2], 16, 32)	×
453	if err != nil {	×
454	return nil	×
455	}	×
456	u.BCD = int(val)	×
457	case "DEVNAME":	×
458	u.DevicePath = filepath.Join("/dev", values[1])	×
459	default:	×
460	log.Log.V(5).Infof("Skipping unhandled line: %s", line)	×
461	}
462	}
463	return &u	×
464	}
465
466	type LocalDevices struct {
467	// For quicker indexing, map devices based on vendor string
468	devices map[int][]*USBDevice
469	}
470
471	// finds by vendor and product
472	func (l LocalDevices) find(vendor, product int) USBDevice {	18✔
473	if devices, exist := l.devices[vendor]; exist {	28✔
474	for _, local := range devices {	20✔
475	if local.Product == product {	20✔
476	return local	10✔
477	}	10✔
478	}
479	}
480	return nil	8✔
481	}
482
483	// remove all cached elements
484	func (l LocalDevices) remove(usbdevs []USBDevice) {	9✔
485	for _, dev := range usbdevs {	19✔
486	devices, exists := l.devices[dev.Vendor]	10✔
487	if !exists {	10✔
488	continue	×
489	}
490
491	for i, usb := range devices {	20✔
492	if usb.GetID() == dev.GetID() {	20✔
493	devices = append(devices[:i], devices[i+1:]...)	10✔
494	break	10✔
495	}
496	}
497
498	l.devices[dev.Vendor] = devices	10✔
499	if len(devices) == 0 {	17✔
500	delete(l.devices, dev.Vendor)	7✔
501	}	7✔
502	}
503	}
504
505	// return a list of USBDevices while removing it from the list of local devices
506	func (l LocalDevices) fetch(selectors []v1.USBSelector) ([]USBDevice, bool) {	17✔
507	usbdevs := []*USBDevice{}	17✔
508		17✔
509	// we have to find all devices under this resource name	17✔
510	for _, selector := range selectors {	35✔
511	vendor, product, err := parseSelector(&selector)	18✔
512	if err != nil {	18✔
513	log.Log.Reason(err).Warningf("Failed to convert selector: %+v", selector)	×
514	return nil, false	×
515	}	×
516
517	local := l.find(vendor, product)	18✔
518	if local == nil {	26✔
519	return nil, false	8✔
520	}	8✔
521
522	usbdevs = append(usbdevs, local)	10✔
523	}
524
525	// To avoid mapping the same usb device to different k8s plugins
526	l.remove(usbdevs)	9✔
527	return usbdevs, true	9✔
528	}
529
530	func discoverPluggedUSBDevices() *LocalDevices {	1✔
531	usbDevices := make(map[int][]*USBDevice)	1✔
532	err := filepath.Walk(pathToUSBDevices, func(path string, info os.FileInfo, err error) error {	2✔
533	if err != nil {	2✔
534	return err	1✔
535	}	1✔
536	// Ignore named usb controllers
537	if strings.HasPrefix(info.Name(), "usb") {	×
538	return nil	×
539	}	×
540	// We are interested in actual USB devices information that
541	// contains idVendor and idProduct. We can skip all others.
542	if _, err := os.Stat(filepath.Join(path, "idVendor")); err != nil {	×
543	return nil	×
544	}	×
545
546	// Get device information
547	if device := parseSysUeventFile(path); device != nil {	×
548	usbDevices[device.Vendor] = append(usbDevices[device.Vendor], device)	×
549	}	×
550	return nil	×
551	})
552
553	if err != nil {	2✔
554	log.Log.Reason(err).Error("Failed when walking usb devices tree")	1✔
555	}	1✔
556	return &LocalDevices{devices: usbDevices}	1✔
557	}
558
559	func parseSelector(s *v1.USBSelector) (int, int, error) {	18✔
560	val, err := strconv.ParseInt(s.Vendor, 16, 32)	18✔
561	if err != nil {	18✔
562	return -1, -1, err	×
563	}	×
564	vendor := int(val)	18✔
565		18✔
566	val, err = strconv.ParseInt(s.Product, 16, 32)	18✔
567	if err != nil {	18✔
568	return -1, -1, err	×
569	}	×
570	product := int(val)	18✔
571		18✔
572	return vendor, product, nil	18✔
573	}
574
575	func discoverAllowedUSBDevices(usbs []v1.USBHostDevice) map[string][]*PluginDevices {	16✔
576	// The return value: USB Device Plugins found and permitted to be exposed	16✔
577	plugins := make(map[string][]*PluginDevices)	16✔
578	// All USB devices found plugged in the Node	16✔
579	localDevices := discoverLocalUSBDevicesFunc()	16✔
580	for _, usbConfig := range usbs {	25✔
581	resourceName := usbConfig.ResourceName	9✔
582	if usbConfig.ExternalResourceProvider {	10✔
583	log.Log.V(6).Infof("Skipping discovery of %s. To be handled by external device-plugin",	1✔
584	resourceName)	1✔
585	continue	1✔
586	}
587	index := 0	8✔
588	usbdevs, foundAll := localDevices.fetch(usbConfig.Selectors)	8✔
589	for foundAll {	17✔
590	// Create new USB Device Plugin with found USB Devices for this resource name	9✔
591	pluginDevices := newPluginDevices(resourceName, index, usbdevs)	9✔
592	plugins[resourceName] = append(plugins[resourceName], pluginDevices)	9✔
593	index++	9✔
594	usbdevs, foundAll = localDevices.fetch(usbConfig.Selectors)	9✔
595	}	9✔
596	}
597	return plugins	16✔
598	}
599
600	func NewUSBDevicePlugin(resourceName string, pluginDevices []PluginDevices) USBDevicePlugin {	×
601	s := strings.Split(resourceName, "/")	×
602	resourceID := s[0]	×
603	if len(s) > 1 {	×
604	resourceID = s[1]	×
605	}	×
606	resourceID = fmt.Sprintf("usb-%s", resourceID)	×
607	usb := &USBDevicePlugin{	×
608	DevicePluginBase: &DevicePluginBase{	×
609	socketPath: SocketPath(resourceID),	×
610	resourceName: resourceName,	×
611	initialized: false,	×
612	lock: &sync.Mutex{},	×
613	health: make(chan deviceHealth),	×
614	done: make(chan struct{}),	×
615	deregistered: make(chan struct{}),	×
616	},	×
617	devices: pluginDevices,	×
618	logger: log.Log.With("subcomponent", resourceID),	×
619	}	×
620	return usb	×
621	}

kubevirt / kubevirt / db9927fa-6ef4-4b9f-b267-16b157867172

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous