This strengthens the 'bios id' -> 'device name' mapping algorithm by
looking at the pci device and interface properties specified in the edd
data.
Includes a unit test.
Resolves: rhbz#621175
Related: rhbz#694800
---
storage/devicelibs/edd.py | 230 +++++++++++++++++++++++++---------
tests/storage/devicelibs/edd_test.py | 133 ++++++++++++++++++++
2 files changed, 303 insertions(+), 60 deletions(-)
create mode 100644 tests/storage/devicelibs/edd_test.py
diff --git a/storage/devicelibs/edd.py b/storage/devicelibs/edd.py
index da03914..20a7e64 100644
--- a/storage/devicelibs/edd.py
+++ b/storage/devicelibs/edd.py
@@ -18,80 +18,190 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# Author(s): Hans de Goede <hdegoede@redhat.com>
+# Ales Kozumplik <akozumpl@redhat.com>
#
+import glob
+import logging
import os
+import re
import struct
-import logging
log = logging.getLogger("storage")
-def get_edd_dict(devices):
- """Given an array of devices return a dict with the BIOS ID for them."""
- edd_dict = {}
+re_host_bus = re.compile("^PCIs*(S*)s*channel: (S*)s*$")
+re_interface_scsi = re.compile("^SCSIs*id: (S*)s*lun: (S*)s*$")
+re_interface_ata = re.compile("^ATAs*device: (S*)s*$")
- for biosdev in range(80, 80 + 15):
- sysfspath = "/sys/firmware/edd/int13_dev%d" % biosdev
- if not os.path.exists(sysfspath):
- break # We are done
+class EddEntry(object):
+ """ This object merely collects what the /sys/firmware/edd/* entries can
+ provide.
+ """
+ def __init__(self, sysfspath):
+ self.type = None
+
+ self.ata_device = None
+ self.channel = None
+ self.mbr_signature = None
+ self.pci_dev = None
+ self.scsi_id = None
+ self.scsi_lun = None
+ self.sectors = None
+
+ self.load(sysfspath)
+
+ def __str__(self):
+ return
+ " type: %(type)s, ata_device: %(ata_device)s
"
+ " channel: %(channel)s, mbr_signature: %(mbr_signature)s
"
+ " pci_dev: %(pci_dev)s, scsi_id: %(scsi_id)s
"
+ " scsi_lun: %(scsi_lun)s, sectors: %(sectors)s" % self.__dict__
+
+ def _read_file(self, filename):
+ contents = None
+ if os.path.exists(filename):
+ with open(filename) as f:
+ contents = f.read().rstrip()
+ return contents
+
+ def load(self, sysfspath):
+ interface = self._read_file(os.path.join(sysfspath, "interface"))
+ if interface:
+ self.type = interface.split()[0]
+ if self.type == "SCSI":
+ match = re_interface_scsi.match(interface)
+ self.scsi_id = int(match.group(1))
+ self.scsi_lun = int(match.group(2))
+ elif self.type == "ATA":
+ match = re_interface_ata.match(interface)
+ self.ata_device = int(match.group(1))
+
+ self.mbr_signature = self._read_file(
+ os.path.join(sysfspath, "mbr_signature"))
+ sectors = self._read_file(os.path.join(sysfspath, "sectors"))
+ if sectors:
+ self.sectors = int(sectors)
+ hbus = self._read_file(os.path.join(sysfspath, "host_bus"))
+ if hbus:
+ match = re_host_bus.match(hbus)
+ self.pci_dev = match.group(1)
+ self.channel = int(match.group(2))
+
+class EddMatcher(object):
+ """ This object tries to match given entry to a disk device name.
+
+ Assuming, heuristic analysis and guessing hapens here.
+ """
+ def __init__(self, edd_entry):
+ self.edd = edd_entry
- sysfspath = "/sys/firmware/edd/int13_dev%d/mbr_signature" % biosdev
+ def devname_from_pci_dev(self):
+ name = None
+ if self.edd.type == "ATA":
+ path = "/sys/devices/pci0000:00/0000:%(pci_dev)s/host%(chan)d/"
+ "target%(chan)d:0:%(dev)d/%(chan)d:0:%(dev)d:0/block" % {
+ 'pci_dev' : self.edd.pci_dev,
+ 'chan' : self.edd.channel,
+ 'dev' : self.edd.ata_device
+ }
+ block_entries = os.listdir(path)
+ if len(block_entries) == 1:
+ name = block_entries[0]
+ elif self.edd.type == "SCSI":
+ pattern = "/sys/devices/pci0000:00/0000:%(pci_dev)s/virtio*/block" %
+ {'pci_dev' : self.edd.pci_dev}
+ matching_paths = glob.glob(pattern)
+ if len(matching_paths) != 1 or not os.path.exists(matching_paths[0]):
+ return None
+ block_entries = os.listdir(matching_paths[0])
+ if len(block_entries) == 1:
+ name = block_entries[0]
+ return name
+
+ def match_via_mbrsigs(self, mbr_dict):
+ """ Try to match the edd entry based on its mbr signature.
+
+ This will obviously fail for a fresh drive/image, but in extreme
+ cases can also show false positives for randomly matching data.
+ """
+ for (name, mbr_signature) in mbr_dict.items():
+ if mbr_signature == self.edd.mbr_signature:
+ return name
+ return None
+
+def biosdev_to_edd_dir(biosdev):
+ return "/sys/firmware/edd/int13_dev%x" % biosdev
+
+def collect_edd_data():
+ edd_data_dict = {}
+ for biosdev in range(0x80, 0x80+16):
+ sysfspath = biosdev_to_edd_dir(biosdev)
if not os.path.exists(sysfspath):
- log.warning("No mbrsig for biosdev: %d" % biosdev)
- continue
+ break
+ edd_data_dict[biosdev] = EddEntry(sysfspath)
+ return edd_data_dict
+def collect_mbrs(devices):
+ mbr_dict = {}
+ for dev in devices:
try:
- file = open(sysfspath, "r")
- eddsig = file.read()
- file.close()
- except (IOError, OSError) as e:
- log.warning("Error reading EDD mbrsig for %d: %s" %
- (biosdev, str(e)))
+ fd = os.open(dev.path, os.O_RDONLY)
+ os.lseek(fd, 440, 0)
+ mbrsig = struct.unpack('I', os.read(fd, 4))
+ os.close(fd)
+ except OSError as e:
+ log.warning("Error reading mbrsig from disk %s: %s" %
+ (dev.name, str(e)))
continue
- sysfspath = "/sys/firmware/edd/int13_dev%d/sectors" % biosdev
- try:
- file = open(sysfspath, "r")
- eddsize = file.read()
- file.close()
- except (IOError, OSError) as e:
- eddsize = None
-
- found = []
- for dev in devices:
- try:
- fd = os.open(dev.path, os.O_RDONLY)
- os.lseek(fd, 440, 0)
- mbrsig = struct.unpack('I', os.read(fd, 4))
- os.close(fd)
- except OSError as e:
- log.warning("Error reading mbrsig from disk %s: %s" %
- (dev.name, str(e)))
- continue
-
- mbrsigStr = "0x%08x
" % mbrsig
- if mbrsigStr == eddsig:
- if eddsize:
- sysfspath = "/sys%s/size" % dev.sysfsPath
- try:
- file = open(sysfspath, "r")
- size = file.read()
- file.close()
- except (IOError, OSError) as e:
- log.warning("Error getting size for: %s" % dev.name)
- continue
- if eddsize != size:
- continue
- found.append(dev.name)
-
- if not found:
- log.error("No matching mbr signature found for biosdev %d" %
- biosdev)
- elif len(found) > 1:
- log.error("Multiple signature matches found for biosdev %d: %s" %
- (biosdev, str(found)))
+ mbrsig_str = "0x%08x" % mbrsig
+ # sanity check
+ if mbrsig_str == '0x00000000':
+ log.info("edd: MBR signature on %s is zero. new disk image?" % dev.name)
+ continue
else:
- log.info("Found %s for biosdev %d" %(found[0], biosdev))
- edd_dict[found[0]] = biosdev
+ for (dev_name, mbrsig_str_old) in mbr_dict.items():
+ if mbrsig_str_old == mbrsig_str:
+ log.error("edd: dupicite MBR signature %s for %s and %s" %
+ (mbrsig_str, dev_name, dev.name))
+ # this actually makes all the other data useless
+ return {}
+ # update the dictionary
+ mbr_dict[dev.name] = mbrsig_str
+ log.info("edd: collected mbr signatures: %s" % mbr_dict)
+ return mbr_dict
+def get_edd_dict(devices):
+ """ Generates the 'device name' -> 'edd number' mapping.
+
+ Note: The EDD kernel module that exposes /sys/firmware/edd is thoroughly
+ broken, the information there is incomplete and sometimes downright
+ wrong.
+ """
+ mbr_dict = collect_mbrs(devices)
+ edd_entries_dict = collect_edd_data()
+ edd_dict = {}
+ for (edd_number, edd_entry) in edd_entries_dict.items():
+ log.debug("edd: data extracted from 0x%x:
%s" % (edd_number, edd_entry))
+ matcher = EddMatcher(edd_entry)
+ # first try to match through the pci dev etc.
+ name = matcher.devname_from_pci_dev()
+ # next try to compare mbr signatures
+ if name:
+ log.debug("matched 0x%x to %s using pci_dev" % (edd_number, name))
+ else:
+ name = matcher.match_via_mbrsigs(mbr_dict)
+ if name:
+ log.info("matched 0x%x to %s using MBR sig" % (edd_number, name))
+
+ if name:
+ old_edd_number = edd_dict.get(name)
+ if old_edd_number:
+ log.info("edd: both edd entries 0x%x and 0x%x seem to map to %s" %
+ old_edd_number, edd_number, name)
+ # this means all the other data can be confused and useless
+ return {}
+ edd_dict[name] = edd_number
+ continue
+ log.error("edd: unable to match edd entry 0x%x" % edd_number)
return edd_dict
diff --git a/tests/storage/devicelibs/edd_test.py b/tests/storage/devicelibs/edd_test.py
new file mode 100644
index 0000000..ab87307
--- /dev/null
+++ b/tests/storage/devicelibs/edd_test.py
@@ -0,0 +1,133 @@
+import mock
+
+class EddTestCase(mock.TestCase):
+ def setUp(self):
+ self.setupModules(
+ ['_isys', 'logging', 'anaconda_log', 'block'])
+
+ def tearDown(self):
+ from storage.devicelibs import edd
+ self.tearDownModules()
+ mock.DiskIO.restore_module(edd)
+
+ def test_biosdev_to_edd_dir(self):
+ from storage.devicelibs import edd
+ path = edd.biosdev_to_edd_dir(138)
+ self.assertEqual("/sys/firmware/edd/int13_dev8a", path)
+
+ def test_collect_edd_data(self):
+ from storage.devicelibs import edd
+
+ # test with vda, vdb
+ fs = EddTestFS(edd).vda_vdb()
+ edd_dict = edd.collect_edd_data()
+ self.assertEqual(len(edd_dict), 2)
+ self.assertEqual(edd_dict[0x80].type, "SCSI")
+ self.assertEqual(edd_dict[0x80].scsi_id, 0)
+ self.assertEqual(edd_dict[0x80].scsi_lun, 0)
+ self.assertEqual(edd_dict[0x80].pci_dev, "00:05.0")
+ self.assertEqual(edd_dict[0x80].channel, 0)
+ self.assertEqual(edd_dict[0x80].sectors, 16777216)
+ self.assertEqual(edd_dict[0x81].pci_dev, "00:06.0")
+
+ # test with sda, vda
+ fs = EddTestFS(edd).sda_vda()
+ edd_dict = edd.collect_edd_data()
+ self.assertEqual(len(edd_dict), 2)
+ self.assertEqual(edd_dict[0x80].type, "ATA")
+ self.assertEqual(edd_dict[0x80].scsi_id, None)
+ self.assertEqual(edd_dict[0x80].scsi_lun, None)
+ self.assertEqual(edd_dict[0x80].pci_dev, "00:01.1")
+ self.assertEqual(edd_dict[0x80].channel, 0)
+ self.assertEqual(edd_dict[0x80].sectors, 2097152)
+ self.assertEqual(edd_dict[0x80].ata_device, 0)
+ self.assertEqual(edd_dict[0x80].mbr_signature, "0x000ccb01")
+
+ def test_edd_entry_str(self):
+ from storage.devicelibs import edd
+ fs = EddTestFS(edd).sda_vda()
+ edd_dict = edd.collect_edd_data()
+ expected_output = """ type: ATA, ata_device: 0
+ channel: 0, mbr_signature: 0x000ccb01
+ pci_dev: 00:01.1, scsi_id: None
+ scsi_lun: None, sectors: 2097152"""
+ self.assertEqual(str(edd_dict[0x80]), expected_output)
+
+ def test_matcher_device_path(self):
+ from storage.devicelibs import edd
+ fs = EddTestFS(edd).sda_vda()
+ edd_dict = edd.collect_edd_data()
+
+ analyzer = edd.EddMatcher(edd_dict[0x80])
+ path = analyzer.devname_from_pci_dev()
+ self.assertEqual(path, "sda")
+
+ analyzer = edd.EddMatcher(edd_dict[0x81])
+ path = analyzer.devname_from_pci_dev()
+ self.assertEqual(path, "vda")
+
+ def test_get_edd_dict_1(self):
+ """ Test get_edd_dict()'s pci_dev matching. """
+ from storage.devicelibs import edd
+ fs = EddTestFS(edd).sda_vda()
+ self.assertEqual(edd.get_edd_dict([]),
+ {'sda' : 0x80,
+ 'vda' : 0x81})
+
+ def test_get_edd_dict_2(self):
+ """ Test get_edd_dict()'s pci_dev matching. """
+ from storage.devicelibs import edd
+ edd.collect_mbrs = mock.Mock(return_value = {
+ 'sda' : '0x000ccb01',
+ 'vda' : '0x0006aef1'})
+ fs = EddTestFS(edd).sda_vda_missing_details()
+ self.assertEqual(edd.get_edd_dict([]),
+ {'sda' : 0x80,
+ 'vda' : 0x81})
+
+
+
+class EddTestFS(object):
+ def __init__(self, target_module):
+ self.fs = mock.DiskIO()
+ self.fs.take_over_module(target_module)
+
+ def sda_vda_missing_details(self):
+ self.fs["/sys/firmware/edd/int13_dev80"] = self.fs.Dir()
+ self.fs["/sys/firmware/edd/int13_dev80/mbr_signature"] = "0x000ccb01"
+ self.fs["/sys/firmware/edd/int13_dev81"] = self.fs.Dir()
+ self.fs["/sys/firmware/edd/int13_dev81/mbr_signature"] = "0x0006aef1"
+
+ def sda_vda(self):
+ self.fs["/sys/firmware/edd/int13_dev80"] = self.fs.Dir()
+ self.fs["/sys/firmware/edd/int13_dev80/host_bus"] = "PCI 00:01.1 channel: 0
"
+ self.fs["/sys/firmware/edd/int13_dev80/interface"] = "ATA device: 0
"
+ self.fs["/sys/firmware/edd/int13_dev80/mbr_signature"] = "0x000ccb01"
+ self.fs["/sys/firmware/edd/int13_dev80/sectors"] = "2097152
"
+
+ self.fs["/sys/firmware/edd/int13_dev81"] = self.fs.Dir()
+ self.fs["/sys/firmware/edd/int13_dev81/host_bus"] = "PCI 00:05.0 channel: 0
"
+ self.fs["/sys/firmware/edd/int13_dev81/interface"] = "SCSI id: 0 lun: 0
"
+ self.fs["/sys/firmware/edd/int13_dev81/mbr_signature"] = "0x0006aef1"
+ self.fs["/sys/firmware/edd/int13_dev81/sectors"] = "16777216
"
+
+ self.fs["/sys/devices/pci0000:00/0000:00:01.1/host0/target0:0:0/0:0:0:0/block"] = self.fs.Dir()
+ self.fs["/sys/devices/pci0000:00/0000:00:01.1/host0/target0:0:0/0:0:0:0/block/sda"] = self.fs.Dir()
+
+ self.fs["/sys/devices/pci0000:00/0000:00:05.0/virtio2/block"] = self.fs.Dir()
+ self.fs["/sys/devices/pci0000:00/0000:00:05.0/virtio2/block/vda"] = self.fs.Dir()
+
+ return self.fs
+
+ def vda_vdb(self):
+ self.fs["/sys/firmware/edd/int13_dev80"] = self.fs.Dir()
+ self.fs["/sys/firmware/edd/int13_dev80/host_bus"] = "PCI 00:05.0 channel: 0
"
+ self.fs["/sys/firmware/edd/int13_dev80/interface"] = "SCSI id: 0 lun: 0
"
+ self.fs["/sys/firmware/edd/int13_dev80/sectors"] = "16777216
"
+
+ self.fs["/sys/firmware/edd/int13_dev81"] = self.fs.Dir()
+ self.fs["/sys/firmware/edd/int13_dev81/host_bus"] = "PCI 00:06.0 channel: 0
"
+ self.fs["/sys/firmware/edd/int13_dev81/interface"] = "SCSI id: 0 lun: 0
"
+ self.fs["/sys/firmware/edd/int13_dev81/sectors"] = "4194304
"
+
+ return self.fs
--
1.7.5.4
_______________________________________________
Anaconda-devel-list mailing list
Anaconda-devel-list@redhat.com
https://www.redhat.com/mailman/listinfo/anaconda-devel-list
edd: refactor and enhance the edd module.
