split actor that switches channel and pins mirror

Author: Monstrofil

repos/system_upgrade/cloudlinux/actors/pinclnmirror/actor.py

@@ -0,0 +1,59 @@
+import json
+import os
+
+from leapp.actors import Actor
+from leapp.libraries.stdlib import api
+from leapp.libraries.common.cllaunch import run_on_cloudlinux
+from leapp.libraries.common.cln_switch import get_target_userspace_path
+from leapp.tags import DownloadPhaseTag, IPUWorkflowTag
+from leapp.libraries.common.config.version import get_target_major_version
+
+
+class PinClnMirror(Actor):
+    """
+    Save CLN mirror that was used last time.
+    """
+
+    name = 'pin_cln_mirror'
+    consumes = ()
+    produces = ()
+    tags = (IPUWorkflowTag, DownloadPhaseTag.Before)
+
+    CLN_REPO_ID = "cloudlinux-x86_64-server-%s"
+    DEFAULT_CLN_MIRROR = "https://xmlrpc.cln.cloudlinux.com/XMLRPC/"
+
+    @run_on_cloudlinux
+    def process(self):
+        """Pin CLN mirror"""
+        target_userspace = get_target_userspace_path()
+        api.current_logger().info("Pin CLN mirror: target userspace=%s", target_userspace)
+
+        # load last mirror URL from dnf spacewalk plugin cache
+        spacewalk_settings = {}
+
+        # find the mirror used in the last transaction
+        # (expecting to find the one used in dnf_package_download actor)
+        spacewalk_json_path = os.path.join(target_userspace, 'var/lib/dnf/_spacewalk.json')
+        try:
+            with open(spacewalk_json_path) as file:
+                spacewalk_settings = json.load(file)
+        except (OSError, IOError, ValueError):
+            api.current_logger().error(
+                "No spacewalk settings found in %s - can't identify the last used CLN mirror",
+                spacewalk_json_path,
+            )
+
+        mirror_url = spacewalk_settings.get(
+            self.CLN_REPO_ID % get_target_major_version(), {}
+        ).get("url", [self.DEFAULT_CLN_MIRROR])[0]
+
+        # pin mirror
+        mirrorlist_path = os.path.join(target_userspace, 'etc/mirrorlist')
+        with open(mirrorlist_path, 'w') as file:
+            file.write(mirror_url + '\n')
+        api.current_logger().info("Pin CLN mirror %s in %s", mirror_url, mirrorlist_path)
+
+        up2date_path = os.path.join(target_userspace, 'etc/sysconfig/rhn/up2date')
+        with open(up2date_path, 'a+') as file:
+            file.write('\nmirrorURL[comment]=Set mirror URL to /etc/mirrorlist\nmirrorURL=file:///etc/mirrorlist\n')
+        api.current_logger().info("Updated up2date_path %s", up2date_path)

repos/system_upgrade/cloudlinux/actors/switchclnchannel/actor.py

@@ -0,0 +1,57 @@
+import os
+import json
+
+from leapp.actors import Actor
+from leapp.libraries.stdlib import api
+from leapp.tags import FinalizationPhaseTag, IPUWorkflowTag
+from leapp.libraries.stdlib import CalledProcessError
+from leapp.libraries.common.cllaunch import run_on_cloudlinux
+from leapp.libraries.common.cln_switch import cln_switch, get_target_userspace_path
+from leapp import reporting
+from leapp.reporting import Report
+from leapp.libraries.common.config.version import get_target_major_version
+
+
+DEFAULT_CLN_MIRROR = "https://xmlrpc.cln.cloudlinux.com/XMLRPC/"
+
+
+class SwitchClnChannelDownload(Actor):
+    """
+    Permanently switch CLN channel to target os version.
+    """
+
+    name = "switch_cln_channel"
+    consumes = ()
+    produces = (Report,)
+    tags = (FinalizationPhaseTag, IPUWorkflowTag)
+
+    @run_on_cloudlinux
+    def process(self):
+        try:
+            cln_switch(target=int(get_target_major_version()))
+        except CalledProcessError as e:
+            reporting.create_report(
+                [
+                    reporting.Title(
+                        "Failed to switch CloudLinux Network channel from 7 to 8."
+                    ),
+                    reporting.Summary(
+                        "Command {} failed with exit code {}."
+                        " The most probable cause of that is a problem with this system's"
+                        " CloudLinux Network registration.".format(e.command, e.exit_code)
+                    ),
+                    reporting.Remediation(
+                        hint="Check the state of this system's registration with \'rhn_check\'."
+                        " Attempt to re-register the system with \'rhnreg_ks --force\'."
+                    ),
+                    reporting.Severity(reporting.Severity.HIGH),
+                    reporting.Groups(
+                        [reporting.Groups.OS_FACTS, reporting.Groups.AUTHENTICATION]
+                    ),
+                    reporting.Groups([reporting.Groups.INHIBITOR]),
+                ]
+            )
+        except OSError as e:
+            api.current_logger().error(
+                "Could not call RHN command: Message: %s", str(e), exc_info=True
+            )