ac29a22a-5e08-4090-9245-9f6e04fbdb07

Committed 24 Nov 2025 10:46PM UTC coverage: 70.499% (+0.01%) from 70.489%

Build # ac29a22a-5e08-4090-9245-9f6e04fbdb07

Build Type

push

prow

Committed by

web-flow

Commit Message

Merge pull request #16157 from fossedihelm/centralize-node-informer

handler: Avoid node GET requests

Run Details

35 of 49 new or added lines in 4 files covered. (71.43%)

2 existing lines in 1 file now uncovered.

70369 of 99815 relevant lines covered (70.5%)

511.1 hits per line

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

83.66

/pkg/virt-handler/device-manager/device_controller.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 (
        "fmt"
        "math"
        "os"
        "strings"
        "sync"
        "time"

        k8sv1 "k8s.io/api/core/v1"
        "k8s.io/client-go/tools/cache"

        "kubevirt.io/client-go/log"

        "kubevirt.io/kubevirt/pkg/storage/reservation"
        virtconfig "kubevirt.io/kubevirt/pkg/virt-config"
        "kubevirt.io/kubevirt/pkg/virt-handler/selinux"
)

var defaultBackoffTime = []time.Duration{1 * time.Second, 2 * time.Second, 5 * time.Second, 10 * time.Second}

type controlledDevice struct {
        devicePlugin Device
        started      bool
        stopChan     chan struct{}
        backoff      []time.Duration
}

func (c *controlledDevice) Start() {
        if c.started {
                return
        }

        stop := make(chan struct{})

        logger := log.DefaultLogger()
        dev := c.devicePlugin
        deviceName := dev.GetDeviceName()
        logger.Infof("Starting a device plugin for device: %s", deviceName)
        retries := 0

        backoff := c.backoff
        if backoff == nil {
                backoff = defaultBackoffTime
        }

        go func() {
                for {
                        err := dev.Start(stop)
                        if err != nil {
                                logger.Reason(err).Errorf("Error starting %s device plugin", deviceName)
                                retries = int(math.Min(float64(retries+1), float64(len(backoff)-1)))
                        } else {
                                retries = 0
                        }

                        select {
                        case <-stop:
                                // Ok we don't want to re-register
                                return
                        case <-time.After(backoff[retries]):
                                // Wait a little and re-register
                                continue
                        }
                }
        }()

        c.stopChan = stop
        c.started = true
}

func (c *controlledDevice) Stop() {
        if !c.started {
                return
        }
        close(c.stopChan)

        c.stopChan = nil
        c.started = false
}

func (c *controlledDevice) GetName() string {
        return c.devicePlugin.GetDeviceName()
}

func PermanentHostDevicePlugins(maxDevices int, permissions string) []Device {
        var permanentDevicePluginPaths = map[string]string{
                "kvm":       "/dev/kvm",
                "tun":       "/dev/net/tun",
                "vhost-net": "/dev/vhost-net",
        }

        ret := make([]Device, 0, len(permanentDevicePluginPaths))
        for name, path := range permanentDevicePluginPaths {
                ret = append(ret, NewGenericDevicePlugin(name, path, maxDevices, permissions, name != "kvm"))
        }
        return ret
}

type DeviceControllerInterface interface {
        Initialized() bool
        RefreshMediatedDeviceTypes()
}

type DeviceController struct {
        permanentPlugins    map[string]Device
        startedPlugins      map[string]controlledDevice
        startedPluginsMutex sync.Mutex
        host                string
        maxDevices          int
        permissions         string
        backoff             []time.Duration
        virtConfig          *virtconfig.ClusterConfig
        stop                chan struct{}
        mdevTypesManager    *MDEVTypesManager
        nodeStore           cache.Store
        mdevRefreshWG       *sync.WaitGroup
}

func NewDeviceController(
        host string,
        maxDevices int,
        permissions string,
        permanentPlugins []Device,
        clusterConfig *virtconfig.ClusterConfig,
        nodeStore cache.Store,
) *DeviceController {
        permanentPluginsMap := make(map[string]Device, len(permanentPlugins))
        for i := range permanentPlugins {
                permanentPluginsMap[permanentPlugins[i].GetDeviceName()] = permanentPlugins[i]
        }

        controller := &DeviceController{
                permanentPlugins: permanentPluginsMap,
                startedPlugins:   map[string]controlledDevice{},
                host:             host,
                maxDevices:       maxDevices,
                permissions:      permissions,
                backoff:          defaultBackoffTime,
                virtConfig:       clusterConfig,
                mdevTypesManager: NewMDEVTypesManager(),
                nodeStore:        nodeStore,
                mdevRefreshWG:    &sync.WaitGroup{},
        }

        return controller
}

