cli: Add command rerun

Per request this patch adds support for rerunning leapp from a certain
point. As for now it only supports the FirstBoot phase to be re-run.

In order to accomplish the re-run functionality leapp clones the context
of the last upgrade execution and removes all checkpoints and errors
from it.
Cloning means it will duplicate all the context aware data in the
database in order to allow a post mortem inspection of what has
happened.
So it will not blindly delete data but remove the errors and checkpoints
only from the cloned part of the data.

Additionally it allows to select only actors which have a certain tag
specified during the re-run. This will help users that want to re-run
their own set of actors only during the FirstBoot phase without having
to create a complicated workaround using snactor and somehow getting it
to use the data from the upgrade. This allows the actors that are re-run
to consume messages that have been generated during the upgrade before
this phase has been run.

Signed-off-by: Vinzenz Feenstra <[email protected]>

leapp/cli/__init__.py

@@ -13,6 +13,6 @@ def cli(args): # noqa; pylint: disable=unused-argument
 
 def main():
     os.environ['LEAPP_HOSTNAME'] = socket.getfqdn()
-    for cmd in [upgrade.list_runs, upgrade.preupgrade, upgrade.upgrade, upgrade.answer]:
+    for cmd in [upgrade.list_runs, upgrade.preupgrade, upgrade.upgrade, upgrade.answer, upgrade.rerun]:
         cli.command.add_sub(cmd.command)
     cli.command.execute('leapp version {}'.format(VERSION))

leapp/cli/__main__.py

@@ -1,3 +1,4 @@
+
 from leapp.cli import main
 import leapp.utils.i18n  # noqa: F401; pylint: disable=unused-import
 

leapp/cli/upgrade/__init__.py

@@ -6,6 +6,7 @@
 import sys
 import tarfile
 import uuid
+from argparse import Namespace
 from datetime import datetime
 
 from leapp.config import get_config
@@ -14,12 +15,16 @@
 from leapp.messaging.answerstore import AnswerStore
 from leapp.repository.scan import find_and_scan_repositories
 from leapp.utils.audit import Execution, get_connection, get_checkpoints
-from leapp.utils.clicmd import command, command_opt
+from leapp.utils.audit.contextclone import clone_context
+from leapp.utils.clicmd import command, command_arg, command_opt
 from leapp.utils.output import (report_errors, report_info, beautify_actor_exception, report_unsupported,
                                 report_inhibitors)
 from leapp.utils.report import fetch_upgrade_report_messages, generate_report_file
 
 
+RERUN_SUPPORTED_PHASES = ('FirstBoot',)
+
+
 def archive_logfiles():
     """ Archive log files from a previous run of Leapp """
     cfg = get_config()
@@ -67,13 +72,17 @@ def load_repositories():
     return manager
 
 
-def fetch_last_upgrade_context():
+def fetch_last_upgrade_context(use_context=None):
     """
     :return: Context of the last execution
     """
     with get_connection(None) as db:
-        cursor = db.execute(
-            "SELECT context, stamp, configuration FROM execution WHERE kind = 'upgrade' ORDER BY id DESC LIMIT 1")
+        if use_context:
+            cursor = db.execute(
+                "SELECT context, stamp, configuration FROM execution WHERE context = ?", (use_context,))
+        else:
+            cursor = db.execute(
+                "SELECT context, stamp, configuration FROM execution WHERE kind = 'upgrade' ORDER BY id DESC LIMIT 1")
         row = cursor.fetchone()
         if row:
             return row[0], json.loads(row[2])
