25047722054

Committed 28 Apr 2026 10:29AM UTC coverage: 55.308% (+4.6%) from 50.752%

Build # 25047722054

Build Type

Pull #102

github

Committed by

web-flow

Commit Message

Merge 9368816b5 into 4ea910cb1

Pull Request Pull Request #102: Support pci addressess

Coverage Stats

522 of 643 new or added lines in 9 files covered. (81.18%)

4 existing lines in 2 files now uncovered.

2261 of 4088 relevant lines covered (55.31%)

5.06 hits per line

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

80.0

/pkg/controller/resourcepolicycontroller.go

/*
 * Copyright 2025 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 controller

import (
        "context"
        "sort"
        "time"

        corev1 "k8s.io/api/core/v1"
        resourceapi "k8s.io/api/resource/v1"
        apierrors "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/util/workqueue"
        "k8s.io/klog/v2"
        ctrl "sigs.k8s.io/controller-runtime"
        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/event"
        "sigs.k8s.io/controller-runtime/pkg/handler"
        "sigs.k8s.io/controller-runtime/pkg/predicate"
        "sigs.k8s.io/controller-runtime/pkg/reconcile"
        "sigs.k8s.io/controller-runtime/pkg/source"

        sriovdrav1alpha1 "github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/api/sriovdra/v1alpha1"
        "github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/consts"
        "github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/devicestate"
)

const (
        resourcePolicySyncEventName = "resource-policy-sync"
)

// SriovResourcePolicyReconciler reconciles SriovResourcePolicy and DeviceAttributes objects
type SriovResourcePolicyReconciler struct {
        client.Client
        nodeName           string
        namespace          string
        log                klog.Logger
        deviceStateManager devicestate.DeviceState
}

// NewSriovResourcePolicyReconciler creates a new SriovResourcePolicyReconciler
func NewSriovResourcePolicyReconciler(client client.Client, nodeName, namespace string, deviceStateManager devicestate.DeviceState) *SriovResourcePolicyReconciler {
        return &SriovResourcePolicyReconciler{
                Client:             client,
                deviceStateManager: deviceStateManager,
                nodeName:           nodeName,
                namespace:          namespace,
                log:                klog.Background().WithName("SriovResourcePolicy"),
        }
}

// Reconcile handles reconciliation of SriovResourcePolicy and DeviceAttributes resources.
// It builds the full picture of which devices to advertise and which attributes to apply.
func (r *SriovResourcePolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
        r.log.Info("Starting reconcile", "request", req.NamespacedName, "watchedNamespace", r.namespace)

        node := &metav1.PartialObjectMetadata{}
        node.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Node"))
        if err := r.Get(ctx, types.NamespacedName{Name: r.nodeName}, node); err != nil {
                if apierrors.IsNotFound(err) {
                        r.log.Error(err, "Node not found", "nodeName", r.nodeName)
                        return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
                }
                r.log.Error(err, "Failed to get node", "nodeName", r.nodeName)
                return ctrl.Result{}, err
        }

        // List all SriovResourcePolicy objects in the operator namespace
        resourcePolicyList := &sriovdrav1alpha1.SriovResourcePolicyList{}
        if err := r.List(ctx, resourcePolicyList, client.InNamespace(r.namespace)); err != nil {
                r.log.Error(err, "Failed to list SriovResourcePolicy objects", "namespace", r.namespace)
                return ctrl.Result{}, err
        }

        // List all DeviceAttributes objects in the operator namespace
        deviceAttrList := &sriovdrav1alpha1.DeviceAttributesList{}
        if err := r.List(ctx, deviceAttrList, client.InNamespace(r.namespace)); err != nil {
                r.log.Error(err, "Failed to list DeviceAttributes objects", "namespace", r.namespace)
                return ctrl.Result{}, err
        }

        // Find matching resource policies for this node
        var matchingPolicies []*sriovdrav1alpha1.SriovResourcePolicy
        for i := range resourcePolicyList.Items {
                policy := &resourcePolicyList.Items[i]
                if r.matchesNodeSelector(node.Labels, policy.Spec.NodeSelector) {
                        matchingPolicies = append(matchingPolicies, policy)
                }
        }

        policyDevices := r.getPolicyDeviceMap(matchingPolicies, deviceAttrList.Items)
        if err := r.deviceStateManager.UpdatePolicyDevices(ctx, policyDevices); err != nil {
                r.log.Error(err, "Failed to update policy devices")
                return ctrl.Result{}, err
        }

        return ctrl.Result{}, nil
}

// getPolicyDeviceMap builds the full map of device name -> attributes for all
// devices matched by the given policies. DeviceAttributes are resolved via
// each config's DeviceAttributesSelector.
func (r *SriovResourcePolicyReconciler) getPolicyDeviceMap(
        policies []*sriovdrav1alpha1.SriovResourcePolicy,
        allDeviceAttrs []sriovdrav1alpha1.DeviceAttributes,
) map[string]map[resourceapi.QualifiedName]resourceapi.DeviceAttribute {
        policyDevices := make(map[string]map[resourceapi.QualifiedName]resourceapi.DeviceAttribute)
        selectedPfPciAddresses := make(map[string]struct{})
        selectedVFsByPfAddress := make(map[string]map[string]struct{})

        if len(policies) == 0 {
                r.log.Info("No matching SriovResourcePolicy found for node", "nodeName", r.nodeName)
                return policyDevices
        }

        sort.Slice(policies, func(i, j int) bool {
                return policies[i].Name < policies[j].Name
        })
        for _, policy := range policies {
                r.log.V(2).Info("Processing policy",
                        "policyName", policy.Name,
                        "configCount", len(policy.Spec.Configs))

                for _, config := range policy.Spec.Configs {
                        resolvedAttrs := r.resolveDeviceAttributes(config.DeviceAttributesSelector, allDeviceAttrs)

                        // If no resource filters are specified, all virtual function devices are candidates
                        if len(config.ResourceFilters) == 0 {
                                candidateDevices := r.deviceStateManager.GetPolicyCandidateDevices(false)
                                for deviceName := range candidateDevices {
                                        if _, exists := policyDevices[deviceName]; exists {
                                                continue
                                        }
                                        attrs := make(map[resourceapi.QualifiedName]resourceapi.DeviceAttribute, len(resolvedAttrs))
                                        for k, v := range resolvedAttrs {
                                                attrs[k] = v
                                        }
                                        policyDevices[deviceName] = attrs
                                        r.log.V(2).Info("Device matches config filter",
                                                "deviceName", deviceName,
                                                "policyName", policy.Name,
                                                "attributes", attrs)
                                }
                                continue
                        }

                        for _, filter := range config.ResourceFilters {
                                candidateDevices := r.deviceStateManager.GetPolicyCandidateDevices(filterRequestsPciAddresses(filter))
                                for deviceName, device := range candidateDevices {
                                        if _, exists := policyDevices[deviceName]; exists {
                                                continue
                                        }

                                        if r.deviceMatchesFilter(device, filter) {
                                                devicePciAddress, _ := deviceAttributeString(device, consts.AttributePciAddress)
                                                interfaceType := getDeviceInterfaceType(device)
                                                parentPfAddress, hasParentPfAddress := deviceAttributeString(device, consts.AttributePfPciAddress)

                                                if interfaceType == consts.InterfaceTypeVirtualFunction && hasParentPfAddress {
                                                        if _, pfSelected := selectedPfPciAddresses[parentPfAddress]; pfSelected {
                                                                r.log.Info("Skipping VF because its parent PF is already selected by pciAddresses",
                                                                        "deviceName", deviceName,
                                                                        "parentPfPciAddress", parentPfAddress)
                                                                continue
                                                        }
                                                }

                                                attrs := make(map[resourceapi.QualifiedName]resourceapi.DeviceAttribute, len(resolvedAttrs))
                                                for k, v := range resolvedAttrs {
                                                        attrs[k] = v
                                                }
                                                policyDevices[deviceName] = attrs
                                                if interfaceType == consts.InterfaceTypeVirtualFunction && hasParentPfAddress {
                                                        if _, exists := selectedVFsByPfAddress[parentPfAddress]; !exists {
                                                                selectedVFsByPfAddress[parentPfAddress] = make(map[string]struct{})
                                                        }
                                                        selectedVFsByPfAddress[parentPfAddress][deviceName] = struct{}{}
                                                }
                                                if filterRequestsPciAddresses(filter) && interfaceType == consts.InterfaceTypeRegular && devicePciAddress != "" {
                                                        selectedPfPciAddresses[devicePciAddress] = struct{}{}
                                                        if vfDeviceNames, exists := selectedVFsByPfAddress[devicePciAddress]; exists {
                                                                for vfDeviceName := range vfDeviceNames {
                                                                        delete(policyDevices, vfDeviceName)
                                                                        r.log.Info("Removing VF because parent PF is selected by pciAddresses",
                                                                                "vfDeviceName", vfDeviceName,
                                                                                "parentPfPciAddress", devicePciAddress,
                                                                                "selectedPfDeviceName", deviceName)
                                                                }
                                                                delete(selectedVFsByPfAddress, devicePciAddress)
                                                        }
                                                }

                                                r.log.V(2).Info("Device matches config filter",
                                                        "deviceName", deviceName,
                                                        "policyName", policy.Name,
                                                        "device", device,
                                                        "attributes", attrs)
                                        }
                                }
                        }
                }
        }

        r.log.Info("Policy devices resolved",
                "matchingDevices", len(policyDevices))
        r.log.V(2).Info("Policy devices details", "policyDevices", policyDevices)

        return policyDevices
}

// filterRequestsPciAddresses reports whether a filter explicitly selects by PCI address.
func filterRequestsPciAddresses(filter sriovdrav1alpha1.ResourceFilter) bool {
        return len(filter.PciAddresses) > 0
}

// getDeviceInterfaceType returns the interface type attribute with fallback inference.
func getDeviceInterfaceType(device resourceapi.Device) string {
        if interfaceType, exists := deviceAttributeString(device, consts.AttributeInterfaceType); exists {
                return interfaceType
        }
        if _, hasParentPf := deviceAttributeString(device, consts.AttributePfPciAddress); hasParentPf {
                return consts.InterfaceTypeVirtualFunction
        }
        return consts.InterfaceTypeRegular
}

// deviceAttributeString reads a string-valued device attribute by qualified name.
func deviceAttributeString(device resourceapi.Device, name resourceapi.QualifiedName) (string, bool) {
        if device.Attributes == nil {
                return "", false
        }
        attr, exists := device.Attributes[name]
        if !exists || attr.StringValue == nil {
                return "", false
        }
        return *attr.StringValue, true
}

// resolveDeviceAttributes finds all DeviceAttributes objects matching the
// given label selector and merges their attributes. When multiple objects
// match and define the same key, the value from the alphabetically last
// object name wins (deterministic).
func (r *SriovResourcePolicyReconciler) resolveDeviceAttributes(
        selector *metav1.LabelSelector,
        allDeviceAttrs []sriovdrav1alpha1.DeviceAttributes,
) map[resourceapi.QualifiedName]resourceapi.DeviceAttribute {
        if selector == nil {
                return nil
        }

        sel, err := metav1.LabelSelectorAsSelector(selector)
        if err != nil {
                r.log.Error(err, "Invalid DeviceAttributesSelector")
                return nil
        }

        // Collect matching DeviceAttributes, sort by name for determinism
        var matched []sriovdrav1alpha1.DeviceAttributes
        for i := range allDeviceAttrs {
                da := allDeviceAttrs[i]
                if sel.Matches(labels.Set(da.Labels)) {
                        matched = append(matched, da)
                }
        }
        sort.Slice(matched, func(i, j int) bool {
                return matched[i].Name < matched[j].Name
        })

        merged := make(map[resourceapi.QualifiedName]resourceapi.DeviceAttribute)
        for _, da := range matched {
                for key, val := range da.Spec.Attributes {
                        merged[key] = val
                }
        }

        return merged
}

// matchesNodeSelector checks if node labels match the given NodeSelector.
// A nil selector matches all nodes.
func (r *SriovResourcePolicyReconciler) matchesNodeSelector(nodeLabels map[string]string, nodeSelector *corev1.NodeSelector) bool {
        if nodeSelector == nil || len(nodeSelector.NodeSelectorTerms) == 0 {
                return true
        }

        // NodeSelectorTerms are ORed: at least one must match.
        for _, term := range nodeSelector.NodeSelectorTerms {
                ls := nodeSelectorTermToLabelSelector(term)
                selector, err := metav1.LabelSelectorAsSelector(ls)
                if err != nil {
                        r.log.V(2).Info("Failed to parse NodeSelectorTerm", "error", err)
                        continue
                }
                if selector.Matches(labels.Set(nodeLabels)) {
                        return true
                }
        }
        return false
}

func nodeSelectorTermToLabelSelector(term corev1.NodeSelectorTerm) *metav1.LabelSelector {
        var exprs []metav1.LabelSelectorRequirement
        for _, req := range term.MatchExpressions {
                exprs = append(exprs, metav1.LabelSelectorRequirement{
                        Key:      req.Key,
                        Operator: metav1.LabelSelectorOperator(req.Operator),
                        Values:   req.Values,
                })
        }
        return &metav1.LabelSelector{MatchExpressions: exprs}
}

// deviceMatchesFilter checks if a device matches a specific resource filter
func (r *SriovResourcePolicyReconciler) deviceMatchesFilter(device resourceapi.Device, filter sriovdrav1alpha1.ResourceFilter) bool {
        if len(filter.Vendors) > 0 {
                vendorAttr, exists := device.Attributes[consts.AttributeVendorID]
                if !exists || vendorAttr.StringValue == nil {
                        return false
                }
                if !stringSliceContains(filter.Vendors, *vendorAttr.StringValue) {
                        return false
                }
        }

        if len(filter.Devices) > 0 {
                deviceAttr, exists := device.Attributes[consts.AttributeDeviceID]
                if !exists || deviceAttr.StringValue == nil {
                        return false
                }
                if !stringSliceContains(filter.Devices, *deviceAttr.StringValue) {
                        return false
                }
        }

        if len(filter.PciAddresses) > 0 {
                pciAttr, exists := device.Attributes[consts.AttributePciAddress]
                if !exists || pciAttr.StringValue == nil {
                        return false
                }
                if !stringSliceContains(filter.PciAddresses, *pciAttr.StringValue) {
                        return false
                }
        }

        if len(filter.PfNames) > 0 {
                pfAttr, exists := device.Attributes[consts.AttributePFName]
                if !exists || pfAttr.StringValue == nil {
                        return false
                }
                if !stringSliceContains(filter.PfNames, *pfAttr.StringValue) {
                        return false
                }
        }

        if len(filter.PfPciAddresses) > 0 {
                parentAttr, exists := device.Attributes[consts.AttributePfPciAddress]
                if !exists || parentAttr.StringValue == nil {
                        return false
                }
                if !stringSliceContains(filter.PfPciAddresses, *parentAttr.StringValue) {
                        return false
                }
        }

        // TODO: Implement driver checking if needed
        if len(filter.Drivers) > 0 {
                r.log.V(3).Info("Driver filtering not yet implemented", "deviceName", device.Name)
        }

        return true
}

func stringSliceContains(slice []string, item string) bool {
        for _, s := range slice {
                if s == item {
                        return true
                }
        }
        return false
}

// SetupWithManager sets up the controller with the Manager.
func (r *SriovResourcePolicyReconciler) SetupWithManager(mgr ctrl.Manager) error {
        qHandler := func(q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
                q.AddAfter(reconcile.Request{NamespacedName: types.NamespacedName{
                        Namespace: r.namespace,
                        Name:      resourcePolicySyncEventName,
                }}, time.Second)
        }

        delayedEventHandler := handler.Funcs{
                CreateFunc: func(_ context.Context, e event.TypedCreateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {
                        r.log.Info("Enqueuing sync for create event",
                                "resource", e.Object.GetName(),
                                "type", e.Object.GetObjectKind().GroupVersionKind().String())
                        qHandler(w)
                },
                UpdateFunc: func(_ context.Context, e event.TypedUpdateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {
                        r.log.Info("Enqueuing sync for update event",
                                "resource", e.ObjectNew.GetName(),
                                "type", e.ObjectNew.GetObjectKind().GroupVersionKind().String())
                        qHandler(w)
                },
                DeleteFunc: func(_ context.Context, e event.TypedDeleteEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {
                        r.log.Info("Enqueuing sync for delete event",
                                "resource", e.Object.GetName(),
                                "type", e.Object.GetObjectKind().GroupVersionKind().String())
                        qHandler(w)
                },
                GenericFunc: func(_ context.Context, e event.TypedGenericEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {
                        r.log.Info("Enqueuing sync for generic event",
                                "resource", e.Object.GetName(),
                                "type", e.Object.GetObjectKind().GroupVersionKind().String())
                        qHandler(w)
                },
        }

        nodeEventHandler := handler.Funcs{
                CreateFunc: func(_ context.Context, e event.TypedCreateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {
                        if e.Object.GetName() == r.nodeName {
                                r.log.Info("Enqueuing sync for node create event", "node", e.Object.GetName())
                                qHandler(w)
                        }
                },
                UpdateFunc: func(_ context.Context, e event.TypedUpdateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {
                        if e.ObjectNew.GetName() == r.nodeName {
                                oldLabels := e.ObjectOld.GetLabels()
                                newLabels := e.ObjectNew.GetLabels()
                                if !labels.Equals(oldLabels, newLabels) {
                                        r.log.Info("Enqueuing sync for node label change event", "node", e.ObjectNew.GetName())
                                        qHandler(w)
                                }
                        }
                },
                DeleteFunc: func(_ context.Context, e event.TypedDeleteEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {
                        if e.Object.GetName() == r.nodeName {
                                r.log.Info("Enqueuing sync for node delete event", "node", e.Object.GetName())
                                qHandler(w)
                        }
                },
        }

        var eventChan = make(chan event.GenericEvent, 1)
        eventChan <- event.GenericEvent{Object: &sriovdrav1alpha1.SriovResourcePolicy{
                ObjectMeta: metav1.ObjectMeta{Name: resourcePolicySyncEventName, Namespace: r.namespace}}}
        close(eventChan)

        namespacePredicate := predicate.NewPredicateFuncs(func(obj client.Object) bool {
                return obj.GetNamespace() == r.namespace
        })

        nodeMetadata := &metav1.PartialObjectMetadata{}
        nodeMetadata.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Node"))

        return ctrl.NewControllerManagedBy(mgr).
                For(&sriovdrav1alpha1.SriovResourcePolicy{}).
                Watches(nodeMetadata, nodeEventHandler).
                Watches(&sriovdrav1alpha1.SriovResourcePolicy{}, delayedEventHandler).
                Watches(&sriovdrav1alpha1.DeviceAttributes{}, delayedEventHandler).
                WithEventFilter(namespacePredicate).
                WatchesRawSource(source.Channel(eventChan, &handler.EnqueueRequestForObject{})).
                Complete(r)
}

1	/*
2	* Copyright 2025 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 controller
18
19	import (
20	"context"
21	"sort"
22	"time"
23
24	corev1 "k8s.io/api/core/v1"
25	resourceapi "k8s.io/api/resource/v1"
26	apierrors "k8s.io/apimachinery/pkg/api/errors"
27	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
28	"k8s.io/apimachinery/pkg/labels"
29	"k8s.io/apimachinery/pkg/types"
30	"k8s.io/client-go/util/workqueue"
31	"k8s.io/klog/v2"
32	ctrl "sigs.k8s.io/controller-runtime"
33	"sigs.k8s.io/controller-runtime/pkg/client"
34	"sigs.k8s.io/controller-runtime/pkg/event"
35	"sigs.k8s.io/controller-runtime/pkg/handler"
36	"sigs.k8s.io/controller-runtime/pkg/predicate"
37	"sigs.k8s.io/controller-runtime/pkg/reconcile"
38	"sigs.k8s.io/controller-runtime/pkg/source"
39
40	sriovdrav1alpha1 "github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/api/sriovdra/v1alpha1"
41	"github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/consts"
42	"github.com/k8snetworkplumbingwg/dra-driver-sriov/pkg/devicestate"
43	)
44
45	const (
46	resourcePolicySyncEventName = "resource-policy-sync"
47	)
48
49	// SriovResourcePolicyReconciler reconciles SriovResourcePolicy and DeviceAttributes objects
50	type SriovResourcePolicyReconciler struct {
51	client.Client
52	nodeName string
53	namespace string
54	log klog.Logger
55	deviceStateManager devicestate.DeviceState
56	}
57
58	// NewSriovResourcePolicyReconciler creates a new SriovResourcePolicyReconciler
59	func NewSriovResourcePolicyReconciler(client client.Client, nodeName, namespace string, deviceStateManager devicestate.DeviceState) *SriovResourcePolicyReconciler {	2✔
60	return &SriovResourcePolicyReconciler{	2✔
61	Client: client,	2✔
62	deviceStateManager: deviceStateManager,	2✔
63	nodeName: nodeName,	2✔
64	namespace: namespace,	2✔
65	log: klog.Background().WithName("SriovResourcePolicy"),	2✔
66	}	2✔
67	}	2✔
68
69	// Reconcile handles reconciliation of SriovResourcePolicy and DeviceAttributes resources.
70	// It builds the full picture of which devices to advertise and which attributes to apply.
71	func (r *SriovResourcePolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	9✔
72	r.log.Info("Starting reconcile", "request", req.NamespacedName, "watchedNamespace", r.namespace)	9✔
73		9✔
74	node := &metav1.PartialObjectMetadata{}	9✔
75	node.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Node"))	9✔
76	if err := r.Get(ctx, types.NamespacedName{Name: r.nodeName}, node); err != nil {	10✔
77	if apierrors.IsNotFound(err) {	2✔
78	r.log.Error(err, "Node not found", "nodeName", r.nodeName)	1✔
79	return ctrl.Result{RequeueAfter: 30 * time.Second}, nil	1✔
80	}	1✔
81	r.log.Error(err, "Failed to get node", "nodeName", r.nodeName)	×
82	return ctrl.Result{}, err	×
83	}
84
85	// List all SriovResourcePolicy objects in the operator namespace
86	resourcePolicyList := &sriovdrav1alpha1.SriovResourcePolicyList{}	8✔
87	if err := r.List(ctx, resourcePolicyList, client.InNamespace(r.namespace)); err != nil {	8✔
88	r.log.Error(err, "Failed to list SriovResourcePolicy objects", "namespace", r.namespace)	×
89	return ctrl.Result{}, err	×
90	}	×
91
92	// List all DeviceAttributes objects in the operator namespace
93	deviceAttrList := &sriovdrav1alpha1.DeviceAttributesList{}	8✔
94	if err := r.List(ctx, deviceAttrList, client.InNamespace(r.namespace)); err != nil {	8✔
95	r.log.Error(err, "Failed to list DeviceAttributes objects", "namespace", r.namespace)	×
96	return ctrl.Result{}, err	×
97	}	×
98
99	// Find matching resource policies for this node
100	var matchingPolicies []*sriovdrav1alpha1.SriovResourcePolicy	8✔
101	for i := range resourcePolicyList.Items {	16✔
102	policy := &resourcePolicyList.Items[i]	8✔
103	if r.matchesNodeSelector(node.Labels, policy.Spec.NodeSelector) {	16✔
104	matchingPolicies = append(matchingPolicies, policy)	8✔
105	}	8✔
106	}
107
108	policyDevices := r.getPolicyDeviceMap(matchingPolicies, deviceAttrList.Items)	8✔
109	if err := r.deviceStateManager.UpdatePolicyDevices(ctx, policyDevices); err != nil {	8✔
110	r.log.Error(err, "Failed to update policy devices")	×
111	return ctrl.Result{}, err	×
112	}	×
113
114	return ctrl.Result{}, nil	8✔
115	}
116
117	// getPolicyDeviceMap builds the full map of device name -> attributes for all
118	// devices matched by the given policies. DeviceAttributes are resolved via
119	// each config's DeviceAttributesSelector.
120	func (r *SriovResourcePolicyReconciler) getPolicyDeviceMap(
121	policies []*sriovdrav1alpha1.SriovResourcePolicy,
122	allDeviceAttrs []sriovdrav1alpha1.DeviceAttributes,
123	) map[string]map[resourceapi.QualifiedName]resourceapi.DeviceAttribute {	16✔
124	policyDevices := make(map[string]map[resourceapi.QualifiedName]resourceapi.DeviceAttribute)	16✔
125	selectedPfPciAddresses := make(map[string]struct{})	16✔
126	selectedVFsByPfAddress := make(map[string]map[string]struct{})	16✔
127		16✔
128	if len(policies) == 0 {	19✔
129	r.log.Info("No matching SriovResourcePolicy found for node", "nodeName", r.nodeName)	3✔
130	return policyDevices	3✔
131	}	3✔
132
133	sort.Slice(policies, func(i, j int) bool {	17✔
134	return policies[i].Name < policies[j].Name	4✔
135	})	4✔
136	for _, policy := range policies {	30✔
137	r.log.V(2).Info("Processing policy",	17✔
138	"policyName", policy.Name,	17✔
139	"configCount", len(policy.Spec.Configs))	17✔
140		17✔
141	for _, config := range policy.Spec.Configs {	36✔
142	resolvedAttrs := r.resolveDeviceAttributes(config.DeviceAttributesSelector, allDeviceAttrs)	19✔
143		19✔
144	// If no resource filters are specified, all virtual function devices are candidates	19✔
145	if len(config.ResourceFilters) == 0 {	26✔
146	candidateDevices := r.deviceStateManager.GetPolicyCandidateDevices(false)	7✔
147	for deviceName := range candidateDevices {	21✔
148	if _, exists := policyDevices[deviceName]; exists {	16✔
149	continue	2✔
150	}
151	attrs := make(map[resourceapi.QualifiedName]resourceapi.DeviceAttribute, len(resolvedAttrs))	12✔
152	for k, v := range resolvedAttrs {	16✔
153	attrs[k] = v	4✔
154	}	4✔
155	policyDevices[deviceName] = attrs	12✔
156	r.log.V(2).Info("Device matches config filter",	12✔
157	"deviceName", deviceName,	12✔
158	"policyName", policy.Name,	12✔
159	"attributes", attrs)	12✔
160	}
161	continue	7✔
162	}
163
164	for _, filter := range config.ResourceFilters {	25✔
165	candidateDevices := r.deviceStateManager.GetPolicyCandidateDevices(filterRequestsPciAddresses(filter))	13✔
166	for deviceName, device := range candidateDevices {	37✔
167	if _, exists := policyDevices[deviceName]; exists {	29✔
168	continue	5✔
169	}
170
171	if r.deviceMatchesFilter(device, filter) {	34✔
172	devicePciAddress, _ := deviceAttributeString(device, consts.AttributePciAddress)	15✔
173	interfaceType := getDeviceInterfaceType(device)	15✔
174	parentPfAddress, hasParentPfAddress := deviceAttributeString(device, consts.AttributePfPciAddress)	15✔
175		15✔
176	if interfaceType == consts.InterfaceTypeVirtualFunction && hasParentPfAddress {	18✔
177	if _, pfSelected := selectedPfPciAddresses[parentPfAddress]; pfSelected {	3✔
NEW 178	r.log.Info("Skipping VF because its parent PF is already selected by pciAddresses",	×
NEW 179	"deviceName", deviceName,	×
NEW 180	"parentPfPciAddress", parentPfAddress)	×
NEW 181	continue	×
182	}
183	}
184
185	attrs := make(map[resourceapi.QualifiedName]resourceapi.DeviceAttribute, len(resolvedAttrs))	15✔
186	for k, v := range resolvedAttrs {	16✔
187	attrs[k] = v	1✔
188	}	1✔
189	policyDevices[deviceName] = attrs	15✔
190	if interfaceType == consts.InterfaceTypeVirtualFunction && hasParentPfAddress {	18✔
191	if _, exists := selectedVFsByPfAddress[parentPfAddress]; !exists {	6✔
192	selectedVFsByPfAddress[parentPfAddress] = make(map[string]struct{})	3✔
193	}	3✔
194	selectedVFsByPfAddress[parentPfAddress][deviceName] = struct{}{}	3✔
195	}
196	if filterRequestsPciAddresses(filter) && interfaceType == consts.InterfaceTypeRegular && devicePciAddress != "" {	21✔
197	selectedPfPciAddresses[devicePciAddress] = struct{}{}	6✔
198	if vfDeviceNames, exists := selectedVFsByPfAddress[devicePciAddress]; exists {	8✔
199	for vfDeviceName := range vfDeviceNames {	4✔
200	delete(policyDevices, vfDeviceName)	2✔
201	r.log.Info("Removing VF because parent PF is selected by pciAddresses",	2✔
202	"vfDeviceName", vfDeviceName,	2✔
203	"parentPfPciAddress", devicePciAddress,	2✔
204	"selectedPfDeviceName", deviceName)	2✔
205	}	2✔
206	delete(selectedVFsByPfAddress, devicePciAddress)	2✔
207	}
208	}
209
210	r.log.V(2).Info("Device matches config filter",	15✔
211	"deviceName", deviceName,	15✔
212	"policyName", policy.Name,	15✔
213	"device", device,	15✔
214	"attributes", attrs)	15✔
215	}
216	}
217	}
218	}
219	}
220
221	r.log.Info("Policy devices resolved",	13✔
222	"matchingDevices", len(policyDevices))	13✔
223	r.log.V(2).Info("Policy devices details", "policyDevices", policyDevices)	13✔
224		13✔
225	return policyDevices	13✔
226	}
227
228	// filterRequestsPciAddresses reports whether a filter explicitly selects by PCI address.
229	func filterRequestsPciAddresses(filter sriovdrav1alpha1.ResourceFilter) bool {	34✔
230	return len(filter.PciAddresses) > 0	34✔
231	}	34✔
232
233	// getDeviceInterfaceType returns the interface type attribute with fallback inference.
234	func getDeviceInterfaceType(device resourceapi.Device) string {	15✔
235	if interfaceType, exists := deviceAttributeString(device, consts.AttributeInterfaceType); exists {	17✔
236	return interfaceType	2✔
237	}	2✔
238	if _, hasParentPf := deviceAttributeString(device, consts.AttributePfPciAddress); hasParentPf {	16✔
239	return consts.InterfaceTypeVirtualFunction	3✔
240	}	3✔
241	return consts.InterfaceTypeRegular	10✔
242	}
243
244	// deviceAttributeString reads a string-valued device attribute by qualified name.
245	func deviceAttributeString(device resourceapi.Device, name resourceapi.QualifiedName) (string, bool) {	58✔
246	if device.Attributes == nil {	58✔
NEW 247	return "", false	×
NEW 248	}	×
249	attr, exists := device.Attributes[name]	58✔
250	if !exists \|\| attr.StringValue == nil {	99✔
251	return "", false	41✔
252	}	41✔
253	return *attr.StringValue, true	17✔
254	}
255
256	// resolveDeviceAttributes finds all DeviceAttributes objects matching the
257	// given label selector and merges their attributes. When multiple objects
258	// match and define the same key, the value from the alphabetically last
259	// object name wins (deterministic).
260	func (r *SriovResourcePolicyReconciler) resolveDeviceAttributes(
261	selector *metav1.LabelSelector,
262	allDeviceAttrs []sriovdrav1alpha1.DeviceAttributes,
263	) map[resourceapi.QualifiedName]resourceapi.DeviceAttribute {	19✔
264	if selector == nil {	35✔
265	return nil	16✔
266	}	16✔
267
268	sel, err := metav1.LabelSelectorAsSelector(selector)	3✔
269	if err != nil {	3✔
270	r.log.Error(err, "Invalid DeviceAttributesSelector")	×
271	return nil	×
272	}	×
273
274	// Collect matching DeviceAttributes, sort by name for determinism
275	var matched []sriovdrav1alpha1.DeviceAttributes	3✔
276	for i := range allDeviceAttrs {	6✔
277	da := allDeviceAttrs[i]	3✔
278	if sel.Matches(labels.Set(da.Labels)) {	6✔
279	matched = append(matched, da)	3✔
280	}	3✔
281	}
282	sort.Slice(matched, func(i, j int) bool {	3✔
283	return matched[i].Name < matched[j].Name	×
284	})	×
285
286	merged := make(map[resourceapi.QualifiedName]resourceapi.DeviceAttribute)	3✔
287	for _, da := range matched {	6✔
288	for key, val := range da.Spec.Attributes {	6✔
289	merged[key] = val	3✔
290	}	3✔
291	}
292
293	return merged	3✔
294	}
295
296	// matchesNodeSelector checks if node labels match the given NodeSelector.
297	// A nil selector matches all nodes.
298	func (r SriovResourcePolicyReconciler) matchesNodeSelector(nodeLabels map[string]string, nodeSelector corev1.NodeSelector) bool {	13✔
299	if nodeSelector == nil \|\| len(nodeSelector.NodeSelectorTerms) == 0 {	23✔
300	return true	10✔
301	}	10✔
302
303	// NodeSelectorTerms are ORed: at least one must match.
304	for _, term := range nodeSelector.NodeSelectorTerms {	7✔
305	ls := nodeSelectorTermToLabelSelector(term)	4✔
306	selector, err := metav1.LabelSelectorAsSelector(ls)	4✔
307	if err != nil {	4✔
308	r.log.V(2).Info("Failed to parse NodeSelectorTerm", "error", err)	×
309	continue	×
310	}
311	if selector.Matches(labels.Set(nodeLabels)) {	6✔
312	return true	2✔
313	}	2✔
314	}
315	return false	1✔
316	}
317
318	func nodeSelectorTermToLabelSelector(term corev1.NodeSelectorTerm) *metav1.LabelSelector {	4✔
319	var exprs []metav1.LabelSelectorRequirement	4✔
320	for _, req := range term.MatchExpressions {	8✔
321	exprs = append(exprs, metav1.LabelSelectorRequirement{	4✔
322	Key: req.Key,	4✔
323	Operator: metav1.LabelSelectorOperator(req.Operator),	4✔
324	Values: req.Values,	4✔
325	})	4✔
326	}	4✔
327	return &metav1.LabelSelector{MatchExpressions: exprs}	4✔
328	}
329
330	// deviceMatchesFilter checks if a device matches a specific resource filter
331	func (r *SriovResourcePolicyReconciler) deviceMatchesFilter(device resourceapi.Device, filter sriovdrav1alpha1.ResourceFilter) bool {	28✔
332	if len(filter.Vendors) > 0 {	39✔
333	vendorAttr, exists := device.Attributes[consts.AttributeVendorID]	11✔
334	if !exists \|\| vendorAttr.StringValue == nil {	11✔
335	return false	×
336	}	×
337	if !stringSliceContains(filter.Vendors, *vendorAttr.StringValue) {	12✔
338	return false	1✔
339	}	1✔
340	}
341
342	if len(filter.Devices) > 0 {	29✔
343	deviceAttr, exists := device.Attributes[consts.AttributeDeviceID]	2✔
344	if !exists \|\| deviceAttr.StringValue == nil {	2✔
345	return false	×
346	}	×
347	if !stringSliceContains(filter.Devices, *deviceAttr.StringValue) {	3✔
348	return false	1✔
349	}	1✔
350	}
351
352	if len(filter.PciAddresses) > 0 {	40✔
353	pciAttr, exists := device.Attributes[consts.AttributePciAddress]	14✔
354	if !exists \|\| pciAttr.StringValue == nil {	16✔
355	return false	2✔
356	}	2✔
357	if !stringSliceContains(filter.PciAddresses, *pciAttr.StringValue) {	17✔
358	return false	5✔
359	}	5✔
360	}
361
362	if len(filter.PfNames) > 0 {	21✔
363	pfAttr, exists := device.Attributes[consts.AttributePFName]	2✔
364	if !exists \|\| pfAttr.StringValue == nil {	2✔
365	return false	×
366	}	×
367	if !stringSliceContains(filter.PfNames, *pfAttr.StringValue) {	3✔
368	return false	1✔
369	}	1✔
370	}
371
372	if len(filter.PfPciAddresses) > 0 {	20✔
373	parentAttr, exists := device.Attributes[consts.AttributePfPciAddress]	2✔
374	if !exists \|\| parentAttr.StringValue == nil {	2✔
375	return false	×
376	}	×
377	if !stringSliceContains(filter.PfPciAddresses, *parentAttr.StringValue) {	3✔
378	return false	1✔
379	}	1✔
380	}
381
382	// TODO: Implement driver checking if needed
383	if len(filter.Drivers) > 0 {	17✔
384	r.log.V(3).Info("Driver filtering not yet implemented", "deviceName", device.Name)	×
385	}	×
386
387	return true	17✔
388	}
389
390	func stringSliceContains(slice []string, item string) bool {	31✔
391	for _, s := range slice {	64✔
392	if s == item {	54✔
393	return true	21✔
394	}	21✔
395	}
396	return false	10✔
397	}
398
399	// SetupWithManager sets up the controller with the Manager.
400	func (r *SriovResourcePolicyReconciler) SetupWithManager(mgr ctrl.Manager) error {	1✔
401	qHandler := func(q workqueue.TypedRateLimitingInterface[reconcile.Request]) {	8✔
402	q.AddAfter(reconcile.Request{NamespacedName: types.NamespacedName{	7✔
403	Namespace: r.namespace,	7✔
404	Name: resourcePolicySyncEventName,	7✔
405	}}, time.Second)	7✔
406	}	7✔
407
408	delayedEventHandler := handler.Funcs{	1✔
409	CreateFunc: func(_ context.Context, e event.TypedCreateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {	6✔
410	r.log.Info("Enqueuing sync for create event",	5✔
411	"resource", e.Object.GetName(),	5✔
412	"type", e.Object.GetObjectKind().GroupVersionKind().String())	5✔
413	qHandler(w)	5✔
414	},	5✔
415	UpdateFunc: func(_ context.Context, e event.TypedUpdateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {	×
416	r.log.Info("Enqueuing sync for update event",	×
417	"resource", e.ObjectNew.GetName(),	×
418	"type", e.ObjectNew.GetObjectKind().GroupVersionKind().String())	×
419	qHandler(w)	×
420	},	×
421	DeleteFunc: func(_ context.Context, e event.TypedDeleteEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {	2✔
422	r.log.Info("Enqueuing sync for delete event",	2✔
423	"resource", e.Object.GetName(),	2✔
424	"type", e.Object.GetObjectKind().GroupVersionKind().String())	2✔
425	qHandler(w)	2✔
426	},	2✔
427	GenericFunc: func(_ context.Context, e event.TypedGenericEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {	×
428	r.log.Info("Enqueuing sync for generic event",	×
429	"resource", e.Object.GetName(),	×
430	"type", e.Object.GetObjectKind().GroupVersionKind().String())	×
431	qHandler(w)	×
432	},	×
433	}
434
435	nodeEventHandler := handler.Funcs{	1✔
436	CreateFunc: func(_ context.Context, e event.TypedCreateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {	1✔
437	if e.Object.GetName() == r.nodeName {	×
438	r.log.Info("Enqueuing sync for node create event", "node", e.Object.GetName())	×
439	qHandler(w)	×
440	}	×
441	},
442	UpdateFunc: func(_ context.Context, e event.TypedUpdateEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {	×
443	if e.ObjectNew.GetName() == r.nodeName {	×
444	oldLabels := e.ObjectOld.GetLabels()	×
445	newLabels := e.ObjectNew.GetLabels()	×
446	if !labels.Equals(oldLabels, newLabels) {	×
447	r.log.Info("Enqueuing sync for node label change event", "node", e.ObjectNew.GetName())	×
448	qHandler(w)	×
449	}	×
450	}
451	},
452	DeleteFunc: func(_ context.Context, e event.TypedDeleteEvent[client.Object], w workqueue.TypedRateLimitingInterface[reconcile.Request]) {	×
453	if e.Object.GetName() == r.nodeName {	×
454	r.log.Info("Enqueuing sync for node delete event", "node", e.Object.GetName())	×
455	qHandler(w)	×
456	}	×
457	},
458	}
459
460	var eventChan = make(chan event.GenericEvent, 1)	1✔
461	eventChan <- event.GenericEvent{Object: &sriovdrav1alpha1.SriovResourcePolicy{	1✔
462	ObjectMeta: metav1.ObjectMeta{Name: resourcePolicySyncEventName, Namespace: r.namespace}}}	1✔
463	close(eventChan)	1✔
464		1✔
465	namespacePredicate := predicate.NewPredicateFuncs(func(obj client.Object) bool {	17✔
466	return obj.GetNamespace() == r.namespace	16✔
467	})	16✔
468
469	nodeMetadata := &metav1.PartialObjectMetadata{}	1✔
470	nodeMetadata.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Node"))	1✔
471		1✔
472	return ctrl.NewControllerManagedBy(mgr).	1✔
473	For(&sriovdrav1alpha1.SriovResourcePolicy{}).	1✔
474	Watches(nodeMetadata, nodeEventHandler).	1✔
475	Watches(&sriovdrav1alpha1.SriovResourcePolicy{}, delayedEventHandler).	1✔
476	Watches(&sriovdrav1alpha1.DeviceAttributes{}, delayedEventHandler).	1✔
477	WithEventFilter(namespacePredicate).	1✔
478	WatchesRawSource(source.Channel(eventChan, &handler.EnqueueRequestForObject{})).	1✔
479	Complete(r)	1✔
480	}

k8snetworkplumbingwg / dra-driver-sriov / 25047722054

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