Merge "Refactor efi_utils for easier maintaining and debugging"

ironic_python_agent/efi_utils.py

@@ -12,7 +12,7 @@
 
 import os
 import re
-import shutil
+import sys
 import tempfile
 
 from ironic_lib import disk_utils
@@ -45,52 +45,53 @@ def manage_uefi(device, efi_system_part_uuid=None):
     efi_partition_mount_point = None
     efi_mounted = False
     LOG.debug('Attempting UEFI loader autodetection and NVRAM record setup.')
+
+    # Force UEFI to rescan the device.
+    utils.rescan_device(device)
+
+    # Trust the contents on the disk in the event of a whole disk image.
+    efi_partition = disk_utils.find_efi_partition(device)
+    if efi_partition:
+        efi_partition = efi_partition['number']
+
+    if not efi_partition and efi_system_part_uuid:
+        # _get_partition returns <device>+<partition> and we only need the
+        # partition number
+        partition = partition_utils.get_partition(
+            device, uuid=efi_system_part_uuid)
+        try:
+            efi_partition = int(partition.replace(device, ""))
+        except ValueError:
+            # NVMe Devices get a partitioning scheme that is different from
+            # traditional block devices like SCSI/SATA
+            efi_partition = int(partition.replace(device + 'p', ""))
+
+    if not efi_partition:
+        # NOTE(dtantsur): we cannot have a valid EFI deployment without an
+        # EFI partition at all. This code path is easily hit when using an
+        # image that is not UEFI compatible (which sadly applies to most
+        # cloud images out there, with a nice exception of Ubuntu).
+        raise errors.DeviceNotFound(
+            "No EFI partition could be detected on device %s and "
+            "EFI partition UUID has not been recorded during deployment "
+            "(which is often the case for whole disk images). "
+            "Are you using a UEFI-compatible image?" % device)
+
+    local_path = tempfile.mkdtemp()
+    efi_partition_mount_point = os.path.join(local_path, "boot/efi")
+    if not os.path.exists(efi_partition_mount_point):
+        os.makedirs(efi_partition_mount_point)
+
+    # The mount needs the device with the partition, in case the
+    # device ends with a digit we add a `p` and the partition number we
+    # found, otherwise we just join the device and the partition number
+    if device[-1].isdigit():
+        efi_device_part = '{}p{}'.format(device, efi_partition)
+    else:
+        efi_device_part = '{}{}'.format(device, efi_partition)
+
     try:
-        # Force UEFI to rescan the device.
+        utils.execute('mount', efi_device_part, efi_partition_mount_point)
-        utils.rescan_device(device)
-
-        local_path = tempfile.mkdtemp()
-        # Trust the contents on the disk in the event of a whole disk image.
-        efi_partition = disk_utils.find_efi_partition(device)
-        if efi_partition:
-            efi_partition = efi_partition['number']
-
-        if not efi_partition and efi_system_part_uuid:
-            # _get_partition returns <device>+<partition> and we only need the
-            # partition number
-            partition = partition_utils.get_partition(
-                device, uuid=efi_system_part_uuid)
-            try:
-                efi_partition = int(partition.replace(device, ""))
-            except ValueError:
-                # NVMe Devices get a partitioning scheme that is different from
-                # traditional block devices like SCSI/SATA
-                efi_partition = int(partition.replace(device + 'p', ""))
-
-        if not efi_partition:
-            # NOTE(dtantsur): we cannot have a valid EFI deployment without an
-            # EFI partition at all. This code path is easily hit when using an
-            # image that is not UEFI compatible (which sadly applies to most
-            # cloud images out there, with a nice exception of Ubuntu).
-            raise errors.DeviceNotFound(
-                "No EFI partition could be detected on device %s and "
-                "EFI partition UUID has not been recorded during deployment "
-                "(which is often the case for whole disk images). "
-                "Are you using a UEFI-compatible image?" % device)
-
-        efi_partition_mount_point = os.path.join(local_path, "boot/efi")
-        if not os.path.exists(efi_partition_mount_point):
-            os.makedirs(efi_partition_mount_point)
-
-        # The mount needs the device with the partition, in case the
-        # device ends with a digit we add a `p` and the partition number we
-        # found, otherwise we just join the device and the partition number
-        if device[-1].isdigit():
-            efi_device_part = '{}p{}'.format(device, efi_partition)
-            utils.execute('mount', efi_device_part, efi_partition_mount_point)
-        else:
-            efi_device_part = '{}{}'.format(device, efi_partition)
-            utils.execute('mount', efi_device_part, efi_partition_mount_point)
         efi_mounted = True
 
         valid_efi_bootloaders = _get_efi_bootloaders(efi_partition_mount_point)