func (c *DeviceController) NodeHasDevice(devicePath string) bool {
        _, err := os.Stat(devicePath)
        // Since this is a boolean question, any error means "no"
        return err == nil
}

// updatePermittedHostDevicePlugins returns a slice of device plugins for permitted devices which are present on the node
func (c *DeviceController) updatePermittedHostDevicePlugins() []Device {
        var permittedDevices []Device

        var featureGatedDevices = []struct {
                Name      string
                Path      string
                IsAllowed func() bool
        }{
                {"sev", "/dev/sev", c.virtConfig.WorkloadEncryptionSEVEnabled},
                {"vhost-vsock", "/dev/vhost-vsock", c.virtConfig.VSOCKEnabled},
        }
        for _, dev := range featureGatedDevices {
                if dev.IsAllowed() {
                        permittedDevices = append(
                                permittedDevices,
                                NewGenericDevicePlugin(dev.Name, dev.Path, c.maxDevices, c.permissions, true),
                        )
                }
        }

        if c.virtConfig.PersistentReservationEnabled() {
                d, err := NewSocketDevicePlugin(reservation.GetPrResourceName(), reservation.GetPrHelperSocketDir(), reservation.GetPrHelperSocket(), c.maxDevices, selinux.SELinuxExecutor{}, NewPermissionManager())
                if err != nil {
                        log.Log.Reason(err).Errorf("failed to configure the desired mdev types, failed to get node details")
                } else {
                        permittedDevices = append(permittedDevices, d)
                }
        }

        hostDevs := c.virtConfig.GetPermittedHostDevices()
        if hostDevs == nil {
                return permittedDevices
        }

        if len(hostDevs.PciHostDevices) != 0 {
                supportedPCIDeviceMap := make(map[string]string)
                for _, pciDev := range hostDevs.PciHostDevices {
                        log.Log.V(4).Infof("Permitted PCI device in the cluster, ID: %s, resourceName: %s, externalProvider: %t",
                                strings.ToLower(pciDev.PCIVendorSelector),
                                pciDev.ResourceName,
                                pciDev.ExternalResourceProvider)
                        // do not add a device plugin for this resource if it's being provided via an external device plugin
                        if !pciDev.ExternalResourceProvider {
                                supportedPCIDeviceMap[strings.ToLower(pciDev.PCIVendorSelector)] = pciDev.ResourceName
                        }
                }
                for pciResourceName, pciDevices := range discoverPermittedHostPCIDevices(supportedPCIDeviceMap) {
                        log.Log.V(4).Infof("Discovered PCIs %d devices on the node for the resource: %s", len(pciDevices), pciResourceName)
                        // add a device plugin only for new devices
                        permittedDevices = append(permittedDevices, NewPCIDevicePlugin(pciDevices, pciResourceName))
                }
        }
        if len(hostDevs.MediatedDevices) != 0 {
                supportedMdevsMap := make(map[string]string)
                for _, supportedMdev := range hostDevs.MediatedDevices {
                        log.Log.V(4).Infof("Permitted mediated device in the cluster, ID: %s, resourceName: %s",
                                supportedMdev.MDEVNameSelector,
                                supportedMdev.ResourceName)
                        // do not add a device plugin for this resource if it's being provided via an external device plugin
                        if !supportedMdev.ExternalResourceProvider {
                                selector := removeSelectorSpaces(supportedMdev.MDEVNameSelector)
                                supportedMdevsMap[selector] = supportedMdev.ResourceName
                        }
                }
                for mdevTypeName, mdevUUIDs := range discoverPermittedHostMediatedDevices(supportedMdevsMap) {
                        mdevResourceName := supportedMdevsMap[mdevTypeName]
                        log.Log.V(4).Infof("Discovered mediated device on the node, type: %s, resourceName: %s", mdevTypeName, mdevResourceName)

                        permittedDevices = append(permittedDevices, NewMediatedDevicePlugin(mdevUUIDs, mdevResourceName))
                }
        }

        for resourceName, pluginDevices := range discoverAllowedUSBDevices(hostDevs.USB) {
                permittedDevices = append(permittedDevices, NewUSBDevicePlugin(resourceName, pluginDevices))
        }

        return permittedDevices
}

