refactor handling of kernel-related information

Refactor the handling of kernel-related information away from
using distributed ad-hoc logic based only on kernel release in
IPUConfiguration. Instead, introduce the KernelInfo message providing
rich information about the booted kernel. These changes also affect
the information about the target kernel which previously only included
target kernel's nevra that was misleadingly marked as 'version'. The new
target kernel info message also contains paths to frequently used files
such as the kernel image path and initramfs location. All old
functionality has been kept in place, but deprecated.

repos/system_upgrade/common/actors/forcedefaultboottotargetkernelversion/actor.py

@@ -1,6 +1,6 @@
 from leapp.actors import Actor
 from leapp.libraries.actor import forcedefaultboot
-from leapp.models import InstalledTargetKernelVersion
+from leapp.models import InstalledTargetKernelInfo
 from leapp.tags import FinalizationPhaseTag, IPUWorkflowTag
 
 
@@ -14,7 +14,7 @@ class ForceDefaultBootToTargetKernelVersion(Actor):
     """
 
     name = 'force_default_boot_to_target_kernel_version'
-    consumes = (InstalledTargetKernelVersion,)
+    consumes = (InstalledTargetKernelInfo,)
     produces = ()
     tags = (FinalizationPhaseTag, IPUWorkflowTag)
 

repos/system_upgrade/common/actors/forcedefaultboottotargetkernelversion/libraries/forcedefaultboot.py

@@ -1,58 +1,33 @@
-import os
-from collections import namedtuple
-
 from leapp.libraries import stdlib
 from leapp.libraries.common.config import architecture
 from leapp.libraries.stdlib import api, config
-from leapp.models import InstalledTargetKernelVersion
-
-KernelInfo = namedtuple('KernelInfo', ('kernel_path', 'initrd_path'))
-
-
-def get_kernel_info(message):
-    kernel_name = 'vmlinuz-{}'.format(message.version)
-    initrd_name = 'initramfs-{}.img'.format(message.version)
-    kernel_path = os.path.join('/boot', kernel_name)
-    initrd_path = os.path.join('/boot', initrd_name)
-
-    target_version_bootable = True
-    if not os.path.exists(kernel_path):
-        target_version_bootable = False
-        api.current_logger().warning('Mandatory kernel %s does not exist', kernel_path)
-    if not os.path.exists(initrd_path):
-        target_version_bootable = False
-        api.current_logger().warning('Mandatory initrd %s does not exist', initrd_path)
-
-    if target_version_bootable:
-        return KernelInfo(kernel_path=kernel_path, initrd_path=initrd_path)
-
-    api.current_logger().warning('Skipping check due to missing mandatory files')
-    return None
+from leapp.models import InstalledTargetKernelInfo
 
 
 def update_default_kernel(kernel_info):
     try:
-        stdlib.run(['grubby', '--info', kernel_info.kernel_path])
+        stdlib.run(['grubby', '--info', kernel_info.kernel_img_path])
     except stdlib.CalledProcessError:
         api.current_logger().error('Expected kernel %s to be installed at the boot loader but cannot be found.',
-                                   kernel_info.kernel_path)
+                                   kernel_info.kernel_img_path)
     except OSError:
         api.current_logger().error('Could not check for kernel existence in boot loader. Is grubby installed?')
     else:
         try:
-            stdlib.run(['grubby', '--set-default', kernel_info.kernel_path])
+            stdlib.run(['grubby', '--set-default', kernel_info.kernel_img_path])
             if architecture.matches_architecture(architecture.ARCH_S390X):
                 # on s390x we need to call zipl explicitly because of issue in grubby,
                 # otherwise the new boot entry will not be set as default
                 # See https://bugzilla.redhat.com/show_bug.cgi?id=1764306
                 stdlib.run(['/usr/sbin/zipl'])
         except (OSError, stdlib.CalledProcessError):
-            api.current_logger().error('Failed to set default kernel to: %s', kernel_info.kernel_path, exc_info=True)
+            api.current_logger().error('Failed to set default kernel to: %s',
+                                       kernel_info.kernel_img_path, exc_info=True)
 
 
 def process():
-    if (config.is_debug and not
-            architecture.matches_architecture(architecture.ARCH_S390X)):  # pylint: disable=using-constant-test
+    is_system_s390x = architecture.matches_architecture(architecture.ARCH_S390X)
+    if config.is_debug and not is_system_s390x:  # pylint: disable=using-constant-test
         try:
             # the following command prints output of grubenv for debugging purposes and is repeated after setting
             # default kernel so we can be sure we have the right saved entry
@@ -65,12 +40,18 @@ def process():
             stdlib.run(['grub2-editenv', 'list'])
         except stdlib.CalledProcessError:
             api.current_logger().error('Failed to execute "grub2-editenv list" command')
-    message = next(api.consume(InstalledTargetKernelVersion), None)
-    if not message:
+
+    kernel_info = next(api.consume(InstalledTargetKernelInfo), None)
+    if not kernel_info:
         api.current_logger().warning(('Skipped - Forcing checking and setting default boot entry to target kernel'
                                       ' version due to missing message'))
         return
 
+    if not kernel_info.kernel_img_path:  # Should be always set
+        api.current_logger().warning(('Skipping forcing of default boot entry - target kernel info '
+                                      'does not contain a kernel image path.'))
+        return
+
     try:
         current_default_kernel = stdlib.run(['grubby', '--default-kernel'])['stdout'].strip()
     except (OSError, stdlib.CalledProcessError):
@@ -84,17 +65,12 @@ def process():
             api.current_logger().warning('Failed to query grubby for default {}'.format(type_), exc_info=True)
             return
 
-    kernel_info = get_kernel_info(message)
-    if not kernel_info:
-        return
-
-    if current_default_kernel != kernel_info.kernel_path:
+    if current_default_kernel != kernel_info.kernel_img_path:
         api.current_logger().warning(('Current default boot entry not target kernel version: Current default: %s.'
                                       'Forcing default kernel to %s'),
-                                     current_default_kernel, kernel_info.kernel_path)
+                                     current_default_kernel, kernel_info.kernel_img_path)
         update_default_kernel(kernel_info)
-    if (config.is_debug and not
-            architecture.matches_architecture(architecture.ARCH_S390X)):  # pylint: disable=using-constant-test
+    if config.is_debug and not is_system_s390x:  # pylint: disable=using-constant-test
         try:
             stdlib.run(['grub2-editenv', 'list'])
         except stdlib.CalledProcessError:

repos/system_upgrade/common/actors/forcedefaultboottotargetkernelversion/tests/test_forcedefaultboot_forcedefaultboottotargetkernelversion.py

@@ -8,7 +8,7 @@
 from leapp.libraries.common.config import architecture
 from leapp.libraries.common.testutils import CurrentActorMocked, logger_mocked
 from leapp.libraries.stdlib import api
-from leapp.models import InstalledTargetKernelVersion
+from leapp.models import InstalledTargetKernelInfo
 
 Expected = namedtuple(
     'Expected', (
@@ -19,15 +19,15 @@
 
 Case = namedtuple(
     'Case',
-    ('initrd_exists',
-     'kernel_exists',
+    ('kernel_exists',
      'entry_default',
      'entry_exists',
      'message_available',
      'arch_s390x'
      )
 )
 
+TARGET_KERNEL_NEVRA = 'kernel-core-1.2.3.4.el8.x86_64'
 TARGET_KERNEL_VERSION = '1.2.3.4.el8.x86_64'
 TARGET_KERNEL_TITLE = 'Red Hat Enterprise Linux ({}) 8.1 (Ootpa)'.format(TARGET_KERNEL_VERSION)
 TARGET_KERNEL_PATH = '/boot/vmlinuz-{}'.format(TARGET_KERNEL_VERSION)
@@ -37,48 +37,27 @@
 OLD_KERNEL_TITLE = 'Red Hat Enterprise Linux ({}) 7.6 (Maipo)'.format(OLD_KERNEL_VERSION)
 OLD_KERNEL_PATH = '/boot/vmlinuz-{}'.format(OLD_KERNEL_VERSION)
 
+
 CASES = (
-    (Case(initrd_exists=True, kernel_exists=True, entry_default=True, entry_exists=True, message_available=True,
-          arch_s390x=False),
-     Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=False, kernel_exists=True, entry_default=False, entry_exists=True, message_available=True,
-          arch_s390x=False),
-     Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=True, kernel_exists=False, entry_default=False, entry_exists=True, message_available=True,
-          arch_s390x=False),
+    (Case(kernel_exists=True, entry_default=True, entry_exists=True, message_available=True, arch_s390x=False),
      Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=False, kernel_exists=False, entry_default=False, entry_exists=True, message_available=True,
-          arch_s390x=False),
+    (Case(kernel_exists=False, entry_default=False, entry_exists=True, message_available=True, arch_s390x=False),
      Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=True, kernel_exists=True, entry_default=False, entry_exists=True, message_available=False,
-          arch_s390x=False),
+    (Case(kernel_exists=True, entry_default=False, entry_exists=True, message_available=False, arch_s390x=False),
      Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=True, kernel_exists=True, entry_default=False, entry_exists=False, message_available=False,
-          arch_s390x=False),
+    (Case(kernel_exists=True, entry_default=False, entry_exists=False, message_available=False, arch_s390x=False),
      Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=True, kernel_exists=True, entry_default=False, entry_exists=True, message_available=True,
-          arch_s390x=False),
+    (Case(kernel_exists=True, entry_default=False, entry_exists=True, message_available=True, arch_s390x=False),
      Expected(grubby_setdefault=True, zipl_called=False)),
-    (Case(initrd_exists=True, kernel_exists=True, entry_default=True, entry_exists=True, message_available=True,
-          arch_s390x=True),
-     Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=False, kernel_exists=True, entry_default=False, entry_exists=True, message_available=True,
-          arch_s390x=True),
-     Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=True, kernel_exists=False, entry_default=False, entry_exists=True, message_available=True,
-          arch_s390x=True),
+    (Case(kernel_exists=True, entry_default=True, entry_exists=True, message_available=True, arch_s390x=True),
      Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=False, kernel_exists=False, entry_default=False, entry_exists=True, message_available=True,
-          arch_s390x=True),
+    (Case(kernel_exists=False, entry_default=False, entry_exists=True, message_available=True, arch_s390x=True),
      Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=True, kernel_exists=True, entry_default=False, entry_exists=True, message_available=False,
-          arch_s390x=True),
+    (Case(kernel_exists=True, entry_default=False, entry_exists=True, message_available=False, arch_s390x=True),
      Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=True, kernel_exists=True, entry_default=False, entry_exists=False, message_available=False,
-          arch_s390x=True),
+    (Case(kernel_exists=True, entry_default=False, entry_exists=False, message_available=False, arch_s390x=True),
      Expected(grubby_setdefault=False, zipl_called=False)),
-    (Case(initrd_exists=True, kernel_exists=True, entry_default=False, entry_exists=True, message_available=True,
-          arch_s390x=True),
+    (Case(kernel_exists=True, entry_default=False, entry_exists=True, message_available=True, arch_s390x=True),
      Expected(grubby_setdefault=True, zipl_called=True))
 )
 
@@ -143,7 +122,11 @@ def grubby_set_default(self, cmd):
 def mocked_consume(case):
     def impl(*args):
         if case.message_available:
-            return iter((InstalledTargetKernelVersion(version=TARGET_KERNEL_VERSION),))
+            kernel_img_path = TARGET_KERNEL_PATH if case.kernel_exists else ''
+            msg = InstalledTargetKernelInfo(pkg_nevra=TARGET_KERNEL_NEVRA,
+                                            kernel_img_path=kernel_img_path,
+                                            initramfs_path=TARGET_INITRD_PATH)
+            return iter((msg,))
         return iter(())
     return impl
 

repos/system_upgrade/common/actors/ipuworkflowconfig/libraries/ipuworkflowconfig.py

@@ -68,6 +68,7 @@ def produce_ipu_config(actor):
     flavour = os.environ.get('LEAPP_UPGRADE_PATH_FLAVOUR')
     target_version = os.environ.get('LEAPP_UPGRADE_PATH_TARGET_RELEASE')
     os_release = get_os_release('/etc/os-release')
+
     actor.produce(IPUConfig(
         leapp_env_vars=get_env_vars(),
         os_release=os_release,

repos/system_upgrade/common/actors/kernel/checkinstalledkernels/actor.py

@@ -1,6 +1,6 @@
 from leapp.actors import Actor
 from leapp.libraries.actor import checkinstalledkernels
-from leapp.models import InstalledRedHatSignedRPM
+from leapp.models import InstalledRedHatSignedRPM, KernelInfo
 from leapp.reporting import Report
 from leapp.tags import ChecksPhaseTag, IPUWorkflowTag
 
@@ -30,7 +30,7 @@ class CheckInstalledKernels(Actor):
     """
 
     name = 'check_installed_kernels'