@@ -149,34 +150,28 @@ def manage_uefi(device, efi_system_part_uuid=None):
         return True
 
     except processutils.ProcessExecutionError as e:
-        error_msg = ('Could not verify uefi on device %(dev)s, '
+        error_msg = ('Could not configure UEFI boot on device %(dev)s: %(err)s'
-                     'failed with %(err)s.' % {'dev': device, 'err': e})
+                     % {'dev': device, 'err': e})
-        LOG.error(error_msg)
+        LOG.exception(error_msg)
         raise errors.CommandExecutionError(error_msg)
     finally:
-        LOG.debug('Executing _manage_uefi clean-up.')
+        if efi_mounted:
-        umount_warn_msg = "Unable to umount %(local_path)s. Error: %(error)s"
+            try:
-
-        try:
-            if efi_mounted:
                 utils.execute('umount', efi_partition_mount_point,
                               attempts=3, delay_on_retry=True)
-        except processutils.ProcessExecutionError as e:
-            error_msg = ('Umounting efi system partition failed. '
-                         'Attempted 3 times. Error: %s' % e)
-            LOG.error(error_msg)
-            raise errors.CommandExecutionError(error_msg)
-
-        else:
-            # If umounting the binds succeed then we can try to delete it
-            try:
-                utils.execute('sync')
             except processutils.ProcessExecutionError as e:
-                LOG.warning(umount_warn_msg, {'path': local_path, 'error': e})
+                error_msg = ('Umounting efi system partition failed. '
+                             'Attempted 3 times. Error: %s' % e)
+                LOG.error(error_msg)
+                # Do not mask the actual failure, if any
+                if sys.exc_info()[0] is None:
+                    raise errors.CommandExecutionError(error_msg)
+
             else:
-                # After everything is umounted we can then remove the
+                try:
-                # temporary directory
+                    utils.execute('sync')
-                shutil.rmtree(local_path)
+                except processutils.ProcessExecutionError as e:
+                    LOG.warning('Unable to sync the local disks: %s', e)
 
 
 # NOTE(TheJulia): Do not add bootia32.csv to this list. That is 32bit

ironic_python_agent/tests/unit/test_efi_utils.py

@@ -16,6 +16,7 @@ import tempfile
 from unittest import mock
 
 from ironic_lib import disk_utils
+from oslo_concurrency import processutils
 
 from ironic_python_agent import efi_utils
 from ironic_python_agent import errors
@@ -371,3 +372,100 @@ Boot0002: VENDMAGIC FvFile(9f3c6294-bf9b-4208-9808-be45dfc34b51)
         mkdir_mock.assert_called_once_with(self.fake_dir + '/boot/efi')
         mock_efi_bl.assert_called_once_with(self.fake_dir + '/boot/efi')
         mock_execute.assert_has_calls(expected)
