feat(opennebula): support PCIx_* passthrough network interfaces

cloudinit/sources/DataSourceOpenNebula.py

@@ -252,6 +252,31 @@ def get_routes(self, dev: str) -> List[Dict[str, str]]:
                 )
         return routes
 
+    # Overridable for tests
+    _pci_sysfs_root: str = "/sys/bus/pci/devices"
+
+    def _get_pci_context_ifaces(self) -> List[str]:
+        """Return sorted PCIx prefixes that have an ADDRESS variable."""
+        seen = set()
+        for key in self.context:
+            m = re.match(r"^(PCI\d+)_ADDRESS$", key)
+            if m:
+                seen.add(m.group(1))
+        return sorted(seen)
+
+    def _pci_addr_to_dev(self, pci_address: str) -> Optional[str]:
+        """Map a PCI address (e.g. 0000:00:06.0) to a network device name.
+
+        Returns the first entry found under the sysfs net/ directory,
+        or None if the path does not exist or is empty.
+        """
+        sysfs = os.path.join(self._pci_sysfs_root, pci_address, "net")
+        try:
+            devs = os.listdir(sysfs)
+            return devs[0] if devs else None
+        except OSError:
+            return None
+
     @overload
     def get_field(self, dev: str, name: str) -> Optional[str]: ...
     @overload
@@ -336,6 +361,44 @@ def gen_conf(self) -> Dict[str, Any]:
             ethernets[dev] = devconf
 
         netconf["ethernets"] = ethernets
+
+        # Configure PCI passthrough interfaces (PCIx_ADDRESS / PCIx_IP …)
+        for pci_prefix in self._get_pci_context_ifaces():
+            pci_addr: Optional[str] = self.get_field(pci_prefix, "address")
+            if not pci_addr:
+                continue
+            pci_dev: Optional[str] = self._pci_addr_to_dev(pci_addr)
+            if not pci_dev:
+                LOG.warning(
+                    "Could not find netdev for PCI %s (%s), skipping",
+                    pci_prefix,
+                    pci_addr,
+                )
+                continue
+
+            pci_devconf: Dict[str, Any] = {}
+            ip = self.get_field(pci_prefix, "ip")
+            if ip:
+                mask = self.get_field(pci_prefix, "mask", "255.255.255.0")
+                prefix = str(net.ipv4_mask_to_net_prefix(mask))
+                pci_devconf["addresses"] = [ip + "/" + prefix]
+            gateway = self.get_field(pci_prefix, "gateway")
+            if gateway:
+                pci_devconf["gateway4"] = gateway
+            mtu = self.get_field(pci_prefix, "mtu")
+            if mtu:
+                pci_devconf["mtu"] = mtu
+
+            vlan_id: Optional[str] = self.get_field(pci_prefix, "vlan_id")
+            if vlan_id:
+                netconf["ethernets"][pci_dev] = {}
+                target_name = "%s.%s" % (pci_dev, vlan_id)
+                pci_devconf["id"] = int(vlan_id)
+                pci_devconf["link"] = pci_dev
+                netconf.setdefault("vlans", {})[target_name] = pci_devconf
+            else:
+                netconf["ethernets"][pci_dev] = pci_devconf
+
         return netconf
 
 

doc/rtd/reference/datasources/opennebula.rst

@@ -74,6 +74,20 @@ the OpenNebula documentation.
     ETH<x>_IP6_GATEWAY
     ETH<x>_ROUTES
 
+::
+
+    PCI<x>_ADDRESS
+    PCI<x>_IP
+    PCI<x>_MASK
+    PCI<x>_GATEWAY
+    PCI<x>_MTU
+    PCI<x>_VLAN_ID
+
+PCI passthrough network interfaces, identified by their PCI address
+(e.g. ``0000:00:06.0``). ``cloud-init`` resolves the address to a
+system device name via sysfs and applies the same IP configuration
+as ``ETH<x>_*`` interfaces. ``PCI<x>_VLAN_ID`` is also supported.
+
 Static `network configuration`_.
 
 ``ETH<x>_ROUTES`` is a comma-separated list of static routes in the form

tests/unittests/sources/test_opennebula.py

