211fba77-7e9b-410a-9b44-0eb131ae6628

Committed 28 Feb 2025 02:41PM UTC coverage: 71.521% (+0.006%) from 71.515%

Build # 211fba77-7e9b-410a-9b44-0eb131ae6628

Build Type

push

prow

Committed by

web-flow

Commit Message

Merge pull request #14049 from lyarwood/instancetype-graduate-referencepolicy-beta

featuregates: Graduate InstancetypeReferencePolicy to Beta

Coverage Stats

1 of 1 new or added line in 1 file covered. (100.0%)

7 existing lines in 2 files now uncovered.

61835 of 86457 relevant lines covered (71.52%)

0.8 hits per line

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

38.55

/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 2023 Red Hat, Inc.
 *
 */

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"
)

type SocketDevicePlugin struct {
        *DevicePluginBase
        socketDir  string
        socket     string
        socketName string
}

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 NewSocketDevicePlugin(socketName, socketDir, socket string, maxDevices int) *SocketDevicePlugin {
        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,
        }

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

// 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)

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

        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 2023 Red Hat, Inc.
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	type SocketDevicePlugin struct {
46	*DevicePluginBase
47	socketDir string
48	socket string
49	socketName string
50	}
51
52	func (dpi *SocketDevicePlugin) Start(stop <-chan struct{}) (err error) {	×
53	logger := log.DefaultLogger()	×
54	dpi.stop = stop	×
55		×
56	err = dpi.cleanup()	×
57	if err != nil {	×
58	return err	×
59	}	×
60
61	sock, err := net.Listen("unix", dpi.socketPath)	×
62	if err != nil {	×
63	return fmt.Errorf("error creating GRPC server socket: %v", err)	×
64	}	×
65
66	dpi.server = grpc.NewServer([]grpc.ServerOption{}...)	×
67	defer dpi.stopDevicePlugin()	×
68		×
69	pluginapi.RegisterDevicePluginServer(dpi.server, dpi)	×
70		×
71	errChan := make(chan error, 2)	×
72		×
73	go func() {	×
74	errChan <- dpi.server.Serve(sock)	×
75	}()	×
76
77	err = waitForGRPCServer(dpi.socketPath, connectionTimeout)	×
78	if err != nil {	×
79	return fmt.Errorf("error starting the GRPC server: %v", err)	×
80	}	×
81
82	err = dpi.register()	×
83	if err != nil {	×
84	return fmt.Errorf("error registering with device plugin manager: %v", err)	×
85	}	×
86
87	go func() {	×
88	errChan <- dpi.healthCheck()	×
89	}()	×
90
91	dpi.setInitialized(true)	×
92	logger.Infof("%s device plugin started", dpi.resourceName)	×
93	err = <-errChan	×
94		×
95	return err	×
96	}
97
98	func NewSocketDevicePlugin(socketName, socketDir, socket string, maxDevices int) *SocketDevicePlugin {	1✔
99	dpi := &SocketDevicePlugin{	1✔
100	DevicePluginBase: &DevicePluginBase{	1✔
101	health: make(chan deviceHealth),	1✔
102	resourceName: fmt.Sprintf("%s/%s", DeviceNamespace, socketName),	1✔
103	initialized: false,	1✔
104	lock: &sync.Mutex{},	1✔
105	done: make(chan struct{}),	1✔
106	deregistered: make(chan struct{}),	1✔
107	socketPath: SocketPath(strings.Replace(socketName, "/", "-", -1)),	1✔
108	},	1✔
109	socket: socket,	1✔
110	socketDir: socketDir,	1✔
111	socketName: socketName,	1✔
112	}	1✔
113		1✔
114	for i := 0; i < maxDevices; i++ {	2✔
115	deviceId := dpi.socketName + strconv.Itoa(i)	1✔
116	dpi.devs = append(dpi.devs, &pluginapi.Device{	1✔
117	ID: deviceId,	1✔
118	Health: pluginapi.Healthy,	1✔
119	})	1✔
120	}	1✔
121	return dpi	1✔
122	}
123
124	// Register registers the device plugin for the given resourceName with Kubelet.
125	func (dpi *SocketDevicePlugin) register() error {	×
126	conn, err := gRPCConnect(pluginapi.KubeletSocket, connectionTimeout)	×
127	if err != nil {	×
128	return err	×
129	}	×
130	defer conn.Close()	×
131		×
132	client := pluginapi.NewRegistrationClient(conn)	×
133	reqt := &pluginapi.RegisterRequest{	×
134	Version: pluginapi.Version,	×
135	Endpoint: path.Base(dpi.socketPath),	×
136	ResourceName: dpi.resourceName,	×
137	}	×
138		×
139	_, err = client.Register(context.Background(), reqt)	×
140	if err != nil {	×
141	return err	×
142	}	×
143	return nil	×
144	}
145
146	func (dpi SocketDevicePlugin) Allocate(ctx context.Context, r pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {	×
147	log.DefaultLogger().Infof("Socket Allocate: resourceName: %s", dpi.socketName)	×
148	log.DefaultLogger().Infof("Socket Allocate: request: %v", r.ContainerRequests)	×
149	response := pluginapi.AllocateResponse{}	×
150	containerResponse := new(pluginapi.ContainerAllocateResponse)	×
151		×
152	prSock, err := safepath.JoinAndResolveWithRelativeRoot("/", dpi.socketDir, dpi.socket)	×
153	if err != nil {	×
154	return nil, fmt.Errorf("error opening the socket %s/%s: %v", dpi.socketDir, dpi.socket, err)	×
155	}	×
156	err = safepath.ChownAtNoFollow(prSock, util.NonRootUID, util.NonRootUID)	×
157	if err != nil {	×
158	return nil, fmt.Errorf("error setting the permission the socket %s/%s:%v", dpi.socketDir, dpi.socket, err)	×
159	}	×
160	if se, exists, err := selinux.NewSELinux(); err == nil && exists {	×
161	if err := selinux.RelabelFilesUnprivileged(se.IsPermissive(), prSock); err != nil {	×
162	return nil, fmt.Errorf("error relabeling required files: %v", err)	×
163	}	×
164	} else if err != nil {	×
165	return nil, fmt.Errorf("failed to detect the presence of selinux: %v", err)	×
166	}	×
167
168	m := new(pluginapi.Mount)	×
169	m.HostPath = dpi.socketDir	×
170	m.ContainerPath = dpi.socketDir	×
171	m.ReadOnly = false	×
172	containerResponse.Mounts = []*pluginapi.Mount{m}	×
173		×
174	response.ContainerResponses = []*pluginapi.ContainerAllocateResponse{containerResponse}	×
175		×
176	return &response, nil	×
177	}
178
179	func (dpi *SocketDevicePlugin) healthCheck() error {	1✔
180	logger := log.DefaultLogger()	1✔
181	watcher, err := fsnotify.NewWatcher()	1✔
182	if err != nil {	1✔
183	return fmt.Errorf("failed to creating a fsnotify watcher: %v", err)	×
184	}	×
185	defer watcher.Close()	1✔
186		1✔
187	devicePath := filepath.Join(dpi.socketDir, dpi.socket)	1✔
188		1✔
189	// Start watching the files before we check for their existence to avoid races	1✔
190	err = watcher.Add(dpi.socketDir)	1✔
191	if err != nil {	1✔
192	return fmt.Errorf("failed to add the device root path to the watcher: %v", err)	×
193	}	×
194
195	_, err = os.Stat(devicePath)	1✔
196	if err != nil {	2✔
197	if !errors.Is(err, os.ErrNotExist) {	1✔
198	return fmt.Errorf("could not stat the device: %v", err)	×
199	}	×
200	logger.Warningf("device '%s' is not present, the device plugin can't expose it.", dpi.socketName)	1✔
201	dpi.health <- deviceHealth{Health: pluginapi.Unhealthy}	1✔
202	}
203	logger.Infof("device '%s' is present.", devicePath)	1✔
204		1✔
205	dirName := filepath.Dir(dpi.socketPath)	1✔
206	err = watcher.Add(dirName)	1✔
207		1✔
208	if err != nil {	1✔
209	return fmt.Errorf("failed to add the device-plugin kubelet path to the watcher: %v", err)	×
210	}	×
211	_, err = os.Stat(dpi.socketPath)	1✔
212	if err != nil {	1✔
213	return fmt.Errorf("failed to stat the device-plugin socket: %v", err)	×
214	}	×
215
216	for {	2✔
217	select {	1✔
218	case <-dpi.stop:	1✔
219	return nil	1✔
220	case err := <-watcher.Errors:	×
221	logger.Reason(err).Errorf("error watching devices and device plugin directory")	×
222	case event := <-watcher.Events:	1✔
223	logger.V(4).Infof("health Event: %v", event)	1✔
224	if event.Name == devicePath {	2✔
225	// Health in this case is if the device path actually exists	1✔
226	if event.Op == fsnotify.Create {	2✔
227	logger.Infof("monitored device %s appeared", dpi.socketName)	1✔
228	dpi.health <- deviceHealth{Health: pluginapi.Healthy}	1✔
229	} else if (event.Op == fsnotify.Remove) \|\| (event.Op == fsnotify.Rename) {	1✔
UNCOV 230	logger.Infof("monitored device %s disappeared", dpi.socketName)	×
UNCOV 231	dpi.health <- deviceHealth{Health: pluginapi.Unhealthy}	×
UNCOV 232	}	×
233	} else if event.Name == dpi.socketPath && event.Op == fsnotify.Remove {	2✔
234	logger.Infof("device socket file for device %s was removed, kubelet probably restarted.", dpi.socketName)	1✔
235	return nil	1✔
236	}	1✔
237	}
238	}
239	}

kubevirt / kubevirt / 211fba77-7e9b-410a-9b44-0eb131ae6628

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