func removeSelectorSpaces(selectorName string) string {
        // The name usually contain spaces which should be replaced with _
        // Such as GRID T4-1Q
        typeNameStr := strings.Replace(selectorName, " ", "_", -1)
        typeNameStr = strings.TrimSpace(typeNameStr)
        return typeNameStr
}

func (c *DeviceController) splitPermittedDevices(devices []Device) (map[string]Device, map[string]struct{}) {
        devicePluginsToRun := make(map[string]Device)
        devicePluginsToStop := make(map[string]struct{})

        // generate a map of currently started device plugins
        for resourceName := range c.startedPlugins {
                _, isPermanent := c.permanentPlugins[resourceName]
                if !isPermanent {
                        devicePluginsToStop[resourceName] = struct{}{}
                }
        }

        for _, device := range devices {
                if _, isRunning := c.startedPlugins[device.GetDeviceName()]; !isRunning {
                        devicePluginsToRun[device.GetDeviceName()] = device
                } else {
                        delete(devicePluginsToStop, device.GetDeviceName())
                }
        }

        return devicePluginsToRun, devicePluginsToStop
}

func (c *DeviceController) RefreshMediatedDeviceTypes() {
        go func() {
                if c.refreshMediatedDeviceTypes() {
                        c.refreshPermittedDevices()
                }
        }()
}

func (c *DeviceController) getExternallyProvidedMdevs() map[string]struct{} {
        externalMdevResourcesMap := make(map[string]struct{})
        if hostDevs := c.virtConfig.GetPermittedHostDevices(); hostDevs != nil {
                for _, supportedMdev := range hostDevs.MediatedDevices {
                        if supportedMdev.ExternalResourceProvider {
                                selector := removeSelectorSpaces(supportedMdev.MDEVNameSelector)
                                externalMdevResourcesMap[selector] = struct{}{}
                        }
                }
        }
        return externalMdevResourcesMap
}

func (c *DeviceController) refreshMediatedDeviceTypes() bool {
        // the handling of mediated device is disabled
        if c.virtConfig.MediatedDevicesHandlingDisabled() {
                return false
        }

        node, err := c.getNode()
        if err != nil {
                log.Log.Reason(err).Errorf("failed to configure the desired mdev types, failed to get node details")
                return false
        }
        externallyProvidedMdevMap := c.getExternallyProvidedMdevs()

        nodeDesiredMdevTypesList := c.virtConfig.GetDesiredMDEVTypes(node)
        requiresDevicePluginsUpdate, err := c.mdevTypesManager.updateMDEVTypesConfiguration(nodeDesiredMdevTypesList, externallyProvidedMdevMap)
        if err != nil {
                log.Log.Reason(err).Errorf("failed to configure the desired mdev types: %s", strings.Join(nodeDesiredMdevTypesList, ", "))
        }
        return requiresDevicePluginsUpdate
}

func (c *DeviceController) getNode() (*k8sv1.Node, error) {
        nodeObj, exists, err := c.nodeStore.GetByKey(c.host)
        if err != nil {
                log.DefaultLogger().Errorf("Unable to get node: %s", err.Error())
                return nil, err
        }
        if !exists {
                log.DefaultLogger().Errorf("node %s does not exist", c.host)
                return nil, fmt.Errorf("node %s does not exist", c.host)
        }

        node, ok := nodeObj.(*k8sv1.Node)
        if !ok {
                return nil, fmt.Errorf("unknown object type found in node informer")
        }

        return node, nil
}

func (c *DeviceController) refreshPermittedDevices() {
        c.mdevRefreshWG.Add(1)
        logger := log.DefaultLogger()
        var debugDevAdded []string
        var debugDevRemoved []string

        // This function can be called multiple times in parallel, either because of multiple
        //   informer callbacks for the same event, or because the configmap was quickly updated
        //   multiple times in a row. To avoid starting/stopping device plugins multiple times,
        //   we need to protect c.startedPlugins, which we read from in
        //   c.updatePermittedHostDevicePlugins() and write to below.
        c.startedPluginsMutex.Lock()
        defer c.startedPluginsMutex.Unlock()

        enabledDevicePlugins, disabledDevicePlugins := c.splitPermittedDevices(
                c.updatePermittedHostDevicePlugins(),
        )

        // start device plugin for newly permitted devices
        for resourceName, dev := range enabledDevicePlugins {
                c.startDevice(resourceName, dev)
                debugDevAdded = append(debugDevAdded, resourceName)
        }
        // remove device plugin for now forbidden devices
        for resourceName := range disabledDevicePlugins {
                c.stopDevice(resourceName)
                debugDevRemoved = append(debugDevRemoved, resourceName)
        }

        logger.V(3).Info("refreshed device plugins for permitted/forbidden host devices")
        if len(debugDevAdded) > 0 {
                logger.Infof("enabled device-plugins for: %v", debugDevAdded)
        }
        if len(debugDevRemoved) > 0 {
                logger.Infof("disabled device-plugins for: %v", debugDevRemoved)
        }
        c.mdevRefreshWG.Done()
}

