26086512318

Committed 19 May 2026 08:47AM UTC coverage: 53.573% (+2.7%) from 50.902%

Build # 26086512318

Build Type

Pull #95

github

Committed by

web-flow

Commit Message

Merge 8c639e2fc into 7e91a77c2

Pull Request Pull Request #95: WIP: Support custom MAC annotation

Coverage Stats

278 of 428 new or added lines in 9 files covered. (64.95%)

1 existing line in 1 file now uncovered.

2099 of 3918 relevant lines covered (53.57%)

3.68 hits per line

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

15.04

/pkg/driver/driver.go

/*
 * Copyright 2023 The Kubernetes Authors.
 *
 * 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.
 */

package driver

import (
        "context"
        "fmt"
        "maps"
        "os"
        "path"
        "time"

        resourceapi "k8s.io/api/resource/v1"
        k8stypes "k8s.io/apimachinery/pkg/types"
        coreclientset "k8s.io/client-go/kubernetes"
        metadatav1alpha1 "k8s.io/dynamic-resource-allocation/api/metadata/v1alpha1"
        "k8s.io/dynamic-resource-allocation/kubeletplugin"
        "k8s.io/dynamic-resource-allocation/resourceslice"
        "k8s.io/klog/v2"

        "github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/cdi"
        "github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/consts"
        "github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/devicestate"
        "github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/podmanager"
        sriovdratype "github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/types"
)

// These option constructors are package variables so unit tests can override
// them with fakes and verify plugin option wiring deterministically.
var (
        enableDeviceMetadataOption = kubeletplugin.EnableDeviceMetadata
        cdiDirectoryOption         = kubeletplugin.CDIDirectory
        metadataVersionsOption     = kubeletplugin.MetadataVersions
)

type Driver struct {
        client             coreclientset.Interface
        helper             *kubeletplugin.Helper
        deviceStateManager *devicestate.Manager
        podManager         *podmanager.PodManager
        healthcheck        *Healthcheck
        cancelCtx          func(error)
        config             *sriovdratype.Config
        cdi                *cdi.Handler
}

func buildPluginOptions(config *sriovdratype.Config) []kubeletplugin.Option {
        pluginOpts := []kubeletplugin.Option{
                kubeletplugin.KubeClient(config.K8sClient.Interface),
                kubeletplugin.NodeName(config.Flags.NodeName),
                kubeletplugin.DriverName(consts.DriverName),
                kubeletplugin.RegistrarDirectoryPath(config.Flags.KubeletRegistrarDirectoryPath),
                kubeletplugin.PluginDataDirectoryPath(config.DriverPluginPath()),
        }
        if config.Flags.EnableDeviceMetadata {
                pluginOpts = append(
                        pluginOpts,
                        enableDeviceMetadataOption(true),
                        cdiDirectoryOption(config.Flags.CdiRoot),
                        metadataVersionsOption(metadatav1alpha1.SchemeGroupVersion),
                )
        }
        return pluginOpts
}

// Start creates a new DRA driver and starts the kubelet plugin. It waits for the plugin to be registered
// with the kubelet before starting the healthcheck service and publishing the available resources.
func Start(ctx context.Context, config *sriovdratype.Config, deviceStateManager *devicestate.Manager, podManager *podmanager.PodManager, cdi *cdi.Handler) (*Driver, error) {
        driver := &Driver{
                client:             config.K8sClient.Interface,
                cancelCtx:          config.CancelMainCtx,
                config:             config,
                deviceStateManager: deviceStateManager,
                podManager:         podManager,
                cdi:                cdi,
        }

        pluginOpts := buildPluginOptions(config)
        helper, err := kubeletplugin.Start(ctx, driver, pluginOpts...)
        if err != nil {
                klog.FromContext(ctx).Error(err, "Failed to start DRA kubelet plugin")
                return nil, err
        }
        driver.helper = helper

        // Wait for plugin registration to complete
        if err = waitForRegistration(ctx, helper); err != nil {
                return nil, err
        }

        driver.healthcheck, err = startHealthcheck(ctx, config)
        if err != nil {
                return nil, fmt.Errorf("start healthcheck: %w", err)
        }

        // Publish resources
        if err = driver.PublishResources(ctx); err != nil {
                return nil, fmt.Errorf("failed to publish resources: %w", err)
        }
        return driver, nil
}

// waitForRegistration waits for the plugin to be registered with the kubelet
func waitForRegistration(ctx context.Context, helper *kubeletplugin.Helper) error {
        logger := klog.FromContext(ctx)
        logger.Info("Waiting for plugin registration")

        ticker := time.NewTicker(100 * time.Millisecond)
        defer ticker.Stop()

        timeout := time.After(30 * time.Second)

        for {
                select {
                case <-timeout:
                        return fmt.Errorf("timeout waiting for plugin registration")
                case <-ticker.C:
                        status := helper.RegistrationStatus()
                        if status != nil && status.PluginRegistered {
                                logger.Info("Plugin registration completed successfully")
                                return nil
                        }
                        if status != nil && status.Error != "" {
                                return fmt.Errorf("plugin registration failed: %s", status.Error)
                        }
                }
        }
}

