feat: dra dynamic labels

images/virtualization-artifact/pkg/common/annotations/annotations.go

@@ -218,6 +218,10 @@ const (
 	// AnnUSBDeviceUser is the annotation for device user (owner) in ResourceClaimTemplate.
 	AnnUSBDeviceUser = "usb.virtualization.deckhouse.io/device-user"
 
+	AnnUSBIPTotalPorts = "usb.virtualization.deckhouse.io/usbip-total-ports"
+	AnnUSBIPUsedPorts  = "usb.virtualization.deckhouse.io/usbip-used-ports"
+	AnnUSBIPAddress    = "usb.virtualization.deckhouse.io/usbip-address"
+
 	// DefaultUSBDeviceGroup is the default device group ID for USB devices.
 	DefaultUSBDeviceGroup = "107"
 	// DefaultUSBDeviceUser is the default device user ID for USB devices.

images/virtualization-artifact/pkg/controller/vm/internal/validators/usb_devices_validator.go

@@ -19,11 +19,15 @@ package validators
 import (
 	"context"
 	"fmt"
+	"strconv"
+	"strings"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
+	"github.com/deckhouse/virtualization-controller/pkg/common/annotations"
 	"github.com/deckhouse/virtualization-controller/pkg/controller/indexer"
 	"github.com/deckhouse/virtualization/api/core/v1alpha2"
 )
@@ -37,20 +41,38 @@ func NewUSBDevicesValidator(client client.Client) *USBDevicesValidator {
 }
 
 func (v *USBDevicesValidator) ValidateCreate(ctx context.Context, vm *v1alpha2.VirtualMachine) (admission.Warnings, error) {
-	return v.validateUSBDevicesUnique(ctx, vm, "", nil)
+	return v.validateUSBDevicesUnique(ctx, vm, nil)
 }
 
 func (v *USBDevicesValidator) ValidateUpdate(ctx context.Context, oldVM, newVM *v1alpha2.VirtualMachine) (admission.Warnings, error) {
 	if equality.Semantic.DeepEqual(oldVM.Spec.USBDevices, newVM.Spec.USBDevices) {
 		return nil, nil
 	}
 
-	return v.validateUSBDevicesUnique(ctx, newVM, newVM.Name, getUSBDeviceNames(oldVM.Spec.USBDevices))
+	oldUsbDevices := getUSBDeviceNames(oldVM.Spec.USBDevices)
+
+	var allWarnings admission.Warnings
+
+	if warnings, err := v.validateUSBDevicesUnique(ctx, newVM, oldUsbDevices); err != nil {
+		allWarnings = append(allWarnings, warnings...)
+		return allWarnings, err
+	} else {
+		allWarnings = append(allWarnings, warnings...)
+	}
+
+	if warnings, err := v.validateAvailableUSBIPPorts(ctx, newVM, oldUsbDevices); err != nil {
+		allWarnings = append(allWarnings, warnings...)
+		return allWarnings, err
+	} else {
+		allWarnings = append(allWarnings, warnings...)
+	}
+
+	return allWarnings, nil
 }
 
 // validateUSBDevicesUnique checks that each USB device is not used by another VM.
 // currentVMName is empty for Create (no VM to exclude), or VM name for Update (exclude current VM from conflict check).
-func (v *USBDevicesValidator) validateUSBDevicesUnique(ctx context.Context, vm *v1alpha2.VirtualMachine, currentVMName string, oldUSBDevices map[string]struct{}) (admission.Warnings, error) {
+func (v *USBDevicesValidator) validateUSBDevicesUnique(ctx context.Context, vm *v1alpha2.VirtualMachine, oldUSBDevices map[string]struct{}) (admission.Warnings, error) {
 	if len(vm.Spec.USBDevices) == 0 {
 		return nil, nil
 	}
@@ -76,7 +98,7 @@ func (v *USBDevicesValidator) validateUSBDevicesUnique(ctx context.Context, vm *
 
 		for i := range vmList.Items {
 			otherVM := &vmList.Items[i]
-			if otherVM.Name == currentVMName {
+			if otherVM.Name == vm.Name {
 				continue
 			}
 			return nil, fmt.Errorf("USB device %s is already used by VirtualMachine %s/%s", ref.Name, otherVM.Namespace, otherVM.Name)
@@ -97,3 +119,69 @@ func getUSBDeviceNames(refs []v1alpha2.USBDeviceSpecRef) map[string]struct{} {
 
 	return names
 }
+
+func (v *USBDevicesValidator) validateAvailableUSBIPPorts(ctx context.Context, vm *v1alpha2.VirtualMachine, oldUSBDevices map[string]struct{}) (admission.Warnings, error) {
+	if vm.Status.Node == "" {
+		return admission.Warnings{}, nil
+	}
+	if vm.Spec.USBDevices == nil {
+		return admission.Warnings{}, nil
+	}
+
+	var usbFromOtherNodes []string
+
+	for _, ref := range vm.Spec.USBDevices {
+		if _, exists := oldUSBDevices[ref.Name]; exists {
+			continue
+		}
+
+		usbDevice := &v1alpha2.USBDevice{}
+		err := v.client.Get(ctx, client.ObjectKey{Name: ref.Name, Namespace: vm.Namespace}, usbDevice)
+		if err != nil {
+			return admission.Warnings{}, fmt.Errorf("failed to get USB device %s: %w", ref.Name, err)
+		}
+
+		if usbDevice.Status.NodeName != vm.Status.Node {
+			usbFromOtherNodes = append(usbFromOtherNodes, ref.Name)
+		}
+	}
+
+	if len(usbFromOtherNodes) == 0 {
+		return admission.Warnings{}, nil
+	}
+
+	node := &corev1.Node{}
+	err := v.client.Get(ctx, client.ObjectKey{Name: vm.Status.Node}, node)
+	if err != nil {
+		return admission.Warnings{}, fmt.Errorf("failed to get node %s: %w", vm.Status.Node, err)
+	}
+
+	totalPorts, exists := node.Annotations[annotations.AnnUSBIPTotalPorts]
+	if !exists {
+		return admission.Warnings{}, fmt.Errorf("node %s does not have %s annotation", vm.Status.Node, annotations.AnnUSBIPTotalPorts)
+	}
+	totalPortsInt, err := strconv.Atoi(totalPorts)
+	if err != nil {
+		return admission.Warnings{}, fmt.Errorf("failed to parse %s annotation: %w", annotations.AnnUSBIPTotalPorts, err)
+	}
+
+	// total for 2 usb hubs (2.0 and 3.0)
+	totalPortsInt /= 2
+
+	usedPorts, exists := node.Annotations[annotations.AnnUSBIPUsedPorts]
+	if !exists {
+		return admission.Warnings{}, fmt.Errorf("node %s does not have %s annotation", vm.Status.Node, annotations.AnnUSBIPUsedPorts)
+	}
+	usedPortsInt, err := strconv.Atoi(usedPorts)
+	if err != nil {
+		return admission.Warnings{}, fmt.Errorf("failed to parse %s annotation: %w", annotations.AnnUSBIPUsedPorts, err)
+	}
+
+	wantedPorts := usedPortsInt + len(usbFromOtherNodes)
+
+	if wantedPorts > totalPortsInt {
+		return admission.Warnings{}, fmt.Errorf("node %s not available ports for sharing USB devices %s. total: %d, used: %d, wanted: %d", vm.Status.Node, strings.Join(usbFromOtherNodes, ", "), totalPortsInt, usedPortsInt, wantedPorts)
+	}
+
+	return admission.Warnings{}, nil
+}

images/virtualization-dra-usb/werf.inc.yaml

@@ -10,7 +10,7 @@ import:
     after: install
   - image: {{ .ModuleNamePrefix }}virtualization-dra-builder
     add: /out/go-usbip
-    to: /usb/bin/go-usbip
+    to: /usr/bin/go-usbip
     after: install
   {{- if eq $.DEBUG_COMPONENT "delve/virtualization-dra-usb" }}
   - image: debugger

images/virtualization-dra/cmd/usb/dra/app/app.go

@@ -70,13 +70,11 @@ func NewVirtualizationDraUSBCommand() *cobra.Command {
 }
 
 type draOptions struct {
-	DriverName           string
-	Kubeconfig           string
-	Namespace            string
-	NodeName             string
-	USBGatewaySecretName string
-	CDIRoot              string
-	HealthzPort          int
+	DriverName  string
+	Kubeconfig  string
+	NodeName    string
+	CDIRoot     string
+	HealthzPort int
 
 	logging      *logger.Options
 	monitor      *libusb.MonitorConfig
@@ -102,9 +100,7 @@ func (o *draOptions) NamedFlags() (fs flag.NamedFlagSets) {
 	mfs := fs.FlagSet("virtualization-usb plugin")
 	mfs.StringVar(&o.DriverName, "driver-name", usb.DriverName, "Driver name")
 	mfs.StringVar(&o.Kubeconfig, "kubeconfig", cli.GetStringEnv("KUBECONFIG", ""), "Path to kubeconfig file")
-	mfs.StringVar(&o.Namespace, "namespace", cli.GetStringEnv("NAMESPACE", ""), "Namespace")
 	mfs.StringVar(&o.NodeName, "node-name", cli.GetStringEnv("NODE_NAME", ""), "Node name")
-	mfs.StringVar(&o.USBGatewaySecretName, "usb-gateway-secret-name", cli.GetStringEnv("USB_GATEWAY_SECRET_NAME", "virtualization-dra-usb-gateway"), "USB gateway secret name")
 	mfs.StringVar(&o.CDIRoot, "cdi-root", cli.GetStringEnv("CDI_ROOT", cdi.SpecDir), "CDI root")
 	mfs.IntVar(&o.HealthzPort, "healthz-port", cli.GetIntEnv("HEALTHZ_PORT", 51515), "Healthz port")
 
@@ -118,22 +114,13 @@ func (o *draOptions) NamedFlags() (fs flag.NamedFlagSets) {
 }
 
 func (o *draOptions) Validate() error {
-	if o.Namespace == "" {
-		return fmt.Errorf("namespace is required")
-	}
 	if o.NodeName == "" {
 		return fmt.Errorf("nodeName is required")
 	}
 	if o.CDIRoot == "" {
 		return fmt.Errorf("cdiRoot is required")
 	}
 
-	if o.usbGatewayEnabled {
-		if o.USBGatewaySecretName == "" {
-			return fmt.Errorf("USBGatewaySecretName is required")
-		}
-	}
-
 	return nil
 }
 
@@ -180,7 +167,7 @@ func (o *draOptions) Run(cmd *cobra.Command, _ []string) error {
 		}
 
 		f := informer.NewFactory(client, nil)
-		secretInformer := f.NamespacedSecret(o.Namespace)
+		nodeInformer := f.Nodes()
 		resourceSliceInformer := f.ResourceSlice()
 
 		group.Go(func() error {
@@ -189,14 +176,12 @@ func (o *draOptions) Run(cmd *cobra.Command, _ []string) error {
 		f.WaitForCacheSync(ctx.Done())
 
 		usbGatewayController, err := usbgateway.NewUSBGatewayController(
-			ctx,
-			o.USBGatewaySecretName,
-			o.Namespace,
 			o.NodeName,
 			o.usbipdConfig.Address,
 			o.usbipdConfig.Port,
 			client,
-			secretInformer,
+			dynamicClient,
+			nodeInformer,
 			resourceSliceInformer,
 			usbip.New(),
 		)
@@ -212,16 +197,6 @@ func (o *draOptions) Run(cmd *cobra.Command, _ []string) error {
 			return controller.Run(usbGatewayController, ctx, 1)
 		})
 
-		marker := usbgateway.NewMarker(dynamicClient, o.NodeName)
-		if err = marker.Mark(ctx); err != nil {
-			return err
-		}
-		defer func() {
-			if err = marker.Unmark(ctx); err != nil {
-				slog.Error("failed to unmark node for USB gateway", slog.Any("error", err))
-			}
-		}()
-
 		usbGateway = usbGatewayController
 	}
 

images/virtualization-dra/cmd/usb/go-usbip/app/attach-info.go

@@ -18,6 +18,7 @@ package app
 
 import (
 	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
 
 	"github.com/deckhouse/virtualization-dra/pkg/usbip"
 )
@@ -31,22 +32,52 @@ func NewAttachInfoCommand() *cobra.Command {
 		RunE:    o.Run,
 	}
 
+	o.AddFlags(cmd.Flags())
+
 	return cmd
 }
 
-type attachInfoOptions struct{}
+type attachInfoOptions struct {
+	watch bool
+}
 
 func (o *attachInfoOptions) Usage() string {
 	return `  # Get attach info
   $ go-usbip attach-info
 `
 }
 
+func (o *attachInfoOptions) AddFlags(fs *pflag.FlagSet) {
+	fs.BoolVarP(&o.watch, "watch", "w", false, "Watch attach info")
+}
+
 func (o *attachInfoOptions) Run(cmd *cobra.Command, _ []string) error {
-	infos, err := usbip.NewUSBAttacher().GetAttachInfo()
+	if o.watch {
+		return o.handleWatch(cmd)
+	}
+	return o.handleGet(cmd)
+}
+
+func (o *attachInfoOptions) handleGet(cmd *cobra.Command) error {
+	info, err := usbip.NewUSBAttacher().GetAttachInfo()
 	if err != nil {
 		return err
 	}
 
-	return printer.PrintObject(cmd, infos)
+	return printer.PrintObject(cmd, info)
+}
+
+func (o *attachInfoOptions) handleWatch(cmd *cobra.Command) error {
+	ch, err := usbip.NewUSBAttacher().WatchAttachInfo(cmd.Context())
+	if err != nil {
+		return err
+	}
+
+	for info := range ch {
+		if err := printer.PrintObject(cmd, info); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }

images/virtualization-dra/internal/consts/consts.go

@@ -28,4 +28,7 @@ const (
 	AnnUSBDeviceAddresses = "usb.virtualization.deckhouse.io/device-addresses"
 	AnnUSBDeviceUser      = "usb.virtualization.deckhouse.io/device-user"
 	AnnUSBDeviceGroup     = "usb.virtualization.deckhouse.io/device-group"
+	AnnUSBIPTotalPorts    = "usb.virtualization.deckhouse.io/usbip-total-ports"
+	AnnUSBIPUsedPorts     = "usb.virtualization.deckhouse.io/usbip-used-ports"
+	AnnUSBIPAddress       = "usb.virtualization.deckhouse.io/usbip-address"
 )

images/virtualization-dra/internal/plugin/wrapresourceslice/resourceslicecontroller.go

@@ -348,8 +348,8 @@ func (c *Controller) Update(resources *DriverResources) error {
 	} else {
 		if c.reconcileOnlyPoolName != "" {
 			_, ok := resources.Pools[c.reconcileOnlyPoolName]
-			if !ok && len(resources.Pools) > 1 {
-				return fmt.Errorf("reconcileOnlyPoolName is set to %q, but multiple pools found (%d total)",
+			if (ok && len(resources.Pools) > 1) || !ok && len(resources.Pools) > 0 {
+				return fmt.Errorf("reconcileOnlyPoolName is set to %q, but other pools found (%d total)",
 					c.reconcileOnlyPoolName, len(resources.Pools))
 			}
 		}
@@ -486,13 +486,14 @@ func (c *Controller) initInformer(ctx context.Context) error {
 				}
 
 				if c.reconcileOnlyPoolName != "" {
+					var newItems []resourcev1.ResourceSlice
 					for i := range slices.Items {
 						if slices.Items[i].Spec.Pool.Name == c.reconcileOnlyPoolName {
-							slices.Items = []resourcev1.ResourceSlice{slices.Items[i]}
-							return slices, nil
+							newItems = append(newItems, slices.Items[i])
+							break
 						}
 					}
-					slices.Items = nil
+					slices.Items = newItems
 					return slices, nil
 				}
 
@@ -812,7 +813,7 @@ func (c *Controller) syncPool(ctx context.Context, poolName string) error {
 		//
 		// When adding new fields here, then also extend sliceStored.
 		slice.Spec.NodeSelector = pool.NodeSelector
-		slice.Spec.AllNodes = refIfNotZero(desiredAllNodes)
+		slice.Spec.AllNodes = refIfNotZero(desiredAllNodes && (pool.Slices[i].PerDeviceNodeSelection == nil || !*pool.Slices[i].PerDeviceNodeSelection))
 		slice.Spec.SharedCounters = pool.Slices[i].SharedCounters
 		slice.Spec.PerDeviceNodeSelection = pool.Slices[i].PerDeviceNodeSelection
 		// Preserve TimeAdded from existing device, if there is a matching device and taint.
@@ -860,7 +861,7 @@ func (c *Controller) syncPool(ctx context.Context, poolName string) error {
 				Pool:                   desiredPool,
 				NodeName:               refIfNotZero(nodeName),
 				NodeSelector:           pool.NodeSelector,
-				AllNodes:               refIfNotZero(desiredAllNodes),
+				AllNodes:               refIfNotZero(desiredAllNodes && (pool.Slices[i].PerDeviceNodeSelection == nil || !*pool.Slices[i].PerDeviceNodeSelection)),
 				Devices:                pool.Slices[i].Devices,
 				SharedCounters:         pool.Slices[i].SharedCounters,
 				PerDeviceNodeSelection: pool.Slices[i].PerDeviceNodeSelection,

images/virtualization-dra/internal/usb-gateway/attach_record.go

@@ -98,14 +98,14 @@ func (r *attachRecordManager) Refresh() error {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
-	infos, err := r.getter.GetAttachInfo()
+	info, err := r.getter.GetAttachInfo()
 	if err != nil {
 		return err
 	}
 
-	ports := make(map[int]struct{}, len(infos))
-	for _, info := range infos {
-		ports[info.Port] = struct{}{}
+	ports := make(map[int]struct{}, len(info.Items))
+	for _, item := range info.Items {
+		ports[item.Port] = struct{}{}
 	}
 
 	b, err := os.ReadFile(r.recordFile)