func (c *DeviceController) startDevice(resourceName string, dev Device) {
        c.stopDevice(resourceName)
        controlledDev := controlledDevice{
                devicePlugin: dev,
                backoff:      c.backoff,
        }
        controlledDev.Start()
        c.startedPlugins[resourceName] = controlledDev
}

func (c *DeviceController) stopDevice(resourceName string) {
        dev, exists := c.startedPlugins[resourceName]
        if exists {
                dev.Stop()
                delete(c.startedPlugins, resourceName)
        }
}

func (c *DeviceController) Run(stop chan struct{}) {
        logger := log.DefaultLogger()

        // start the permanent DevicePlugins
        func() {
                c.startedPluginsMutex.Lock()
                defer c.startedPluginsMutex.Unlock()
                for name, dev := range c.permanentPlugins {
                        c.startDevice(name, dev)
                }
        }()

        refreshMediatedDeviceTypesFn := func() {
                c.refreshMediatedDeviceTypes()
        }
        c.virtConfig.SetConfigModifiedCallback(refreshMediatedDeviceTypesFn)
        c.virtConfig.SetConfigModifiedCallback(c.refreshPermittedDevices)
        c.refreshPermittedDevices()

        // keep running until stop
        <-stop

        // stop all device plugins
        func() {
                c.startedPluginsMutex.Lock()
                defer c.startedPluginsMutex.Unlock()
                for name := range c.startedPlugins {
                        c.stopDevice(name)
                }
        }()

        // wait for any concurrent mdev refreshes to finish
        c.mdevRefreshWG.Wait()

        logger.Info("Shutting down device plugin controller")
}

