class IpCommandError(SriovNicError):
- message = _("ip command failed on device %(dev_name)s: %(reason)s")
+ message = _("ip command failed: %(reason)s")
class IpCommandOperationNotSupportedError(SriovNicError):
class InvalidPciSlotError(SriovNicError):
message = _("Invalid pci slot %(pci_slot)s")
+
+
+class IpCommandDeviceError(SriovNicError):
+ message = _("ip command failed on device %(dev_name)s: %(reason)s")
# See the License for the specific language governing permissions and
# limitations under the License.
-import glob
import os
import re
PCI_PATH = "/sys/class/net/%s/device/virtfn%s/net"
VIRTFN_FORMAT = r"^virtfn(?P<vf_index>\d+)"
VIRTFN_REG_EX = re.compile(VIRTFN_FORMAT)
- MAC_VTAP_PREFIX = "upper_macvtap*"
@classmethod
def scan_vf_devices(cls, dev_name):
by checking the relevant path in the system:
VF is assigned if:
Direct VF: PCI_PATH does not exist.
- Macvtap VF: upper_macvtap path exists.
+ Macvtap VF: macvtap@<vf interface> interface exists in ip link show
@param dev_name: pf network device name
@param vf_index: vf index
"""
path = cls.PCI_PATH % (dev_name, vf_index)
- if not os.path.isdir(path):
+
+ try:
+ ifname_list = os.listdir(path)
+ except OSError:
+ # PCI_PATH does not exist means that the DIRECT VF assigend
return True
- upper_macvtap_path = os.path.join(path, "*", cls.MAC_VTAP_PREFIX)
- return bool(glob.glob(upper_macvtap_path))
+
+ # Note(moshele) kernel < 3.13 doesn't create symbolic link
+ # for macvtap interface. Therefore we workaround it
+ # by parsing ip link show and checking if macvtap interface exists
+ for ifname in ifname_list:
+ if pci_lib.PciDeviceIPWrapper.is_macvtap_assigned(ifname):
+ return True
+ return False
class EmbSwitch(object):
MAC_PATTERN = r"MAC\s+(?P<mac>[a-fA-F0-9:]+),"
STATE_PATTERN = r"\s+link-state\s+(?P<state>\w+)"
ANY_PATTERN = ".*,"
+ MACVTAP_PATTERN = r".*macvtap[0-9]+@(?P<vf_interface>[a-zA-Z0-9]+):"
VF_LINE_FORMAT = VF_PATTERN + MAC_PATTERN + ANY_PATTERN + STATE_PATTERN
VF_DETAILS_REG_EX = re.compile(VF_LINE_FORMAT)
+ MACVTAP_REG_EX = re.compile(MACVTAP_PATTERN)
IP_LINK_OP_NOT_SUPPORTED = 'RTNETLINK answers: Operation not supported'
raise exc.IpCommandOperationNotSupportedError(
dev_name=self.dev_name)
else:
- raise exc.IpCommandError(dev_name=self.dev_name,
- reason=str(e))
+ raise exc.IpCommandDeviceError(dev_name=self.dev_name,
+ reason=str(e))
def get_assigned_macs(self, vf_list):
"""Get assigned mac addresses for vf list.
out = self._as_root([], "link", ("show", self.dev_name))
except Exception as e:
LOG.exception(_LE("Failed executing ip command"))
- raise exc.IpCommandError(dev_name=self.dev_name,
- reason=e)
+ raise exc.IpCommandDeviceError(dev_name=self.dev_name,
+ reason=e)
vf_to_mac_mapping = {}
vf_lines = self._get_vf_link_show(vf_list, out)
if vf_lines:
out = self._as_root([], "link", ("show", self.dev_name))
except Exception as e:
LOG.exception(_LE("Failed executing ip command"))
- raise exc.IpCommandError(dev_name=self.dev_name,
- reason=e)
+ raise exc.IpCommandDeviceError(dev_name=self.dev_name,
+ reason=e)
vf_lines = self._get_vf_link_show([vf_index], out)
if vf_lines:
vf_details = self._parse_vf_link_show(vf_lines[0])
"for %(device)s"),
{'line': vf_line, 'device': self.dev_name})
return vf_details
+
+ @classmethod
+ def is_macvtap_assigned(cls, ifname):
+ """Check if vf has macvtap interface assigned
+
+ Parses the output of ip link show command and checks
+ if macvtap[0-9]+@<vf interface> regex matches the
+ output.
+ @param ifname: vf interface name
+ @return: True on match otherwise False
+ """
+ try:
+ out = cls._execute([], "link", ("show", ), run_as_root=True)
+ except Exception as e:
+ LOG.error(_LE("Failed executing ip command: %s"), e)
+ raise exc.IpCommandError(reason=e)
+
+ for line in out.splitlines():
+ pattern_match = cls.MACVTAP_REG_EX.match(line)
+ if pattern_match:
+ if ifname == pattern_match.group('vf_interface'):
+ return True
+ return False
import mock
import testtools
-
from neutron.plugins.ml2.drivers.mech_sriov.agent.common \
import exceptions as exc
from neutron.plugins.ml2.drivers.mech_sriov.agent import eswitch_manager as esm
esm.PciOsWrapper.scan_vf_devices,
self.DEV_NAME)
- def _mock_assign_vf(self, dir_exists):
- with mock.patch("os.path.isdir",
- return_value=dir_exists):
- result = esm.PciOsWrapper.is_assigned_vf(self.DEV_NAME,
- self.VF_INDEX)
- self.assertEqual(not dir_exists, result)
-
- def test_is_assigned_vf_true(self):
- self._mock_assign_vf(True)
-
- def test_is_assigned_vf_false(self):
- self._mock_assign_vf(False)
-
- def _mock_assign_vf_macvtap(self, macvtap_exists):
- def _glob(file_path):
- return ["upper_macvtap0"] if macvtap_exists else []
+ @mock.patch("os.listdir", side_effect=OSError())
+ def test_is_assigned_vf_true(self, *args):
+ self.assertTrue(esm.PciOsWrapper.is_assigned_vf(
+ self.DEV_NAME, self.VF_INDEX))
- with mock.patch("os.path.isdir", return_value=True),\
- mock.patch("glob.glob", side_effect=_glob):
- result = esm.PciOsWrapper.is_assigned_vf(self.DEV_NAME,
- self.VF_INDEX)
- self.assertEqual(macvtap_exists, result)
+ @mock.patch("os.listdir", return_value=[DEV_NAME, "eth1"])
+ @mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
+ "PciDeviceIPWrapper.is_macvtap_assigned", return_value=False)
+ def test_is_assigned_vf_false(self, *args):
+ self.assertFalse(esm.PciOsWrapper.is_assigned_vf(
+ self.DEV_NAME, self.VF_INDEX))
- def test_is_assigned_vf_macvtap_true(self):
- self._mock_assign_vf_macvtap(True)
+ @mock.patch("os.listdir", return_value=["eth0", "eth1"])
+ @mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
+ "PciDeviceIPWrapper.is_macvtap_assigned", return_value=True)
+ def test_is_assigned_vf_macvtap(
+ self, mock_is_macvtap_assigned, *args):
+ esm.PciOsWrapper.is_assigned_vf(self.DEV_NAME, self.VF_INDEX)
+ mock_is_macvtap_assigned.called_with(self.VF_INDEX, "eth0")
- def test_is_assigned_vf_macvtap_false(self):
- self._mock_assign_vf_macvtap(False)
+ @mock.patch("os.listdir", side_effect=OSError())
+ @mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
+ "PciDeviceIPWrapper.is_macvtap_assigned")
+ def test_is_assigned_vf_macvtap_failure(
+ self, mock_is_macvtap_assigned, *args):
+ esm.PciOsWrapper.is_assigned_vf(self.DEV_NAME, self.VF_INDEX)
+ self.assertFalse(mock_is_macvtap_assigned.called)
' checking off, link-state enable')
VF_LINK_SHOW = '\n'.join((PF_LINK_SHOW, PF_MAC, VF_0_LINK_SHOW,
VF_1_LINK_SHOW, VF_2_LINK_SHOW))
+ MACVTAP_LINK_SHOW = ('63: macvtap1@enp129s0f1: <BROADCAST,MULTICAST> mtu '
+ '1500 qdisc noop state DOWN mode DEFAULT group '
+ 'default qlen 500 link/ether 4a:9b:6d:de:65:b5 brd '
+ 'ff:ff:ff:ff:ff:ff')
+
+ IP_LINK_SHOW_WITH_MACVTAP = '\n'.join((VF_LINK_SHOW, MACVTAP_LINK_SHOW))
MAC_MAPPING = {
0: "fa:16:3e:b4:81:ac",
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception()
- self.assertRaises(exc.IpCommandError,
+ self.assertRaises(exc.IpCommandDeviceError,
self.pci_wrapper.get_assigned_macs,
[self.VF_INDEX])
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception()
- self.assertRaises(exc.IpCommandError,
+ self.assertRaises(exc.IpCommandDeviceError,
self.pci_wrapper.get_vf_state,
self.VF_INDEX)
with mock.patch.object(self.pci_wrapper,
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception()
- self.assertRaises(exc.IpCommandError,
+ self.assertRaises(exc.IpCommandDeviceError,
self.pci_wrapper.set_vf_state,
self.VF_INDEX,
True)
with mock.patch.object(self.pci_wrapper,
"_execute") as mock_exec:
mock_exec.side_effect = Exception()
- self.assertRaises(exc.IpCommandError,
+ self.assertRaises(exc.IpCommandDeviceError,
self.pci_wrapper.set_vf_spoofcheck,
self.VF_INDEX,
True)
with mock.patch.object(self.pci_wrapper,
"_execute") as mock_exec:
mock_exec.side_effect = Exception()
- self.assertRaises(exc.IpCommandError,
+ self.assertRaises(exc.IpCommandDeviceError,
self.pci_wrapper.set_vf_max_rate,
self.VF_INDEX,
1000)
self.pci_wrapper.set_vf_state,
self.VF_INDEX,
state=True)
+
+ def test_is_macvtap_assigned(self):
+ with mock.patch.object(pci_lib.PciDeviceIPWrapper,
+ "_execute") as mock_exec:
+ mock_exec.return_value = self.IP_LINK_SHOW_WITH_MACVTAP
+ self.assertTrue(
+ pci_lib.PciDeviceIPWrapper.is_macvtap_assigned('enp129s0f1'))
+
+ def test_is_macvtap_assigned_not_assigned(self):
+ with mock.patch.object(pci_lib.PciDeviceIPWrapper,
+ "_execute") as mock_exec:
+ mock_exec.return_value = self.IP_LINK_SHOW_WITH_MACVTAP
+ self.assertFalse(
+ pci_lib.PciDeviceIPWrapper.is_macvtap_assigned('enp129s0f2'))
+
+ def test_is_macvtap_assigned_failed(self):
+ with mock.patch.object(pci_lib.PciDeviceIPWrapper,
+ "_execute") as mock_exec:
+ mock_exec.side_effect = Exception()
+ self.assertRaises(exc.IpCommandError,
+ pci_lib.PciDeviceIPWrapper.is_macvtap_assigned,
+ 'enp129s0f3')
--- /dev/null
+---
+fixes:
+ - Fix SR-IOV agent macvtap assigned VF check when linux kernel < 3.13
Code Review / openstack-build / neutron-build.git / commitdiff