@@ -1045,6 +1045,94 @@ def test_gen_conf_no_routes_key_when_absent(self, m_get_phys_by_mac):
         conf = net.gen_conf()
         assert "routes" not in conf["ethernets"]["eth0"]
 
+    # --- PCIx_* passthrough interface tests ---
+
+    def test_get_pci_context_ifaces_none(self):
+        """No PCIx_ADDRESS keys → empty list."""
+        net = ds.OpenNebulaNetwork({}, mock.Mock())
+        assert net._get_pci_context_ifaces() == []
+
+    def test_get_pci_context_ifaces_single(self):
+        """Single PCI0_ADDRESS key → ['PCI0']."""
+        context = {"PCI0_ADDRESS": "0000:00:06.0"}
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        assert net._get_pci_context_ifaces() == ["PCI0"]
+
+    def test_get_pci_context_ifaces_multiple(self):
+        """Multiple PCIx_ADDRESS keys → sorted list."""
+        context = {
+            "PCI2_ADDRESS": "0000:00:08.0",
+            "PCI0_ADDRESS": "0000:00:06.0",
+            "PCI1_ADDRESS": "0000:00:07.0",
+        }
+        net = ds.OpenNebulaNetwork(context, mock.Mock())
+        assert net._get_pci_context_ifaces() == ["PCI0", "PCI1", "PCI2"]
+
+    def test_pci_addr_to_dev(self, tmp_path):
+        """_pci_addr_to_dev reads net/ dir under sysfs path."""
+        sysfs = tmp_path / "0000:00:06.0" / "net"
+        sysfs.mkdir(parents=True)
+        (sysfs / "enp0s6").mkdir()
+        with mock.patch.object(
+            ds.OpenNebulaNetwork,
+            "_pci_sysfs_root",
+            str(tmp_path),
+        ):
+            net = ds.OpenNebulaNetwork({}, mock.Mock())
+            assert net._pci_addr_to_dev("0000:00:06.0") == "enp0s6"
+
+    def test_pci_addr_to_dev_missing(self, tmp_path):
+        """_pci_addr_to_dev returns None when sysfs path is absent."""
+        with mock.patch.object(
+            ds.OpenNebulaNetwork,
+            "_pci_sysfs_root",
+            str(tmp_path),
+        ):
+            net = ds.OpenNebulaNetwork({}, mock.Mock())
+            assert net._pci_addr_to_dev("0000:00:99.0") is None
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_pci_static(self, m_get_phys_by_mac, tmp_path):
+        """PCI interface with static IP appears in ethernets output."""
+        sysfs = tmp_path / "0000:00:06.0" / "net"
+        sysfs.mkdir(parents=True)
+        (sysfs / "enp0s6").mkdir()
+        m_get_phys_by_mac.return_value = {}
+        context = {
+            "PCI0_ADDRESS": "0000:00:06.0",
+            "PCI0_IP": "10.5.0.1",
+            "PCI0_MASK": "255.255.255.0",
+        }
+        with mock.patch.object(
+            ds.OpenNebulaNetwork, "_pci_sysfs_root", str(tmp_path)
+        ):
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            conf = net.gen_conf()
+        assert "enp0s6" in conf["ethernets"]
+        assert "10.5.0.1/24" in conf["ethernets"]["enp0s6"]["addresses"]
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_pci_absent_no_output(self, m_get_phys_by_mac):
+        """No PCIx_ADDRESS → no extra entries in ethernets."""
+        m_get_phys_by_mac.return_value = {}
+        net = ds.OpenNebulaNetwork({}, mock.Mock())
+        conf = net.gen_conf()
+        assert conf["ethernets"] == {}
+
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_gen_conf_pci_sysfs_missing_skips(
+        self, m_get_phys_by_mac, tmp_path
+    ):
+        """PCI device with no sysfs entry is skipped with a warning."""
+        m_get_phys_by_mac.return_value = {}
+        context = {"PCI0_ADDRESS": "0000:00:99.0"}
+        with mock.patch.object(
+            ds.OpenNebulaNetwork, "_pci_sysfs_root", str(tmp_path)
+        ):
+            net = ds.OpenNebulaNetwork(context, mock.Mock())
+            conf = net.gen_conf()
+        assert conf["ethernets"] == {}
+
 
 class TestParseShellConfig:
     @pytest.mark.allow_subp_for("bash", "sh")