Skip to content
56 changes: 56 additions & 0 deletions sunbeam-python/sunbeam/provider/common/nic_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,62 @@ def whitelist_sriov_nic(
)


def whitelist_remote_managed_vf(
node_name: str,
vf_nic: dict,
pci_whitelist: list[dict],
excluded_devices: dict[str, list],
physnet: str | None,
):
"""Whitelist a DPU-backed SR-IOV virtual function as a remote-managed device.

For off-path SmartNIC DPUs the networking control plane runs on the DPU,
while Nova on the hypervisor host claims the virtual functions (VFs). Nova
matches these VFs by PCI ``address``, ``vendor_id``/``product_id`` and
requires the ``remote_managed`` tag. The MAAS physnet tag only selects the
PF; remote-managed VF specs always use a null physical network.

https://docs.openstack.org/neutron/2024.1/admin/ovn/smartnic_dpu.html
"""
LOG.debug(
"Whitelisting remote-managed VF: %s %s",
vf_nic.get("name"),
vf_nic.get("pci_address"),
)
pci_address = vf_nic["pci_address"]

node_excluded_devices = excluded_devices.get(node_name) or []
if pci_address in node_excluded_devices:
LOG.debug(
"Removing remote-managed VF from the exclusion list: %s %s",
vf_nic.get("name"),
pci_address,
)
node_excluded_devices.remove(pci_address)

physnet = None

new_dev_spec = {
"address": pci_address,
"vendor_id": vf_nic["vendor_id"].replace("0x", ""),
"product_id": vf_nic["product_id"].replace("0x", ""),
"physical_network": physnet,
"remote_managed": "true",
}

# Avoid exact duplicates if the same VF is encountered more than once.
if new_dev_spec in pci_whitelist:
LOG.debug(
"Remote-managed VF spec already whitelisted: %s/%s (physnet: %s)",
new_dev_spec["vendor_id"],
new_dev_spec["product_id"],
physnet,
)
return

pci_whitelist.append(new_dev_spec)


def exclude_sriov_nic(
node_name: str,
nic: dict,
Expand Down
6 changes: 6 additions & 0 deletions sunbeam-python/sunbeam/provider/maas/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -335,6 +335,11 @@ def _convert_raw_machine(machine_raw: dict, root_disk: dict | None) -> dict:
architecture = raw_arch.split("/")[0] if raw_arch else DEFAULT_ARCHITECTURE

is_dpu = bool(machine_raw.get("is_dpu", False))
# For off-path SmartNIC DPUs, MAAS models the DPU machine as a child of
# its hypervisor host. The `parent` field is a dict holding the host's
# system_id (or None for regular machines). We surface it so that the
# SR-IOV configuration can pair a compute host with its DPU.
parent_system_id = (machine_raw.get("parent") or {}).get("system_id")
tag_names = machine_raw.get("tag_names") or []
image_name = parse_image_name_from_tags(tag_names)

Expand All @@ -352,6 +357,7 @@ def _convert_raw_machine(machine_raw: dict, root_disk: dict | None) -> dict:
"memory": machine_raw["memory"],
"architecture": architecture,
"is_dpu": is_dpu,
"parent_system_id": parent_system_id,
}
if image_name:
machine["image_name"] = image_name
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
# SPDX-FileCopyrightText: 2026 - Canonical Ltd
# SPDX-License-Identifier: Apache-2.0

from sunbeam.provider.common import nic_utils


class TestWhitelistRemoteManagedVf:
def _vf(self, pci_address="0000:41:00.3", vendor="0x15b3", product="0x101e"):
return {
"name": "ens1f0v0",
"pci_address": pci_address,
"vendor_id": vendor,
"product_id": product,
"pf_pci_address": "0000:41:00.0",
"sriov_available": False,
}

def test_builds_remote_managed_spec_with_vf_address_and_null_physnet(self):
pci_whitelist: list[dict] = []
excluded_devices: dict[str, list] = {}

nic_utils.whitelist_remote_managed_vf(
"compute-1", self._vf(), pci_whitelist, excluded_devices, "physnet1"
)

assert pci_whitelist == [
{
"address": "0000:41:00.3",
"vendor_id": "15b3",
"product_id": "101e",
"physical_network": None,
"remote_managed": "true",
}
]

def test_keeps_distinct_vf_addresses(self):
pci_whitelist: list[dict] = []
excluded_devices: dict[str, list] = {}

for func in (3, 4, 5, 6):
nic_utils.whitelist_remote_managed_vf(
"compute-1",
self._vf(pci_address=f"0000:41:00.{func}"),
pci_whitelist,
excluded_devices,
"physnet1",
)

assert [spec["address"] for spec in pci_whitelist] == [
"0000:41:00.3",
"0000:41:00.4",
"0000:41:00.5",
"0000:41:00.6",
]

def test_ignores_maas_physnet_for_remote_managed_vfs(self):
pci_whitelist: list[dict] = []
excluded_devices: dict[str, list] = {}

nic_utils.whitelist_remote_managed_vf(
"compute-1", self._vf(), pci_whitelist, excluded_devices, "physnet1"
)

assert pci_whitelist[0]["physical_network"] is None

def test_removes_vf_from_exclusion_list(self):
pci_whitelist: list[dict] = []
excluded_devices: dict[str, list] = {"compute-1": ["0000:41:00.3"]}

nic_utils.whitelist_remote_managed_vf(
"compute-1", self._vf(), pci_whitelist, excluded_devices, "physnet1"
)

assert "0000:41:00.3" not in excluded_devices["compute-1"]
Loading