+
+    @mock.patch.object(os.path, 'exists', lambda *_: False)
+    @mock.patch.object(hardware, 'is_md_device', autospec=True)
+    @mock.patch.object(efi_utils, '_get_efi_bootloaders', autospec=True)
+    @mock.patch.object(os, 'makedirs', autospec=True)
+    def test_failure(self, mkdir_mock, mock_efi_bl, mock_is_md_device,
+                     mock_utils_efi_part, mock_get_part_uuid, mock_execute,
+                     mock_rescan):
+        mock_utils_efi_part.return_value = {'number': '1'}
+        mock_get_part_uuid.return_value = self.fake_dev
+        mock_is_md_device.return_value = False
+
+        mock_efi_bl.return_value = ['EFI/BOOT/BOOTX64.EFI']
+
+        mock_execute.side_effect = processutils.ProcessExecutionError('boom')
+
+        self.assertRaisesRegex(errors.CommandExecutionError, 'boom',
+                               efi_utils.manage_uefi,
+                               self.fake_dev, self.fake_root_uuid)
+        mkdir_mock.assert_called_once_with(self.fake_dir + '/boot/efi')
+        mock_efi_bl.assert_not_called()
+        mock_execute.assert_called_once_with(
+            'mount', self.fake_efi_system_part, self.fake_dir + '/boot/efi')
+        mock_rescan.assert_called_once_with(self.fake_dev)
+
+    @mock.patch.object(os.path, 'exists', lambda *_: False)
+    @mock.patch.object(hardware, 'is_md_device', autospec=True)
+    @mock.patch.object(efi_utils, '_get_efi_bootloaders', autospec=True)
+    @mock.patch.object(os, 'makedirs', autospec=True)
+    def test_failure_after_mount(self, mkdir_mock, mock_efi_bl,
+                                 mock_is_md_device, mock_utils_efi_part,
+                                 mock_get_part_uuid, mock_execute,
+                                 mock_rescan):
+        mock_utils_efi_part.return_value = {'number': '1'}
+        mock_get_part_uuid.return_value = self.fake_dev
+        mock_is_md_device.return_value = False
+
+        mock_efi_bl.return_value = ['EFI/BOOT/BOOTX64.EFI']
+
+        mock_execute.side_effect = [
+            ('', ''),
+            processutils.ProcessExecutionError('boom'),
+            ('', ''),
+            ('', ''),
+        ]
+
+        expected = [mock.call('mount', self.fake_efi_system_part,
+                              self.fake_dir + '/boot/efi'),
+                    mock.call('efibootmgr', '-v'),
+                    mock.call('umount', self.fake_dir + '/boot/efi',
+                              attempts=3, delay_on_retry=True),
+                    mock.call('sync')]
+
+        self.assertRaisesRegex(errors.CommandExecutionError, 'boom',
+                               efi_utils.manage_uefi,
+                               self.fake_dev, self.fake_root_uuid)
+        mkdir_mock.assert_called_once_with(self.fake_dir + '/boot/efi')
+        mock_efi_bl.assert_called_once_with(self.fake_dir + '/boot/efi')
+        mock_execute.assert_has_calls(expected)
+        self.assertEqual(4, mock_execute.call_count)
+        mock_rescan.assert_called_once_with(self.fake_dev)
+
+    @mock.patch.object(os.path, 'exists', lambda *_: False)
+    @mock.patch.object(hardware, 'is_md_device', autospec=True)
+    @mock.patch.object(efi_utils, '_get_efi_bootloaders', autospec=True)
+    @mock.patch.object(os, 'makedirs', autospec=True)
+    def test_failure_after_failure(self, mkdir_mock, mock_efi_bl,
+                                   mock_is_md_device, mock_utils_efi_part,
+                                   mock_get_part_uuid, mock_execute,
+                                   mock_rescan):
+        mock_utils_efi_part.return_value = {'number': '1'}
+        mock_get_part_uuid.return_value = self.fake_dev
+        mock_is_md_device.return_value = False
+
+        mock_efi_bl.return_value = ['EFI/BOOT/BOOTX64.EFI']
+
+        mock_execute.side_effect = [
+            ('', ''),
+            processutils.ProcessExecutionError('boom'),
+            processutils.ProcessExecutionError('no umount'),
+            ('', ''),
+        ]
+
+        expected = [mock.call('mount', self.fake_efi_system_part,
+                              self.fake_dir + '/boot/efi'),
+                    mock.call('efibootmgr', '-v'),
+                    mock.call('umount', self.fake_dir + '/boot/efi',
+                              attempts=3, delay_on_retry=True)]
+
+        self.assertRaisesRegex(errors.CommandExecutionError, 'boom',
+                               efi_utils.manage_uefi,
+                               self.fake_dev, self.fake_root_uuid)
+        mkdir_mock.assert_called_once_with(self.fake_dir + '/boot/efi')
+        mock_efi_bl.assert_called_once_with(self.fake_dir + '/boot/efi')
+        mock_execute.assert_has_calls(expected)
+        self.assertEqual(3, mock_execute.call_count)
+        mock_rescan.assert_called_once_with(self.fake_dev)