func (c *DeviceController) Initialized() bool {
        c.startedPluginsMutex.Lock()
        defer c.startedPluginsMutex.Unlock()
        for _, dev := range c.startedPlugins {
                if !dev.devicePlugin.GetInitialized() {
                        return false
                }
        }

        return true
}

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	"fmt"
24	"math"
25	"os"
26	"strings"
27	"sync"
28	"time"
29
30	k8sv1 "k8s.io/api/core/v1"
31	"k8s.io/client-go/tools/cache"
32
33	"kubevirt.io/client-go/log"
34
35	"kubevirt.io/kubevirt/pkg/storage/reservation"
36	virtconfig "kubevirt.io/kubevirt/pkg/virt-config"
37	"kubevirt.io/kubevirt/pkg/virt-handler/selinux"
38	)
39
40	var defaultBackoffTime = []time.Duration{1 * time.Second, 2 * time.Second, 5 * time.Second, 10 * time.Second}
41
42	type controlledDevice struct {
43	devicePlugin Device
44	started bool
45	stopChan chan struct{}
46	backoff []time.Duration
47	}
48
49	func (c *controlledDevice) Start() {	8✔
50	if c.started {	8✔
51	return	×
52	}	×
53
54	stop := make(chan struct{})	8✔
55		8✔
56	logger := log.DefaultLogger()	8✔
57	dev := c.devicePlugin	8✔
58	deviceName := dev.GetDeviceName()	8✔
59	logger.Infof("Starting a device plugin for device: %s", deviceName)	8✔
60	retries := 0	8✔
61		8✔
62	backoff := c.backoff	8✔
63	if backoff == nil {	8✔
64	backoff = defaultBackoffTime	×
65	}	×
66
67	go func() {	16✔
68	for {	17✔
69	err := dev.Start(stop)	9✔
70	if err != nil {	11✔
71	logger.Reason(err).Errorf("Error starting %s device plugin", deviceName)	2✔
72	retries = int(math.Min(float64(retries+1), float64(len(backoff)-1)))	2✔
73	} else {	9✔
74	retries = 0	7✔
75	}	7✔
76
77	select {	9✔
78	case <-stop:	8✔
79	// Ok we don't want to re-register	8✔
80	return	8✔
81	case <-time.After(backoff[retries]):	1✔
82	// Wait a little and re-register	1✔
83	continue	1✔
84	}
85	}
86	}()
87
88	c.stopChan = stop	8✔
89	c.started = true	8✔
90	}
91
92	func (c *controlledDevice) Stop() {	8✔
93	if !c.started {	8✔
94	return	×
95	}	×
96	close(c.stopChan)	8✔
97		8✔
98	c.stopChan = nil	8✔
99	c.started = false	8✔
100	}
101
102	func (c *controlledDevice) GetName() string {	×
103	return c.devicePlugin.GetDeviceName()	×
104	}	×
105
106	func PermanentHostDevicePlugins(maxDevices int, permissions string) []Device {	114✔
107	var permanentDevicePluginPaths = map[string]string{	114✔
108	"kvm": "/dev/kvm",	114✔
109	"tun": "/dev/net/tun",	114✔
110	"vhost-net": "/dev/vhost-net",	114✔
111	}	114✔
112		114✔
113	ret := make([]Device, 0, len(permanentDevicePluginPaths))	114✔
114	for name, path := range permanentDevicePluginPaths {	456✔
115	ret = append(ret, NewGenericDevicePlugin(name, path, maxDevices, permissions, name != "kvm"))	342✔
116	}	342✔
117	return ret	114✔
118	}
119
120	type DeviceControllerInterface interface {
121	Initialized() bool
122	RefreshMediatedDeviceTypes()
123	}
124
125	type DeviceController struct {
126	permanentPlugins map[string]Device
127	startedPlugins map[string]controlledDevice
128	startedPluginsMutex sync.Mutex
129	host string
130	maxDevices int
131	permissions string
132	backoff []time.Duration
133	virtConfig *virtconfig.ClusterConfig
134	stop chan struct{}
135	mdevTypesManager *MDEVTypesManager
136	nodeStore cache.Store
137	mdevRefreshWG *sync.WaitGroup
138	}
139
140	func NewDeviceController(
141	host string,
142	maxDevices int,
143	permissions string,
144	permanentPlugins []Device,
145	clusterConfig *virtconfig.ClusterConfig,
146	nodeStore cache.Store,
147	) *DeviceController {	127✔
148	permanentPluginsMap := make(map[string]Device, len(permanentPlugins))	127✔
149	for i := range permanentPlugins {	475✔
150	permanentPluginsMap[permanentPlugins[i].GetDeviceName()] = permanentPlugins[i]	348✔
151	}	348✔
152
153	controller := &DeviceController{	127✔
154	permanentPlugins: permanentPluginsMap,	127✔
155	startedPlugins: map[string]controlledDevice{},	127✔
156	host: host,	127✔
157	maxDevices: maxDevices,	127✔
158	permissions: permissions,	127✔
159	backoff: defaultBackoffTime,	127✔
160	virtConfig: clusterConfig,	127✔
161	mdevTypesManager: NewMDEVTypesManager(),	127✔
162	nodeStore: nodeStore,	127✔
163	mdevRefreshWG: &sync.WaitGroup{},	127✔
164	}	127✔
165		127✔
166	return controller	127✔
167	}
168
169	func (c *DeviceController) NodeHasDevice(devicePath string) bool {	4✔
170	_, err := os.Stat(devicePath)	4✔
171	// Since this is a boolean question, any error means "no"	4✔
172	return err == nil	4✔
173	}	4✔
174
175	// updatePermittedHostDevicePlugins returns a slice of device plugins for permitted devices which are present on the node
176	func (c *DeviceController) updatePermittedHostDevicePlugins() []Device {	14✔
177	var permittedDevices []Device	14✔
178		14✔
179	var featureGatedDevices = []struct {	14✔
180	Name string	14✔
181	Path string	14✔
182	IsAllowed func() bool	14✔
183	}{	14✔
184	{"sev", "/dev/sev", c.virtConfig.WorkloadEncryptionSEVEnabled},	14✔
185	{"vhost-vsock", "/dev/vhost-vsock", c.virtConfig.VSOCKEnabled},	14✔
186	}	14✔
187	for _, dev := range featureGatedDevices {	42✔
188	if dev.IsAllowed() {	28✔
189	permittedDevices = append(	×
190	permittedDevices,	×
191	NewGenericDevicePlugin(dev.Name, dev.Path, c.maxDevices, c.permissions, true),	×
192	)	×
193	}	×
194	}
195
196	if c.virtConfig.PersistentReservationEnabled() {	14✔
197	d, err := NewSocketDevicePlugin(reservation.GetPrResourceName(), reservation.GetPrHelperSocketDir(), reservation.GetPrHelperSocket(), c.maxDevices, selinux.SELinuxExecutor{}, NewPermissionManager())	×
198	if err != nil {	×
199	log.Log.Reason(err).Errorf("failed to configure the desired mdev types, failed to get node details")	×
200	} else {	×
201	permittedDevices = append(permittedDevices, d)	×
202	}	×
203	}
204
205	hostDevs := c.virtConfig.GetPermittedHostDevices()	14✔
206	if hostDevs == nil {	18✔
207	return permittedDevices	4✔
208	}	4✔
209
210	if len(hostDevs.PciHostDevices) != 0 {	17✔
211	supportedPCIDeviceMap := make(map[string]string)	7✔
212	for _, pciDev := range hostDevs.PciHostDevices {	20✔
213	log.Log.V(4).Infof("Permitted PCI device in the cluster, ID: %s, resourceName: %s, externalProvider: %t",	13✔
214	strings.ToLower(pciDev.PCIVendorSelector),	13✔
215	pciDev.ResourceName,	13✔
216	pciDev.ExternalResourceProvider)	13✔
217	// do not add a device plugin for this resource if it's being provided via an external device plugin	13✔
218	if !pciDev.ExternalResourceProvider {	14✔
219	supportedPCIDeviceMap[strings.ToLower(pciDev.PCIVendorSelector)] = pciDev.ResourceName	1✔
220	}	1✔
221	}
222	for pciResourceName, pciDevices := range discoverPermittedHostPCIDevices(supportedPCIDeviceMap) {	8✔
223	log.Log.V(4).Infof("Discovered PCIs %d devices on the node for the resource: %s", len(pciDevices), pciResourceName)	1✔
224	// add a device plugin only for new devices	1✔
225	permittedDevices = append(permittedDevices, NewPCIDevicePlugin(pciDevices, pciResourceName))	1✔
226	}	1✔
227	}
228	if len(hostDevs.MediatedDevices) != 0 {	11✔
229	supportedMdevsMap := make(map[string]string)	1✔
230	for _, supportedMdev := range hostDevs.MediatedDevices {	2✔
231	log.Log.V(4).Infof("Permitted mediated device in the cluster, ID: %s, resourceName: %s",	1✔
232	supportedMdev.MDEVNameSelector,	1✔
233	supportedMdev.ResourceName)	1✔
234	// do not add a device plugin for this resource if it's being provided via an external device plugin	1✔
235	if !supportedMdev.ExternalResourceProvider {	2✔
236	selector := removeSelectorSpaces(supportedMdev.MDEVNameSelector)	1✔
237	supportedMdevsMap[selector] = supportedMdev.ResourceName	1✔
238	}	1✔
239	}
240	for mdevTypeName, mdevUUIDs := range discoverPermittedHostMediatedDevices(supportedMdevsMap) {	2✔
241	mdevResourceName := supportedMdevsMap[mdevTypeName]	1✔
242	log.Log.V(4).Infof("Discovered mediated device on the node, type: %s, resourceName: %s", mdevTypeName, mdevResourceName)	1✔
243		1✔
244	permittedDevices = append(permittedDevices, NewMediatedDevicePlugin(mdevUUIDs, mdevResourceName))	1✔
245	}	1✔
246	}
247
248	for resourceName, pluginDevices := range discoverAllowedUSBDevices(hostDevs.USB) {	10✔
249	permittedDevices = append(permittedDevices, NewUSBDevicePlugin(resourceName, pluginDevices))	×
250	}	×
251
252	return permittedDevices	10✔
253	}
254
255	func removeSelectorSpaces(selectorName string) string {	5✔
256	// The name usually contain spaces which should be replaced with _	5✔
257	// Such as GRID T4-1Q	5✔
258	typeNameStr := strings.Replace(selectorName, " ", "_", -1)	5✔
259	typeNameStr = strings.TrimSpace(typeNameStr)	5✔
260	return typeNameStr	5✔
261	}	5✔
262
263	func (c *DeviceController) splitPermittedDevices(devices []Device) (map[string]Device, map[string]struct{}) {	14✔
264	devicePluginsToRun := make(map[string]Device)	14✔
265	devicePluginsToStop := make(map[string]struct{})	14✔
266		14✔
267	// generate a map of currently started device plugins	14✔
268	for resourceName := range c.startedPlugins {	30✔
269	_, isPermanent := c.permanentPlugins[resourceName]	16✔
270	if !isPermanent {	20✔
271	devicePluginsToStop[resourceName] = struct{}{}	4✔
272	}	4✔
273	}
274
275	for _, device := range devices {	16✔
276	if _, isRunning := c.startedPlugins[device.GetDeviceName()]; !isRunning {	4✔
277	devicePluginsToRun[device.GetDeviceName()] = device	2✔
278	} else {	2✔
279	delete(devicePluginsToStop, device.GetDeviceName())	×
280	}	×
281	}
282
283	return devicePluginsToRun, devicePluginsToStop	14✔
284	}
285
286	func (c *DeviceController) RefreshMediatedDeviceTypes() {	×
287	go func() {	×
288	if c.refreshMediatedDeviceTypes() {	×
289	c.refreshPermittedDevices()	×
290	}	×
291	}()
292	}
293
294	func (c *DeviceController) getExternallyProvidedMdevs() map[string]struct{} {	11✔
295	externalMdevResourcesMap := make(map[string]struct{})	11✔
296	if hostDevs := c.virtConfig.GetPermittedHostDevices(); hostDevs != nil {	11✔
297	for _, supportedMdev := range hostDevs.MediatedDevices {	×
298	if supportedMdev.ExternalResourceProvider {	×
299	selector := removeSelectorSpaces(supportedMdev.MDEVNameSelector)	×
300	externalMdevResourcesMap[selector] = struct{}{}	×
301	}	×
302	}
303	}
304	return externalMdevResourcesMap	11✔
305	}
306
307	func (c *DeviceController) refreshMediatedDeviceTypes() bool {	21✔
308	// the handling of mediated device is disabled	21✔
309	if c.virtConfig.MediatedDevicesHandlingDisabled() {	21✔
310	return false	×
311	}	×
312
313	node, err := c.getNode()	21✔
314	if err != nil {	31✔
315	log.Log.Reason(err).Errorf("failed to configure the desired mdev types, failed to get node details")	10✔
316	return false	10✔
317	}	10✔
318	externallyProvidedMdevMap := c.getExternallyProvidedMdevs()	11✔
319		11✔
320	nodeDesiredMdevTypesList := c.virtConfig.GetDesiredMDEVTypes(node)	11✔
321	requiresDevicePluginsUpdate, err := c.mdevTypesManager.updateMDEVTypesConfiguration(nodeDesiredMdevTypesList, externallyProvidedMdevMap)	11✔
322	if err != nil {	11✔
323	log.Log.Reason(err).Errorf("failed to configure the desired mdev types: %s", strings.Join(nodeDesiredMdevTypesList, ", "))	×
324	}	×
325	return requiresDevicePluginsUpdate	11✔
326	}
327
328	func (c DeviceController) getNode() (k8sv1.Node, error) {	21✔
329	nodeObj, exists, err := c.nodeStore.GetByKey(c.host)	21✔
330	if err != nil {	21✔
NEW 331	log.DefaultLogger().Errorf("Unable to get node: %s", err.Error())	×
NEW 332	return nil, err	×
NEW 333	}	×
334	if !exists {	31✔
335	log.DefaultLogger().Errorf("node %s does not exist", c.host)	10✔
336	return nil, fmt.Errorf("node %s does not exist", c.host)	10✔
337	}	10✔
338
339	node, ok := nodeObj.(*k8sv1.Node)	11✔
340	if !ok {	11✔
NEW 341	return nil, fmt.Errorf("unknown object type found in node informer")	×
NEW 342	}	×
343
344	return node, nil	11✔
345	}
346
347	func (c *DeviceController) refreshPermittedDevices() {	10✔
348	c.mdevRefreshWG.Add(1)	10✔
349	logger := log.DefaultLogger()	10✔
350	var debugDevAdded []string	10✔
351	var debugDevRemoved []string	10✔
352		10✔
353	// This function can be called multiple times in parallel, either because of multiple	10✔
354	// informer callbacks for the same event, or because the configmap was quickly updated	10✔
355	// multiple times in a row. To avoid starting/stopping device plugins multiple times,	10✔
356	// we need to protect c.startedPlugins, which we read from in	10✔
357	// c.updatePermittedHostDevicePlugins() and write to below.	10✔
358	c.startedPluginsMutex.Lock()	10✔
359	defer c.startedPluginsMutex.Unlock()	10✔
360		10✔
361	enabledDevicePlugins, disabledDevicePlugins := c.splitPermittedDevices(	10✔
362	c.updatePermittedHostDevicePlugins(),	10✔
363	)	10✔
364		10✔
365	// start device plugin for newly permitted devices	10✔
366	for resourceName, dev := range enabledDevicePlugins {	10✔
367	c.startDevice(resourceName, dev)	×
368	debugDevAdded = append(debugDevAdded, resourceName)	×
369	}	×
370	// remove device plugin for now forbidden devices
371	for resourceName := range disabledDevicePlugins {	12✔
372	c.stopDevice(resourceName)	2✔
373	debugDevRemoved = append(debugDevRemoved, resourceName)	2✔
374	}	2✔
375
376	logger.V(3).Info("refreshed device plugins for permitted/forbidden host devices")	10✔
377	if len(debugDevAdded) > 0 {	10✔
378	logger.Infof("enabled device-plugins for: %v", debugDevAdded)	×
379	}	×
380	if len(debugDevRemoved) > 0 {	11✔
381	logger.Infof("disabled device-plugins for: %v", debugDevRemoved)	1✔
382	}	1✔
383	c.mdevRefreshWG.Done()	10✔
384	}
385
386	func (c *DeviceController) startDevice(resourceName string, dev Device) {	8✔
387	c.stopDevice(resourceName)	8✔
388	controlledDev := controlledDevice{	8✔
389	devicePlugin: dev,	8✔
390	backoff: c.backoff,	8✔
391	}	8✔
392	controlledDev.Start()	8✔
393	c.startedPlugins[resourceName] = controlledDev	8✔
394	}	8✔
395
396	func (c *DeviceController) stopDevice(resourceName string) {	16✔
397	dev, exists := c.startedPlugins[resourceName]	16✔
398	if exists {	24✔
399	dev.Stop()	8✔
400	delete(c.startedPlugins, resourceName)	8✔
401	}	8✔
402	}
403
404	func (c *DeviceController) Run(stop chan struct{}) {	5✔
405	logger := log.DefaultLogger()	5✔
406		5✔
407	// start the permanent DevicePlugins	5✔
408	func() {	10✔
409	c.startedPluginsMutex.Lock()	5✔
410	defer c.startedPluginsMutex.Unlock()	5✔
411	for name, dev := range c.permanentPlugins {	11✔
412	c.startDevice(name, dev)	6✔
413	}	6✔
414	}()
415
416	refreshMediatedDeviceTypesFn := func() {	15✔
417	c.refreshMediatedDeviceTypes()	10✔
418	}	10✔
419	c.virtConfig.SetConfigModifiedCallback(refreshMediatedDeviceTypesFn)	5✔
420	c.virtConfig.SetConfigModifiedCallback(c.refreshPermittedDevices)	5✔
421	c.refreshPermittedDevices()	5✔
422		5✔
423	// keep running until stop	5✔
424	<-stop	5✔
425		5✔
426	// stop all device plugins	5✔
427	func() {	10✔
428	c.startedPluginsMutex.Lock()	5✔
429	defer c.startedPluginsMutex.Unlock()	5✔
430	for name := range c.startedPlugins {	11✔
431	c.stopDevice(name)	6✔
432	}	6✔
433	}()
434
435	// wait for any concurrent mdev refreshes to finish
436	c.mdevRefreshWG.Wait()	5✔
437		5✔
438	logger.Info("Shutting down device plugin controller")	5✔
439	}
440
441	func (c *DeviceController) Initialized() bool {	1✔
442	c.startedPluginsMutex.Lock()	1✔
443	defer c.startedPluginsMutex.Unlock()	1✔
444	for _, dev := range c.startedPlugins {	2✔
445	if !dev.devicePlugin.GetInitialized() {	1✔
446	return false	×
447	}	×
448	}
449
450	return true	1✔
451	}

kubevirt / kubevirt / ac29a22a-5e08-4090-9245-9f6e04fbdb07

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