// Shutdown shuts down the driver
func (d *Driver) Shutdown(logger klog.Logger) error {
        if d.healthcheck != nil {
                d.healthcheck.Stop(logger)
        }
        d.helper.Stop()

        // remove the socket files
        // TODO: this is not needed after https://github.com/kubernetes/kubernetes/pull/133934 is merged
        err := os.Remove(path.Join(d.config.Flags.KubeletRegistrarDirectoryPath, consts.DriverName+"-reg.sock"))
        if err != nil && !os.IsNotExist(err) {
                return fmt.Errorf("error removing socket file: %w", err)
        }
        err = os.Remove(path.Join(d.config.DriverPluginPath(), "dra.sock"))
        if err != nil && !os.IsNotExist(err) {
                return fmt.Errorf("error removing socket file: %w", err)
        }

        return nil
}

// PublishResources publishes policy-matched devices to the DRA resource slice.
// Only devices matched by a SriovResourcePolicy are advertised.
func (d *Driver) PublishResources(ctx context.Context) error {
        advertised := d.deviceStateManager.GetAdvertisedDevices()
        devices := make([]resourceapi.Device, 0, len(advertised))
        for device := range maps.Values(advertised) {
                devices = append(devices, device)
        }
        resources := resourceslice.DriverResources{
                Pools: map[string]resourceslice.Pool{
                        d.config.Flags.NodeName: {
                                Slices: []resourceslice.Slice{
                                        {
                                                Devices: devices,
                                        },
                                },
                        },
                },
        }

        if err := d.helper.PublishResources(ctx, resources); err != nil {
                return err
        }
        return nil
}

// UpdateRequestMetadata refreshes per-request metadata files for a prepared claim.
func (d *Driver) UpdateRequestMetadata(
        ctx context.Context,
        claimNamespace, claimName string,
        claimUID k8stypes.UID,
        requestName string,
        devices []kubeletplugin.Device,
) error {
        if d.helper == nil {
                return fmt.Errorf("kubelet plugin helper is not initialized")
        }
        return d.helper.UpdateRequestMetadata(ctx, claimNamespace, claimName, claimUID, requestName, devices)
}