-    consumes = (InstalledRedHatSignedRPM,)
+    consumes = (InstalledRedHatSignedRPM, KernelInfo)
     produces = (Report,)
     tags = (IPUWorkflowTag, ChecksPhaseTag)
 

repos/system_upgrade/common/actors/kernel/checkinstalledkernels/libraries/checkinstalledkernels.py

@@ -13,36 +13,12 @@ def labelCompare(*args):
 
 from leapp import reporting
 from leapp.exceptions import StopActorExecutionError
-from leapp.libraries.common.config import architecture, version
+from leapp.libraries.common.config import architecture, utils
 from leapp.libraries.stdlib import api
-from leapp.models import InstalledRedHatSignedRPM
+from leapp.models import InstalledRedHatSignedRPM, KernelInfo
 
 
-def get_current_kernel_version():
-    """
-    Get the version of the running kernel as a string.
-    """
-    return api.current_actor().configuration.kernel.split('-')[0]
-
-
-def get_current_kernel_release():
-    """
-    Get the release of the current kernel as a string.
-    """
-    return api.current_actor().configuration.kernel.split('-')[1]
-
-
-def get_current_kernel_evr():
-    """
-    Get a 3-tuple (EVR) of the current booted kernel.
-
-    Epoch in this case is always empty string. In case of kernel, epoch is
-    never expected to be set.
-    """
-    return ('', get_current_kernel_version(), get_current_kernel_release())
-
-
-def get_pkgs(pkg_name):
+def get_all_pkgs_with_name(pkg_name):
     """
     Get all installed packages of the given name signed by Red Hat.
     """
