4282cb34-bade-47c2-9e61-a1b730ef4465

Committed 15 Jan 2026 01:57PM UTC coverage: 70.825% (+0.002%) from 70.823%

Build # 4282cb34-bade-47c2-9e61-a1b730ef4465

Build Type

push

prow

Committed by

web-flow

Commit Message

Merge pull request #16488 from lyarwood/clone-api-add-volumerestore-policy

Add VolumeNamePolicy support to VirtualMachineClone API

Run Details

8 of 8 new or added lines in 2 files covered. (100.0%)

3 existing lines in 1 file now uncovered.

72055 of 101736 relevant lines covered (70.83%)

580.49 hits per line

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

42.0

/pkg/virt-handler/device-manager/socket_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 (
        "context"
        "errors"
        "fmt"
        "net"
        "os"
        "path"
        "path/filepath"
        "strconv"
        "strings"
        "sync"

        "github.com/fsnotify/fsnotify"
        "google.golang.org/grpc"

        "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"
        "kubevirt.io/kubevirt/pkg/virt-handler/selinux"
)

//go:generate mockgen -source $GOFILE -package=$GOPACKAGE -destination=generated_mock_$GOFILE

type PermissionManager interface {
        ChownAtNoFollow(path *safepath.Path, uid, gid int) error
}

type permissionManager struct{}

func NewPermissionManager() PermissionManager {
        return &permissionManager{}
}

func (p *permissionManager) ChownAtNoFollow(path *safepath.Path, uid, gid int) error {
        return safepath.ChownAtNoFollow(path, uid, gid)
}

type SocketDevicePlugin struct {
        *DevicePluginBase
        socketDir  string
        socket     string
        socketName string
        executor   selinux.Executor
        p          PermissionManager
}

func (dpi *SocketDevicePlugin) Start(stop <-chan struct{}) (err error) {
        logger := log.DefaultLogger()
        dpi.stop = stop

        err = dpi.cleanup()
        if err != nil {
                return err
        }

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

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

        pluginapi.RegisterDevicePluginServer(dpi.server, dpi)

        errChan := make(chan error, 2)

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

        err = waitForGRPCServer(dpi.socketPath, connectionTimeout)
        if err != nil {
                return fmt.Errorf("error starting the GRPC server: %v", err)
        }

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

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

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

        return err
}

func (dpi *SocketDevicePlugin) setSocketPermissions() error {
        prSock, err := safepath.JoinAndResolveWithRelativeRoot("/", dpi.socketDir, dpi.socket)
        if err != nil {
                return fmt.Errorf("error opening the socket %s/%s: %v", dpi.socketDir, dpi.socketName, err)
        }
        err = dpi.p.ChownAtNoFollow(prSock, util.NonRootUID, util.NonRootUID)
        if err != nil {
                return fmt.Errorf("error setting the permission the socket %s/%s:%v", dpi.socketDir, dpi.socketName, err)
        }
        if se, exists, err := dpi.executor.NewSELinux(); err == nil && exists {
                if err := selinux.RelabelFilesUnprivileged(se.IsPermissive(), prSock); err != nil {
                        return fmt.Errorf("error relabeling required files: %v", err)
                }
        } else if err != nil {
                return fmt.Errorf("failed to detect the presence of selinux: %v", err)
        }

        return nil
}

func (dpi *SocketDevicePlugin) setSocketDirectoryPermissions() error {
        dir, err := safepath.JoinAndResolveWithRelativeRoot("/", dpi.socketDir)
        if err != nil {
                return fmt.Errorf("error opening the socket dir %s: %v", dpi.socket, err)
        }
        err = dpi.p.ChownAtNoFollow(dir, util.NonRootUID, util.NonRootUID)
        if err != nil {
                return fmt.Errorf("error setting the permission the socket dir %s: %v", dpi.socketDir, err)
        }
        if se, exists, err := dpi.executor.NewSELinux(); err == nil && exists {
                if err := selinux.RelabelFilesUnprivileged(se.IsPermissive(), dir); err != nil {
                        return fmt.Errorf("error relabeling required files: %v", err)
                }
        } else if err != nil {
                return fmt.Errorf("failed to detect the presence of selinux: %v", err)
        }

        return nil
}