1	/*
2	* Copyright 2023 The Kubernetes Authors.
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
17	package driver
18
19	import (
20	"context"
21	"fmt"
22	"maps"
23	"os"
24	"path"
25	"time"
26
27	resourceapi "k8s.io/api/resource/v1"
28	k8stypes "k8s.io/apimachinery/pkg/types"
29	coreclientset "k8s.io/client-go/kubernetes"
30	metadatav1alpha1 "k8s.io/dynamic-resource-allocation/api/metadata/v1alpha1"
31	"k8s.io/dynamic-resource-allocation/kubeletplugin"
32	"k8s.io/dynamic-resource-allocation/resourceslice"
33	"k8s.io/klog/v2"
34
35	"github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/cdi"
36	"github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/consts"
37	"github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/devicestate"
38	"github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/podmanager"
39	sriovdratype "github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/types"
40	)
41
42	// These option constructors are package variables so unit tests can override
43	// them with fakes and verify plugin option wiring deterministically.
44	var (
45	enableDeviceMetadataOption = kubeletplugin.EnableDeviceMetadata
46	cdiDirectoryOption = kubeletplugin.CDIDirectory
47	metadataVersionsOption = kubeletplugin.MetadataVersions
48	)
49
50	type Driver struct {
51	client coreclientset.Interface
52	helper *kubeletplugin.Helper
53	deviceStateManager *devicestate.Manager
54	podManager *podmanager.PodManager
55	healthcheck *Healthcheck
56	cancelCtx func(error)
57	config *sriovdratype.Config
58	cdi *cdi.Handler
59	}
60
61	func buildPluginOptions(config *sriovdratype.Config) []kubeletplugin.Option {	2✔
62	pluginOpts := []kubeletplugin.Option{	2✔
63	kubeletplugin.KubeClient(config.K8sClient.Interface),	2✔
64	kubeletplugin.NodeName(config.Flags.NodeName),	2✔
65	kubeletplugin.DriverName(consts.DriverName),	2✔
66	kubeletplugin.RegistrarDirectoryPath(config.Flags.KubeletRegistrarDirectoryPath),	2✔
67	kubeletplugin.PluginDataDirectoryPath(config.DriverPluginPath()),	2✔
68	}	2✔
69	if config.Flags.EnableDeviceMetadata {	3✔
70	pluginOpts = append(	1✔
71	pluginOpts,	1✔
72	enableDeviceMetadataOption(true),	1✔
73	cdiDirectoryOption(config.Flags.CdiRoot),	1✔
74	metadataVersionsOption(metadatav1alpha1.SchemeGroupVersion),	1✔
75	)	1✔
76	}	1✔
77	return pluginOpts	2✔
78	}
79
80	// Start creates a new DRA driver and starts the kubelet plugin. It waits for the plugin to be registered
81	// with the kubelet before starting the healthcheck service and publishing the available resources.
82	func Start(ctx context.Context, config sriovdratype.Config, deviceStateManager devicestate.Manager, podManager podmanager.PodManager, cdi cdi.Handler) (*Driver, error) {	×
83	driver := &Driver{	×
84	client: config.K8sClient.Interface,	×
85	cancelCtx: config.CancelMainCtx,	×
86	config: config,	×
87	deviceStateManager: deviceStateManager,	×
88	podManager: podManager,	×
89	cdi: cdi,	×
90	}	×
91		×
NEW 92	pluginOpts := buildPluginOptions(config)	×
NEW 93	helper, err := kubeletplugin.Start(ctx, driver, pluginOpts...)	×
94	if err != nil {	×
95	klog.FromContext(ctx).Error(err, "Failed to start DRA kubelet plugin")	×
96	return nil, err	×
97	}	×
98	driver.helper = helper	×
99		×
100	// Wait for plugin registration to complete	×
101	if err = waitForRegistration(ctx, helper); err != nil {	×
102	return nil, err	×
103	}	×
104
105	driver.healthcheck, err = startHealthcheck(ctx, config)	×
106	if err != nil {	×
107	return nil, fmt.Errorf("start healthcheck: %w", err)	×
108	}	×
109
110	// Publish resources
111	if err = driver.PublishResources(ctx); err != nil {	×
112	return nil, fmt.Errorf("failed to publish resources: %w", err)	×
113	}	×
114	return driver, nil	×
115	}
116
117	// waitForRegistration waits for the plugin to be registered with the kubelet
118	func waitForRegistration(ctx context.Context, helper *kubeletplugin.Helper) error {	×
119	logger := klog.FromContext(ctx)	×
120	logger.Info("Waiting for plugin registration")	×
121		×
122	ticker := time.NewTicker(100 * time.Millisecond)	×
123	defer ticker.Stop()	×
124		×
125	timeout := time.After(30 * time.Second)	×
126		×
127	for {	×
128	select {	×
129	case <-timeout:	×
130	return fmt.Errorf("timeout waiting for plugin registration")	×
131	case <-ticker.C:	×
132	status := helper.RegistrationStatus()	×
133	if status != nil && status.PluginRegistered {	×
134	logger.Info("Plugin registration completed successfully")	×
135	return nil	×
136	}	×
137	if status != nil && status.Error != "" {	×
138	return fmt.Errorf("plugin registration failed: %s", status.Error)	×
139	}	×
140	}
141	}
142	}
143
144	// Shutdown shuts down the driver
145	func (d *Driver) Shutdown(logger klog.Logger) error {	×
146	if d.healthcheck != nil {	×
147	d.healthcheck.Stop(logger)	×
148	}	×
149	d.helper.Stop()	×
150		×
151	// remove the socket files	×
152	// TODO: this is not needed after https://github.com/kubernetes/kubernetes/pull/133934 is merged	×
153	err := os.Remove(path.Join(d.config.Flags.KubeletRegistrarDirectoryPath, consts.DriverName+"-reg.sock"))	×
154	if err != nil && !os.IsNotExist(err) {	×
155	return fmt.Errorf("error removing socket file: %w", err)	×
156	}	×
157	err = os.Remove(path.Join(d.config.DriverPluginPath(), "dra.sock"))	×
158	if err != nil && !os.IsNotExist(err) {	×
159	return fmt.Errorf("error removing socket file: %w", err)	×
160	}	×
161
162	return nil	×
163	}
164
165	// PublishResources publishes policy-matched devices to the DRA resource slice.
166	// Only devices matched by a SriovResourcePolicy are advertised.
167	func (d *Driver) PublishResources(ctx context.Context) error {	×
168	advertised := d.deviceStateManager.GetAdvertisedDevices()	×
169	devices := make([]resourceapi.Device, 0, len(advertised))	×
170	for device := range maps.Values(advertised) {	×
171	devices = append(devices, device)	×
172	}	×
173	resources := resourceslice.DriverResources{	×
174	Pools: map[string]resourceslice.Pool{	×
175	d.config.Flags.NodeName: {	×
176	Slices: []resourceslice.Slice{	×
177	{	×
178	Devices: devices,	×
179	},	×
180	},	×
181	},	×
182	},	×
183	}	×
184		×
185	if err := d.helper.PublishResources(ctx, resources); err != nil {	×
186	return err	×
187	}	×
188	return nil	×
189	}
190
191	// UpdateRequestMetadata refreshes per-request metadata files for a prepared claim.
192	func (d *Driver) UpdateRequestMetadata(
193	ctx context.Context,
194	claimNamespace, claimName string,
195	claimUID k8stypes.UID,
196	requestName string,
197	devices []kubeletplugin.Device,
NEW 198	) error {	×
NEW 199	if d.helper == nil {	×
NEW 200	return fmt.Errorf("kubelet plugin helper is not initialized")	×
NEW 201	}	×
NEW 202	return d.helper.UpdateRequestMetadata(ctx, claimNamespace, claimName, claimUID, requestName, devices)	×
203	}

k8snetworkplumbingwg / dra-driver-sriov / 26086512318

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