@@ -56,17 +32,8 @@ def get_EVR(pkg):
 
     Epoch is always set as an empty string as in case of kernel epoch is not
     expected to be set - ever.
-
-    The release includes an architecture as well.
-    """
-    return ('', pkg.version, '{}.{}'.format(pkg.release, pkg.arch))
-
-
-def _get_pkgs_evr(pkgs):
-    """
-    Return 3-tuples (EVR) of the given packages.
     """
-    return [get_EVR(pkg) for pkg in pkgs]
+    return ('', pkg.version, pkg.release)
 
 
 def get_newest_evr(pkgs):
@@ -78,35 +45,22 @@ def get_newest_evr(pkgs):
     """
     if not pkgs:
         return None
-    rpms_evr = _get_pkgs_evr(pkgs)
 
-    newest_evr = rpms_evr.pop()
-    for pkg in rpms_evr:
-        if labelCompare(newest_evr, pkg) < 0:
-            newest_evr = pkg
-    return newest_evr
-
-
-def _get_kernel_rpm_name():
-    base_name = 'kernel'
-    if version.is_rhel_realtime():
-        api.current_logger().info('The Real Time kernel boot detected.')
-        base_name = 'kernel-rt'
+    newest_evr = get_EVR(pkgs[0])
+    for pkg in pkgs:
+        evr = get_EVR(pkg)
+        if labelCompare(newest_evr, evr) < 0:
+            newest_evr = evr
 