func NewSocketDevicePlugin(socketName, socketDir, socket string, maxDevices int, executor selinux.Executor, p PermissionManager) (*SocketDevicePlugin, error) {
        dpi := &SocketDevicePlugin{
                DevicePluginBase: &DevicePluginBase{
                        health:       make(chan deviceHealth),
                        resourceName: fmt.Sprintf("%s/%s", DeviceNamespace, socketName),
                        initialized:  false,
                        lock:         &sync.Mutex{},
                        done:         make(chan struct{}),
                        deregistered: make(chan struct{}),
                        socketPath:   SocketPath(strings.Replace(socketName, "/", "-", -1)),
                },
                socket:     socket,
                socketDir:  socketDir,
                socketName: socketName,
                executor:   executor,
                p:          p,
        }

        for i := 0; i < maxDevices; i++ {
                deviceId := dpi.socketName + strconv.Itoa(i)
                dpi.devs = append(dpi.devs, &pluginapi.Device{
                        ID:     deviceId,
                        Health: pluginapi.Healthy,
                })
        }
        if err := dpi.setSocketDirectoryPermissions(); err != nil {
                return nil, err
        }
        if err := dpi.setSocketPermissions(); err != nil {
                return nil, err
        }

        return dpi, nil
}

// Register registers the device plugin for the given resourceName with Kubelet.
func (dpi *SocketDevicePlugin) register() error {
        conn, err := gRPCConnect(pluginapi.KubeletSocket, connectionTimeout)
        if err != nil {
                return err
        }
        defer conn.Close()

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

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

func (dpi *SocketDevicePlugin) Allocate(ctx context.Context, r *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
        log.DefaultLogger().Infof("Socket Allocate: resourceName: %s", dpi.socketName)
        log.DefaultLogger().Infof("Socket Allocate: request: %v", r.ContainerRequests)
        response := pluginapi.AllocateResponse{}
        containerResponse := new(pluginapi.ContainerAllocateResponse)

        m := new(pluginapi.Mount)
        m.HostPath = dpi.socketDir
        m.ContainerPath = dpi.socketDir
        m.ReadOnly = false
        containerResponse.Mounts = []*pluginapi.Mount{m}

        response.ContainerResponses = []*pluginapi.ContainerAllocateResponse{containerResponse}

        return &response, nil
}

func (dpi *SocketDevicePlugin) healthCheck() error {
        logger := log.DefaultLogger()
        watcher, err := fsnotify.NewWatcher()
        if err != nil {
                return fmt.Errorf("failed to creating a fsnotify watcher: %v", err)
        }
        defer watcher.Close()

        devicePath := filepath.Join(dpi.socketDir, dpi.socket)

        // Start watching the files before we check for their existence to avoid races
        err = watcher.Add(dpi.socketDir)
        if err != nil {
                return fmt.Errorf("failed to add the device root path to the watcher: %v", err)
        }

        _, err = os.Stat(devicePath)
        if err != nil {
                if !errors.Is(err, os.ErrNotExist) {
                        return fmt.Errorf("could not stat the device: %v", err)
                }
                logger.Warningf("device '%s' is not present, the device plugin can't expose it.", dpi.socketName)
                dpi.health <- deviceHealth{Health: pluginapi.Unhealthy}
        }
        logger.Infof("device '%s' is present.", devicePath)

        dirName := filepath.Dir(dpi.socketPath)
        err = watcher.Add(dirName)

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

        for {
                select {
                case <-dpi.stop:
                        return nil
                case err := <-watcher.Errors:
                        logger.Reason(err).Errorf("error watching devices and device plugin directory")
                case event := <-watcher.Events:
                        logger.V(4).Infof("health Event: %v", event)
                        if event.Name == devicePath {
                                // Health in this case is if the device path actually exists
                                if event.Op == fsnotify.Create {
                                        logger.Infof("monitored device %s appeared", dpi.socketName)
                                        dpi.health <- deviceHealth{Health: pluginapi.Healthy}
                                } else if (event.Op == fsnotify.Remove) || (event.Op == fsnotify.Rename) {
                                        logger.Infof("monitored device %s disappeared", dpi.socketName)
                                        dpi.health <- deviceHealth{Health: pluginapi.Unhealthy}
                                }
                        } else if event.Name == dpi.socketPath && event.Op == fsnotify.Remove {
                                logger.Infof("device socket file for device %s was removed, kubelet probably restarted.", dpi.socketName)
                                return nil
                        }
                }
        }
}

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	"context"
24	"errors"
25	"fmt"
26	"net"
27	"os"
28	"path"
29	"path/filepath"
30	"strconv"
31	"strings"
32	"sync"
33
34	"github.com/fsnotify/fsnotify"
35	"google.golang.org/grpc"
36
37	"kubevirt.io/client-go/log"
38
39	"kubevirt.io/kubevirt/pkg/safepath"
40	"kubevirt.io/kubevirt/pkg/util"
41	pluginapi "kubevirt.io/kubevirt/pkg/virt-handler/device-manager/deviceplugin/v1beta1"
42	"kubevirt.io/kubevirt/pkg/virt-handler/selinux"
43	)
44
45	//go:generate mockgen -source $GOFILE -package=$GOPACKAGE -destination=generated_mock_$GOFILE
46
47	type PermissionManager interface {
48	ChownAtNoFollow(path *safepath.Path, uid, gid int) error
49	}
50
51	type permissionManager struct{}
52
53	func NewPermissionManager() PermissionManager {	×
54	return &permissionManager{}	×
55	}	×
56
57	func (p permissionManager) ChownAtNoFollow(path safepath.Path, uid, gid int) error {	×
58	return safepath.ChownAtNoFollow(path, uid, gid)	×
59	}	×
60
61	type SocketDevicePlugin struct {
62	*DevicePluginBase
63	socketDir string
64	socket string
65	socketName string
66	executor selinux.Executor
67	p PermissionManager
68	}
69
70	func (dpi *SocketDevicePlugin) Start(stop <-chan struct{}) (err error) {	×
71	logger := log.DefaultLogger()	×
72	dpi.stop = stop	×
73		×
74	err = dpi.cleanup()	×
75	if err != nil {	×
76	return err	×
77	}	×
78
79	sock, err := net.Listen("unix", dpi.socketPath)	×
80	if err != nil {	×
81	return fmt.Errorf("error creating GRPC server socket: %v", err)	×
82	}	×
83
84	dpi.server = grpc.NewServer([]grpc.ServerOption{}...)	×
85	defer dpi.stopDevicePlugin()	×
86		×
87	pluginapi.RegisterDevicePluginServer(dpi.server, dpi)	×
88		×
89	errChan := make(chan error, 2)	×
90		×
91	go func() {	×
92	errChan <- dpi.server.Serve(sock)	×
93	}()	×
94
95	err = waitForGRPCServer(dpi.socketPath, connectionTimeout)	×
96	if err != nil {	×
97	return fmt.Errorf("error starting the GRPC server: %v", err)	×
98	}	×
99
100	err = dpi.register()	×
101	if err != nil {	×
102	return fmt.Errorf("error registering with device plugin manager: %v", err)	×
103	}	×
104
105	go func() {	×
106	errChan <- dpi.healthCheck()	×
107	}()	×
108
109	dpi.setInitialized(true)	×
110	logger.Infof("%s device plugin started", dpi.resourceName)	×
111	err = <-errChan	×
112		×
113	return err	×
114	}
115
116	func (dpi *SocketDevicePlugin) setSocketPermissions() error {	2✔
117	prSock, err := safepath.JoinAndResolveWithRelativeRoot("/", dpi.socketDir, dpi.socket)	2✔
118	if err != nil {	2✔
119	return fmt.Errorf("error opening the socket %s/%s: %v", dpi.socketDir, dpi.socketName, err)	×
120	}	×
121	err = dpi.p.ChownAtNoFollow(prSock, util.NonRootUID, util.NonRootUID)	2✔
122	if err != nil {	2✔
123	return fmt.Errorf("error setting the permission the socket %s/%s:%v", dpi.socketDir, dpi.socketName, err)	×
124	}	×
125	if se, exists, err := dpi.executor.NewSELinux(); err == nil && exists {	4✔
126	if err := selinux.RelabelFilesUnprivileged(se.IsPermissive(), prSock); err != nil {	2✔
127	return fmt.Errorf("error relabeling required files: %v", err)	×
128	}	×
129	} else if err != nil {	×
130	return fmt.Errorf("failed to detect the presence of selinux: %v", err)	×
131	}	×
132
133	return nil	2✔
134	}
135
136	func (dpi *SocketDevicePlugin) setSocketDirectoryPermissions() error {	2✔
137	dir, err := safepath.JoinAndResolveWithRelativeRoot("/", dpi.socketDir)	2✔
138	if err != nil {	2✔
139	return fmt.Errorf("error opening the socket dir %s: %v", dpi.socket, err)	×
140	}	×
141	err = dpi.p.ChownAtNoFollow(dir, util.NonRootUID, util.NonRootUID)	2✔
142	if err != nil {	2✔
143	return fmt.Errorf("error setting the permission the socket dir %s: %v", dpi.socketDir, err)	×
144	}	×
145	if se, exists, err := dpi.executor.NewSELinux(); err == nil && exists {	4✔
146	if err := selinux.RelabelFilesUnprivileged(se.IsPermissive(), dir); err != nil {	2✔
147	return fmt.Errorf("error relabeling required files: %v", err)	×
148	}	×
149	} else if err != nil {	×
150	return fmt.Errorf("failed to detect the presence of selinux: %v", err)	×
151	}	×
152
153	return nil	2✔
154	}
155
156	func NewSocketDevicePlugin(socketName, socketDir, socket string, maxDevices int, executor selinux.Executor, p PermissionManager) (*SocketDevicePlugin, error) {	2✔
157	dpi := &SocketDevicePlugin{	2✔
158	DevicePluginBase: &DevicePluginBase{	2✔
159	health: make(chan deviceHealth),	2✔
160	resourceName: fmt.Sprintf("%s/%s", DeviceNamespace, socketName),	2✔
161	initialized: false,	2✔
162	lock: &sync.Mutex{},	2✔
163	done: make(chan struct{}),	2✔
164	deregistered: make(chan struct{}),	2✔
165	socketPath: SocketPath(strings.Replace(socketName, "/", "-", -1)),	2✔
166	},	2✔
167	socket: socket,	2✔
168	socketDir: socketDir,	2✔
169	socketName: socketName,	2✔
170	executor: executor,	2✔
171	p: p,	2✔
172	}	2✔
173		2✔
174	for i := 0; i < maxDevices; i++ {	4✔
175	deviceId := dpi.socketName + strconv.Itoa(i)	2✔
176	dpi.devs = append(dpi.devs, &pluginapi.Device{	2✔
177	ID: deviceId,	2✔
178	Health: pluginapi.Healthy,	2✔
179	})	2✔
180	}	2✔
181	if err := dpi.setSocketDirectoryPermissions(); err != nil {	2✔
182	return nil, err	×
183	}	×
184	if err := dpi.setSocketPermissions(); err != nil {	2✔
185	return nil, err	×
186	}	×
187
188	return dpi, nil	2✔
189	}
190
191	// Register registers the device plugin for the given resourceName with Kubelet.
192	func (dpi *SocketDevicePlugin) register() error {	×
193	conn, err := gRPCConnect(pluginapi.KubeletSocket, connectionTimeout)	×
194	if err != nil {	×
195	return err	×
196	}	×
197	defer conn.Close()	×
198		×
199	client := pluginapi.NewRegistrationClient(conn)	×
200	reqt := &pluginapi.RegisterRequest{	×
201	Version: pluginapi.Version,	×
202	Endpoint: path.Base(dpi.socketPath),	×
203	ResourceName: dpi.resourceName,	×
204	}	×
205		×
206	_, err = client.Register(context.Background(), reqt)	×
207	if err != nil {	×
208	return err	×
209	}	×
210	return nil	×
211	}
212
213	func (dpi SocketDevicePlugin) Allocate(ctx context.Context, r pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {	×
214	log.DefaultLogger().Infof("Socket Allocate: resourceName: %s", dpi.socketName)	×
215	log.DefaultLogger().Infof("Socket Allocate: request: %v", r.ContainerRequests)	×
216	response := pluginapi.AllocateResponse{}	×
217	containerResponse := new(pluginapi.ContainerAllocateResponse)	×
218		×
219	m := new(pluginapi.Mount)	×
220	m.HostPath = dpi.socketDir	×
221	m.ContainerPath = dpi.socketDir	×
222	m.ReadOnly = false	×
223	containerResponse.Mounts = []*pluginapi.Mount{m}	×
224		×
225	response.ContainerResponses = []*pluginapi.ContainerAllocateResponse{containerResponse}	×
226		×
227	return &response, nil	×
228	}	×
229
230	func (dpi *SocketDevicePlugin) healthCheck() error {	2✔
231	logger := log.DefaultLogger()	2✔
232	watcher, err := fsnotify.NewWatcher()	2✔
233	if err != nil {	2✔
234	return fmt.Errorf("failed to creating a fsnotify watcher: %v", err)	×
235	}	×
236	defer watcher.Close()	2✔
237		2✔
238	devicePath := filepath.Join(dpi.socketDir, dpi.socket)	2✔
239		2✔
240	// Start watching the files before we check for their existence to avoid races	2✔
241	err = watcher.Add(dpi.socketDir)	2✔
242	if err != nil {	2✔
243	return fmt.Errorf("failed to add the device root path to the watcher: %v", err)	×
244	}	×
245
246	_, err = os.Stat(devicePath)	2✔
247	if err != nil {	3✔
248	if !errors.Is(err, os.ErrNotExist) {	1✔
249	return fmt.Errorf("could not stat the device: %v", err)	×
250	}	×
251	logger.Warningf("device '%s' is not present, the device plugin can't expose it.", dpi.socketName)	1✔
252	dpi.health <- deviceHealth{Health: pluginapi.Unhealthy}	1✔
253	}
254	logger.Infof("device '%s' is present.", devicePath)	2✔
255		2✔
256	dirName := filepath.Dir(dpi.socketPath)	2✔
257	err = watcher.Add(dirName)	2✔
258		2✔
259	if err != nil {	2✔
260	return fmt.Errorf("failed to add the device-plugin kubelet path to the watcher: %v", err)	×
261	}	×
262	_, err = os.Stat(dpi.socketPath)	2✔
263	if err != nil {	2✔
264	return fmt.Errorf("failed to stat the device-plugin socket: %v", err)	×
265	}	×
266
267	for {	5✔
268	select {	3✔
269	case <-dpi.stop:	1✔
270	return nil	1✔
271	case err := <-watcher.Errors:	×
272	logger.Reason(err).Errorf("error watching devices and device plugin directory")	×
273	case event := <-watcher.Events:	2✔
274	logger.V(4).Infof("health Event: %v", event)	2✔
275	if event.Name == devicePath {	3✔
276	// Health in this case is if the device path actually exists	1✔
277	if event.Op == fsnotify.Create {	2✔
278	logger.Infof("monitored device %s appeared", dpi.socketName)	1✔
279	dpi.health <- deviceHealth{Health: pluginapi.Healthy}	1✔
280	} else if (event.Op == fsnotify.Remove) \|\| (event.Op == fsnotify.Rename) {	1✔
UNCOV 281	logger.Infof("monitored device %s disappeared", dpi.socketName)	×
UNCOV 282	dpi.health <- deviceHealth{Health: pluginapi.Unhealthy}	×
UNCOV 283	}	×
284	} else if event.Name == dpi.socketPath && event.Op == fsnotify.Remove {	2✔
285	logger.Infof("device socket file for device %s was removed, kubelet probably restarted.", dpi.socketName)	1✔
286	return nil	1✔
287	}	1✔
288	}
289	}
290	}

kubevirt / kubevirt / 4282cb34-bade-47c2-9e61-a1b730ef4465

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