@@ -183,6 +192,68 @@ def process_whitelist_experimental(repositories, workflow, configuration, logger
             raise CommandError(msg)
 
 
+@command('rerun', help='Re-runs the upgrade from the given phase and using the information and progress '
+         'from the last invocation of leapp upgrade.')
+@command_arg('from-phase',
+             help='Phase to start running from again. Supported values: {}'.format(', '.join(RERUN_SUPPORTED_PHASES)))
+@command_opt('only-actors-with-tag', action='append', metavar='TagName',
+             help='Restrict actors to be re-run only with given tags. Others will not be executed')
+@command_opt('debug', is_flag=True, help='Enable debug mode', inherit=False)
+@command_opt('verbose', is_flag=True, help='Enable verbose logging', inherit=False)
+def rerun(args):
+
+    if os.environ.get('LEAPP_UNSUPPORTED') != '1':
+        raise CommandError('This command requires the environment variable LEAPP_UNSUPPORTED="1" to be set!')
+
+    if args.from_phase not in RERUN_SUPPORTED_PHASES:
+        raise CommandError('This command is only supported for {}'.format(', '.join(RERUN_SUPPORTED_PHASES)))
+
+    context = str(uuid.uuid4())
+    last_context, configuration = fetch_last_upgrade_context()
+    if args.from_phase not in set([chkpt['phase'] for chkpt in get_checkpoints(context=last_context)]):
+        raise CommandError('Phase {} has not been executed in the last leapp upgrade execution. '
+                           'Cannot rerun not executed phase'.format(args.from_phase))
+
+    if not last_context:
+        raise CommandError('No previous upgrade run to rerun - '
+                           'leapp upgrade has to be run before leapp rerun can be used')
+
+    with get_connection(None) as db:
+        e = Execution(context=context, kind='rerun', configuration=configuration)
+
+        e.store(db)
+
+        clone_context(last_context, context, db)
+        db.execute('''
+            DELETE FROM audit WHERE id IN (
+                SELECT
+                    audit.id          AS id
+                FROM
+                    audit
+                JOIN
+                    data_source ON data_source.id = audit.data_source_id
+                WHERE
+                    audit.context = ? AND audit.event = 'checkpoint'
+                    AND data_source.phase LIKE 'FirstBoot%'
+            );
+        ''', (context,))
+        db.execute('''DELETE FROM message WHERE context = ? and type = 'ErrorModel';''', (context,))
+
+    archive_logfiles()
+    upgrade(Namespace(
+        resume=True,
+        resume_context=context,
+        only_with_tags=args.only_actors_with_tag or [],
+        debug=args.debug,
+        verbose=args.verbose,
+        reboot=False,
+        no_rhsm=False,
+        whitelist_experimental=[],
+        enablerepo=[]))
+
+
+# If you are adding new parameters please ensure that they are set in the upgrade function invocation in `rerun`
+# otherwise there might be errors.
 @command('upgrade', help='Upgrade the current system to the next available major version.')
 @command_opt('resume', is_flag=True, help='Continue the last execution after it was stopped (e.g. after reboot)')
 @command_opt('reboot', is_flag=True, help='Automatically performs reboot when requested.')
@@ -202,11 +273,16 @@ def upgrade(args):
     answerfile_path = cfg.get('report', 'answerfile')
     userchoices_path = cfg.get('report', 'userchoices')
 
+    # Processing of parameters passed by the rerun call, these aren't actually command line arguments
+    # therefore we have to assume that they aren't even in `args` as they are added only by rerun.
+    only_with_tags = args.only_with_tags if 'only_with_tags' in args else None
+    resume_context = args.resume_context if 'resume_context' in args else None
+
     if os.getuid():
         raise CommandError('This command has to be run under the root user.')
 
     if args.resume:
-        context, configuration = fetch_last_upgrade_context()
+        context, configuration = fetch_last_upgrade_context(resume_context)
         if not context:
             raise CommandError('No previous upgrade run to continue, remove `--resume` from leapp invocation to'
                                ' start a new upgrade flow')
@@ -238,7 +314,8 @@ def upgrade(args):
     with beautify_actor_exception():
         logger.info("Using answerfile at %s", answerfile_path)
         workflow.load_answers(answerfile_path, userchoices_path)
-        workflow.run(context=context, skip_phases_until=skip_phases_until, skip_dialogs=True)
+        workflow.run(context=context, skip_phases_until=skip_phases_until, skip_dialogs=True,
+                     only_with_tags=only_with_tags)
 
     logger.info("Answerfile will be created at %s", answerfile_path)
     workflow.save_answers(answerfile_path, userchoices_path)

leapp/utils/audit/contextclone.py

@@ -0,0 +1,84 @@
+from leapp.utils.audit import dict_factory, get_connection
+
+
+def _fetch_table_for_context(db, table, context):
+    cursor = db.execute('''
+            SELECT * FROM {table} WHERE context = ?
+        '''.format(table=table), (context,))
+    cursor.row_factory = dict_factory
+    while True:
+        row = cursor.fetchone()
+        if not row:
+            break
+        yield row
+    del cursor
+
+
+def _row_tuple(row, *fields):
+    return tuple([row[name] for name in fields or row.keys()])
+
+
+def _dup_host(db, newcontext, oldcontext):
+    lookup = {}
+    for row in _fetch_table_for_context(db, 'host', oldcontext):
+        # id, context, hostname
+        row_id, hostname = _row_tuple(row, 'id', 'hostname')
+        cursor = db.execute('INSERT INTO host (context, hostname) VALUES(?, ?)',
+                            (newcontext, hostname))
+        lookup[row_id] = cursor.lastrowid
+    return lookup
+
+
+def _dup_data_source(db, host, newcontext, oldcontext):
+    lookup = {}
+    for row in _fetch_table_for_context(db, 'data_source', oldcontext):
+        # id, context, hostname
+        row_id, host_id, actor, phase = _row_tuple(row, 'id', 'host_id', 'actor', 'phase')
+        cursor = db.execute('INSERT INTO data_source (context, host_id, actor, phase) VALUES(?, ?, ?, ?)',
+                            (newcontext, host[host_id], actor, phase))
+        lookup[row_id] = cursor.lastrowid
+    return lookup
+
+
+def _dup_message(db, data_source, newcontext, oldcontext):
+    lookup = {}
+    for row in _fetch_table_for_context(db, 'message', oldcontext):
+        # id, context, data_source_id, stamp, topic, type, message_data_hash
+        row_id, data_source_id, stamp, topic, type_, message_data_hash = _row_tuple(
+            row, 'id', 'data_source_id', 'stamp', 'topic', 'type', 'message_data_hash')
+        cursor = db.execute(
+            'INSERT INTO message (context, data_source_id, stamp, topic, type, message_data_hash) '
+            ' VALUES(?, ?, ?, ?, ?, ?)',
+            (newcontext, data_source[data_source_id], stamp, topic, type_, message_data_hash))
+        lookup[row_id] = cursor.lastrowid
+    return lookup
+
+
+def _dup_audit(db, message, data_source, newcontext, oldcontext):
+    lookup = {}
+    for row in _fetch_table_for_context(db, 'audit', oldcontext):
+        # id, context, event, stamp, data_source_id, message_id, data
+        row_id, event, stamp, data_source_id, message_id, data = _row_tuple(
+            row, 'id', 'event', 'stamp', 'data_source_id', 'message_id', 'data')
+        if message_id is not None:
+            message_id = message[message_id]
+
+        cursor = db.execute(
+            'INSERT INTO audit (context, event, stamp, data_source_id, message_id, data) VALUES(?, ?, ?, ?, ?, ?)',
+            (newcontext, event, stamp, data_source[data_source_id], message_id, data))
+        lookup[row_id] = cursor.lastrowid
+    return lookup
+
+
+def clone_context(oldcontext, newcontext, use_db=None):
+    # Enter transaction - In case of any exception automatic rollback is issued
+    # and it is automatically committed if there was no exception
+    with get_connection(use_db) as db:
+        # First clone host entries
+        host = _dup_host(db=db, newcontext=newcontext, oldcontext=oldcontext)
+        # Next clone data_source entries and use the lookup table generated by the host duplication
+        data_source = _dup_data_source(db=db, host=host, newcontext=newcontext, oldcontext=oldcontext)
+        # Next clone message entries and use the lookup table generated by the data_source duplication
+        message = _dup_message(db=db, data_source=data_source, newcontext=newcontext, oldcontext=oldcontext)
+        # Last clone message entries and use the lookup table generated by the data_source and message duplications
+        _dup_audit(db=db, data_source=data_source, message=message, newcontext=newcontext, oldcontext=oldcontext)

leapp/workflows/__init__.py

@@ -39,6 +39,17 @@ def phase_names(phase=None):
     return (phase[0].__name__.lower(), phase[0].name.lower()) if phase else ()
 
 
+def tag_names(tag=None):
+    return (tag.__name__.lower(), tag.name.lower()) if tag else ()
+
+
+def contains_tag(needle_tags, actor_tags):
+    hay = set()
+    for tag in actor_tags:
+        hay.update(tag_names(tag))
+    return bool(hay.intersection([tag.lower() for tag in needle_tags]))
+
+
 class WorkflowMeta(type):
     """
     Meta class for the registration of workflows
@@ -230,7 +241,8 @@ def is_valid_phase(self, phase=None):
         if phase:
             return phase in [name for phs in self._phase_actors for name in phase_names(phs)]
 
-    def run(self, context=None, until_phase=None, until_actor=None, skip_phases_until=None, skip_dialogs=False):
+    def run(self, context=None, until_phase=None, until_actor=None, skip_phases_until=None, skip_dialogs=False,
+            only_with_tags=None):
         """
         Executes the workflow
 
@@ -255,6 +267,8 @@ def run(self, context=None, until_phase=None, until_actor=None, skip_phases_unti
                              The value of skip_dialogs will be passed to the actors that can theoretically use it for
                              their purposes.
         :type skip_dialogs: bool
+        :param only_with_tags: Executes only actors with the given tag, any other actor is going to get skipped.
+        :type only_with_tags: List[str]
 
         """
         context = context or str(uuid.uuid4())
@@ -310,6 +324,11 @@ def run(self, context=None, until_phase=None, until_actor=None, skip_phases_unti
                             current_logger.info("Skipping experimental actor {actor}".format(actor=actor.name))
                             continue
 
+                    if only_with_tags and not contains_tag(only_with_tags, actor.tags):
+                        current_logger.info(
+                            "Actor {actor} does not contain any required tag. Skipping.".format(actor=actor.name))
+                        continue
+
                     display_status_current_actor(actor, designation=designation)
                     current_logger.info("Executing actor {actor} {designation}".format(designation=designation,
                                                                                        actor=actor.name))

tests/data/leapp-rerun-tests-repos/ipu-rerun-repo/.leapp/info

@@ -0,0 +1 @@
+{"name": "ipu-rerun-repo", "id": "bae57f03-85e0-4dea-9a6d-88c40205c7a3"}

tests/data/leapp-rerun-tests-repos/ipu-rerun-repo/.leapp/leapp.conf

@@ -0,0 +1,6 @@
+
+[repositories]
+repo_path=${repository:root_dir}
+
+[database]
+path=${repository:state_dir}/leapp.db

tests/data/leapp-rerun-tests-repos/ipu-rerun-repo/actors/firstbootactor/actor.py

@@ -0,0 +1,16 @@
+from leapp.actors import Actor
+from leapp.tags import InplaceUpgradeWorkflowTag, FirstBootTag
+
+
+class FirstBootActor(Actor):
+    """
+    No documentation has been provided for the first_boot_actor actor.
+    """
+
+    name = 'first_boot_actor'
+    consumes = ()
+    produces = ()
+    tags = (InplaceUpgradeWorkflowTag, FirstBootTag)
+
+    def process(self):
+        print('<<<TEST>>>: {}'.format(self.__class__.__name__))

tests/data/leapp-rerun-tests-repos/ipu-rerun-repo/actors/phaseaactor/actor.py

@@ -0,0 +1,16 @@
+from leapp.actors import Actor
+from leapp.tags import InplaceUpgradeWorkflowTag, PhaseATag
+
+
+class PhaseAActor(Actor):
+    """
+    No documentation has been provided for the phase_a_actor actor.
+    """
+
+    name = 'phase_a_actor'
+    consumes = ()
+    produces = ()
+    tags = (InplaceUpgradeWorkflowTag, PhaseATag)
+
+    def process(self):
+        print('<<<TEST>>>: {}'.format(self.__class__.__name__))

tests/data/leapp-rerun-tests-repos/ipu-rerun-repo/actors/phasebactor/actor.py

@@ -0,0 +1,16 @@
+from leapp.actors import Actor
+from leapp.tags import InplaceUpgradeWorkflowTag, PhaseBTag
+
+
+class PhaseBActor(Actor):
+    """
+    No documentation has been provided for the phase_b_actor actor.
+    """
+
+    name = 'phase_b_actor'
+    consumes = ()
+    produces = ()
+    tags = (InplaceUpgradeWorkflowTag, PhaseBTag)
+
+    def process(self):
+        print('<<<TEST>>>: {}'.format(self.__class__.__name__))

tests/data/leapp-rerun-tests-repos/ipu-rerun-repo/actors/rerunactor/actor.py

@@ -0,0 +1,16 @@
+from leapp.actors import Actor
+from leapp.tags import InplaceUpgradeWorkflowTag, FirstBootTag, ReRunVerifyTag
+
+
+class ReRunActor(Actor):
+    """
+    No documentation has been provided for the re_run_actor actor.
+    """
+
+    name = 're_run_actor'
+    consumes = ()
+    produces = ()
+    tags = (InplaceUpgradeWorkflowTag, FirstBootTag, ReRunVerifyTag)
+
+    def process(self):
+        print('<<<TEST>>>: {}'.format(self.__class__.__name__))

tests/data/leapp-rerun-tests-repos/ipu-rerun-repo/actors/rerunactorother/actor.py

@@ -0,0 +1,16 @@
+from leapp.actors import Actor
+from leapp.tags import InplaceUpgradeWorkflowTag, FirstBootTag, ReRunVerifyOtherTag
+
+
+class ReRunActorOther(Actor):
+    """
+    No documentation has been provided for the re_run_actor_other actor.
+    """
+
+    name = 're_run_actor_other'
+    consumes = ()
+    produces = ()
+    tags = (InplaceUpgradeWorkflowTag, FirstBootTag, ReRunVerifyOtherTag)
+
+    def process(self):
+        print('<<<TEST>>>: {}'.format(self.__class__.__name__))

tests/data/leapp-rerun-tests-repos/ipu-rerun-repo/tags/firstboot.py

@@ -0,0 +1,5 @@
+from leapp.tags import Tag
+
+
+class FirstBootTag(Tag):
+    name = 'first_boot'

tests/data/leapp-rerun-tests-repos/ipu-rerun-repo/tags/inplaceupgradeworkflow.py

@@ -0,0 +1,5 @@
+from leapp.tags import Tag
+
+
+class InplaceUpgradeWorkflowTag(Tag):
+    name = 'inplace_upgrade_workflow'