-    if version.get_source_major_version() == '7':
-        return base_name
-
-    # Since RHEL 8, the kernel|kernel-rt rpm is just a metapackage that even
-    # does not have to be installed on the system.
-    # The kernel-core|kernel-rt-core rpm is the one we care about instead.
-    return '{}-core'.format(base_name)
+    return newest_evr
 
 
 def process():
-    kernel_name = _get_kernel_rpm_name()
-    pkgs = get_pkgs(kernel_name)
+    kernel_info = utils.require_exactly_one_message_of_type(KernelInfo)
+    pkgs = get_all_pkgs_with_name(kernel_info.pkg.name)
+
     if not pkgs:
-        # Hypothatical, user is not allowed to install any kernel that is not signed by RH
+        # Hypothetical, user is not allowed to install any kernel that is not signed by RH
         # In case we would like to be cautious, we could check whether there are no other
         # kernels installed as well.
         api.current_logger().error('Cannot find any installed kernel signed by Red Hat.')
@@ -130,13 +84,13 @@ def process():
             reporting.RelatedResource('package', 'kernel')
         ])
 
-    current_evr = get_current_kernel_evr()
-    newest_evr = get_newest_evr(pkgs)
+    current_kernel_evr = get_EVR(kernel_info.pkg)
+    newest_kernel_evr = get_newest_evr(pkgs)
 
-    api.current_logger().debug('Current kernel EVR: {}'.format(current_evr))
-    api.current_logger().debug('Newest kernel EVR: {}'.format(newest_evr))
+    api.current_logger().debug('Current kernel EVR: {}'.format(current_kernel_evr))
+    api.current_logger().debug('Newest kernel EVR: {}'.format(newest_kernel_evr))
 
-    if current_evr != newest_evr:
+    if current_kernel_evr != newest_kernel_evr:
         title = 'Newest installed kernel not in use'
         summary = ('To ensure a stable upgrade, the machine needs to be'
                    ' booted into the latest installed kernel.')