Merge ~sylvain-pineau/checkbox-ng:mini-me into checkbox-ng:remote-api-bump

[Sylvain Pineau]

the checkbox-cli mini-me command (see the commits for details)

How to test?

The standalone command (quick an easy):

$ checkbox-cli mini-me --host 192.168.1.33 com.canonical.certification::usb/detect

Nested inside an existing checkbox remote/service session:

1. Prepare the following launcher (adjust the target host ip of course):

    [launcher]
    app_id = com.canonical.certification:checkbox-test
    launcher_version = 1
    stock_reports = text, submission_files

    [test plan]
    unit = com.canonical.certification::demo
    forced = yes

    [test selection]
    forced = yes

    [ui]
    output = hide-resource-and-attachment

    [environment]
    STRESS_BOOT_ITERATIONS=2
    STRESS_BOOT_WAIT_DELAY=10
    STRESS_BOOT_WAKEUP_DELAY=15
    TARGET_HOST=192.168.1.33

2. A small test plan like this:

    id: demo
    unit: test plan
    _name: demo tests
    _description:
     demo tests
    include:
     demo/usb/detect

3. A demo job calling the new command:

    plugin: shell
    category_id: com.canonical.plainbox::usb
    id: demo/usb/detect
    command:
     checkbox-cli mini-me com.canonical.certification::usb/detect
    _summary: Display USB devices attached to SUT
    _description: Detects and shows USB devices attached to this system.

4. Install checkbox-ng and the checkbox provider from the dev ppa on a spare system

5. From your laptop, start a checkbox service (after adding the test plan and the new job to one of your local providers, side-loading one is a good idea):

    sudo checkbox-cli service

6. Finally execute your launcher file with:

    checkbox-cli master 127.0.0.1 ./mylauncher

If everything goes well, the usb/detect output should list devices from your spare system

[Maciej Kisielewski (kissiel)]

Awesome stuff.

Is this based on remote-api-bump branch? There are some changes that landed there, and may conflict with this.

I think we need a way to mark session as a "mini-me" session. Reason is, if someone started a session with normal remote, connecting to it will result on MiniMe controller picking up the execution with no way of disconnecting (interrupt is bruteforce) and with different handling of things. MiniMe should be able to connect only to Idle services. And also it should have a global lock on the service (with a timeout) - those session should be short lived as I understand it. Also disconnecting controller should terminate this ephemeral session (Unless we want to support reboots as part of mini-me, but I doubt it, as we want this to be ephemeral, stateless thing.

There are a few bits below that need fixing, mostly because the new code is heavily based on existing controller.

Taking into consideration some of the things I noted here I think the whole thing can be greatly simplified.

review: Needs Fixing

[Maciej Kisielewski (kissiel)]

I also think we should s/slave/service and s/master/controller wherever possible, so it's less work in the future, and also to enforce good habits.

[Maciej Kisielewski (kissiel)]

I also don't think that stuff like:
 [test selection]
    forced = yes

should be required. There's no other option to run it, so it should be implied.

Unmerged commits

e1314da... by Sylvain Pineau on 2022-02-11

mini-me: new checkbox-cli command

The "remote" version of the "local" `checkbox-cli run` command.

Supported pattern:
- Only a single job id is supported
- Only automated tests, no manual or semi-auto plugins
- Test plan ids are rejected

Selection of the target host:
- Target machine IP can be provided with either --host or the env var TARGET_HOST

Standard streams
- Remote stdout/stderr are printed locally on the the same local streams
- No colorizer

Session management:
- Ephemeral sessions on service side, removed after job execution
- launcher settings are transferred to the TARGET_HOST

617c9ac... by Sylvain Pineau on 2022-02-11

session:remote_assistant: Expose the hand_pick_jobs() method

71bfd7b... by Sylvain Pineau on 2022-02-11

session:assistant: add a method to save the launcher file in the session dir

save_launcher_file is also exposed over rpyc

4010421... by Sylvain Pineau on 2022-02-11

execution.py: Set a new env var PLAINBOX_SESSION_DIR

Full path to the current session directory

3ef87c1... by Sylvain Pineau on 2022-02-11

execution.py: Set PYTHONUNBUFFERED for all child processes

c865d8f... by Sylvain Pineau on 2021-12-18

fix: restart - Disable the checkbox-ng.service if __respawn_checkbox is used

Disable the service provided by checkbox-ng if another mechanism handles
the session restart.
e.g pm_test currently manages several reboots in a single job.

diff --git a/checkbox_ng/launcher/checkbox_cli.py b/checkbox_ng/launcher/checkbox_cli.py
index 9740eb8..0ae7c14 100644
--- a/checkbox_ng/launcher/checkbox_cli.py
+++ b/checkbox_ng/launcher/checkbox_cli.py
@@ -38,6 +38,7 @@ from checkbox_ng.launcher.check_config import CheckConfig
 from checkbox_ng.launcher.merge_reports import MergeReports
 from checkbox_ng.launcher.merge_submissions import MergeSubmissions
 from checkbox_ng.launcher.master import RemoteMaster
+from checkbox_ng.launcher.mini_me import RemoteMe
 from checkbox_ng.launcher.slave import RemoteSlave
 
 
@@ -67,6 +68,7 @@ def main():
         'tp-export': TestPlanExport,
         'service': RemoteSlave,
         'remote': RemoteMaster,
+        'mini-me': RemoteMe,
     }
     deprecated_commands = {
         'slave': 'service',
diff --git a/checkbox_ng/launcher/mini_me.py b/checkbox_ng/launcher/mini_me.py
new file mode 100644
index 0000000..266ed80
--- /dev/null
+++ b/checkbox_ng/launcher/mini_me.py
@@ -0,0 +1,327 @@
+# This file is part of Checkbox.
+#
+# Copyright 2022 Canonical Ltd.
+# Written by:
+#   Sylvain Pineau <[e-mail]>
+#
+# Checkbox is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Checkbox.  If not, see <http://www.gnu.org/licenses/>.
+"""
+This module contains implementation of the mini me end of the remote execution
+functionality.
+"""
+import contextlib
+import ipaddress
+import json
+import logging
+import os
+import select
+import socket
+import time
+import sys
+
+from functools import partial
+
+from plainbox.impl.session.remote_assistant import RemoteSessionAssistant
+from plainbox.vendor import rpyc
+from checkbox_ng.launcher.stages import MainLoopStage
+from checkbox_ng.launcher.stages import ReportsStage
+from tqdm import tqdm
+_logger = logging.getLogger("mini-me")
+
+
+def environ_or_required(key):
+    """Mapping for argparse to supply required or default from $ENV."""
+    if os.environ.get(key):
+        return {'default': os.environ.get(key)}
+    else:
+        return {'required': True}
+
+
+class RemoteMe(ReportsStage, MainLoopStage):
+    """
+    Control remote slave instance
+
+    This class implements the part that presents UI to the operator and
+    steers the session.
+    """
+
+    name = 'remote-me'
+
+    @property
+    def is_interactive(self):
+        return False
+
+    @property
+    def C(self):
+        return None
+
+    @property
+    def sa(self):
+        return self._sa
+
+    def invoked(self, ctx):
+        self._launcher_text = ''
+        self._is_bootstrapping = False
+        self._target_host = ctx.args.host
+        self._normal_user = ''
+        timeout = 600
+        deadline = time.time() + timeout
+        port = ctx.args.port
+        self.selection = [ctx.args.PATTERN]
+        if not ipaddress.ip_address(ctx.args.host).is_loopback:
+            _logger.info("Connecting to {}:{}. Timeout: {}s".format(
+                ctx.args.host, port, timeout))
+        while time.time() < deadline:
+            try:
+                self.connect_and_run(ctx.args.host, port)
+                break
+            except (ConnectionRefusedError, socket.timeout, OSError):
+                print('.', end='', flush=True)
+                time.sleep(1)
+        else:
+            print("\nConnection timed out.")
+
+    def connect_and_run(self, host, port=18871):
+        config = rpyc.core.protocol.DEFAULT_CONFIG.copy()
+        config['allow_all_attrs'] = True
+        config['sync_request_timeout'] = 120
+        keep_running = False
+        server_msg = None
+        self._prepare_transports()
+        interrupted = False
+        while True:
+            try:
+                if interrupted:
+                    _logger.info("remote: Session interrupted")
+                    interrupted = False  # we are handling the interruption ATM
+                    # next line can raise exception due to connection being
+                    # lost so let's set the default behavior to quitting
+                    keep_running = False
+                    keep_running = self._handle_interrupt()
+                    if not keep_running:
+                        break
+                conn = rpyc.connect(host, port, config=config)
+                keep_running = True
+
+                def quitter(msg):
+                    # this will be called when the slave decides to disconnect
+                    # this master
+                    nonlocal server_msg
+                    nonlocal keep_running
+                    keep_running = False
+                    server_msg = msg
+                with contextlib.suppress(AttributeError):
+                    # TODO: REMOTE_API
+                    # when bumping the remote api make this bit obligatory
+                    # i.e. remove the suppressing
+                    conn.root.register_master_blaster(quitter)
+                self._sa = conn.root.get_sa()
+                self.sa.conn = conn
+                # TODO: REMOTE API RAPI: Remove this API on the next RAPI bump
+                # the check and bailout is not needed if the slave as up to
+                # date as this master, so after bumping RAPI we can assume
+                # that slave is always passwordless
+                if not self.sa.passwordless_sudo:
+                    raise SystemExit(
+                        "This version of Checkbox requires the service"
+                        " to be run as root")
+                try:
+                    slave_api_version = self.sa.get_remote_api_version()
+                except AttributeError:
+                    raise SystemExit("Service doesn't declare Remote API"
+                                     " version. Update Checkbox on the"
+                                     " SUT!")
+                master_api_version = RemoteSessionAssistant.REMOTE_API_VERSION
+                if slave_api_version != master_api_version:
+                    raise SystemExit(
+                        "Remote API version mismatch. Service "
+                        "uses: {}. Remote uses: {}".format(
+                            slave_api_version, master_api_version))
+                state, payload = self.sa.whats_up()
+                _logger.info("remote: Main dispatch with state: %s", state)
+                keep_running = {
+                    'idle': self.new_session,
+                    'running': self.wait_and_continue,
+                    'finalizing': self.finish_session,
+                    'testsselected': partial(
+                        self.run_jobs, resumed_session_info=payload),
+                    'bootstrapping': self.restart,
+                    'started': self.restart,
+                    'interacting': partial(
+                        self.resume_interacting, interaction=payload),
+                }[state]()
+            except EOFError as exc:
+                if keep_running:
+                    print("Connection lost!")
+                    # this is yucky but it works, in case of explicit
+                    # connection closing by the slave we get this msg
+                    _logger.info("remote: Connection lost due to: %s", exc)
+                    if str(exc) == 'stream has been closed':
+                        print('Service explicitly disconnected you. Possible '
+                              'reason: new remote connected to the service')
+                        break
+                    print(exc)
+                    time.sleep(1)
+                else:
+                    # if keep_running got set to False it means that the
+                    # network interruption was planned, AKA slave disconnected
+                    # this master
+                    print(server_msg)
+                    break
+            except (ConnectionRefusedError, socket.timeout, OSError) as exc:
+                _logger.info("remote: Connection lost due to: %s", exc)
+                if not keep_running:
+                    raise
+                # it's reconnecting, so we can ignore refuses
+                print('.', flush=True)
+                time.sleep(0.5)
+            except KeyboardInterrupt:
+                interrupted = True
+
+            if not keep_running:
+                break
+
+    def new_session(self):
+        _logger.info("remote: Starting new session.")
+        configuration = dict()
+        session_dir = os.environ.get('PLAINBOX_SESSION_DIR', '/tmp')
+        launcher_file = os.path.join(session_dir, 'launcher')
+        if os.path.isfile(launcher_file):
+            with open(launcher_file, 'rt', encoding='UTF-8') as f:
+                self._launcher_text = f.read()
+        configuration['launcher'] = self._launcher_text
+        configuration['normal_user'] = self._normal_user
+        self.sa.start_session(configuration)
+        self.sa.hand_pick_jobs(self.selection)
+        self.run_jobs()
+
+    def register_arguments(self, parser):
+        parser.add_argument('--host', help="target host",
+                        **environ_or_required('TARGET_HOST'))
+        parser.add_argument('--port', type=int, default=18871, help=(
+            "port to connect to"))
+        parser.add_argument(
+            'PATTERN',
+            help="run a job matching the given regular expression")
+
+    def _handle_interrupt(self):
+        """
+        Returns True if the remote should keep running.
+        And False if it should quit.
+        """
+        self._sa.terminate()
+        return False
+
+    def cleanup_ephemeral_session(self):
+        storage = self.sa.manager.storage
+        self.sa.finalize_session()
+        storage.remove()
+
+    def finish_session(self):
+        self.cleanup_ephemeral_session()
+        return_code = self.job_result.return_code
+        if return_code:
+            raise SystemExit(return_code)
+
+    def wait_and_continue(self):
+        progress = self.sa.whats_up()[1]
+        print("Rejoined session.")
+        print("In progress: {} ({}/{})".format(
+            progress[2], progress[0], progress[1]))
+        self.wait_for_job()
+        self.run_jobs()
+
+    def _handle_last_job_after_resume(self, resumed_session_info):
+        jobs_repr = json.loads(
+            self.sa.get_jobs_repr([resumed_session_info['last_job']]))
+        job = jobs_repr[-1]
+        self.job_result = self.sa.get_job_result(job['id'])
+
+    def run_jobs(self, resumed_session_info=None):
+        if resumed_session_info and resumed_session_info['last_job']:
+            self._handle_last_job_after_resume(resumed_session_info)
+        _logger.info("remote: Running jobs.")
+        jobs = self.sa.get_session_progress()
+        _logger.debug("remote: Jobs to be run:\n%s",
+                      '\n'.join(['  ' + job for job in jobs]))
+        total_num = len(jobs['done']) + len(jobs['todo'])
+        if total_num > 1:
+            self.cleanup_ephemeral_session()
+            raise SystemExit("More than one job to run!")
+        if total_num == 0:
+            self.cleanup_ephemeral_session()
+            raise SystemExit(
+                "No job found with this id: {}".format(self.selection[0]))
+        jobs_repr = json.loads(
+            self.sa.get_jobs_repr(jobs['todo'], len(jobs['done'])))
+
+        self._run_jobs(jobs_repr, total_num)
+        self.finish_session()
+
+    def resume_interacting(self, interaction):
+        self.sa.remember_users_response('rollback')
+        self.run_jobs()
+
+    def wait_for_job(self, dont_finish=False):
+        _logger.info("remote: Waiting for job to finish.")
+        while True:
+            state, payload = self.sa.monitor_job()
+            if payload and not self._is_bootstrapping:
+                for line in payload.splitlines():
+                    if line.startswith('stderr'):
+                        print(line[6:], file=sys.stderr)
+                    else:
+                        print(line[6:])
+            if state == 'running':
+                time.sleep(0.5)
+                while True:
+                    res = select.select([sys.stdin], [], [], 0)
+                    if not res[0]:
+                        break
+                    # XXX: this assumes that sys.stdin is chunked in lines
+                    buff = res[0][0].readline()
+                    self.sa.transmit_input(buff)
+                    if not buff:
+                        break
+            else:
+                if dont_finish:
+                    return
+                self.finish_job()
+                break
+
+    def finish_job(self, result=None):
+        _logger.info("remote: Finishing job with a result: %s", result)
+        self.job_result = self.sa.finish_job(result)
+
+    def abandon(self):
+        _logger.info("remote: Abandoning session.")
+        self.sa.finalize_session()
+
+    def restart(self):
+        _logger.info("remote: Restarting session.")
+        self.abandon()
+        self.new_session()
+
+    def _run_jobs(self, jobs_repr, total_num=0):
+        for job in jobs_repr:
+            # print("ID: {0}".format(job['id']))
+            if 'user' in job['plugin'] or 'manual' in job['plugin']:
+                self.cleanup_ephemeral_session()
+                raise SystemExit("Not an automated job: {0}".format(job['id']))
+            next_job = False
+            while next_job is False:
+                list(self.sa.run_job(job['id']))
+                self.wait_for_job()
+                break
+            if next_job:
+                continue
diff --git a/plainbox/impl/execution.py b/plainbox/impl/execution.py
index e1aa65b..06c19a1 100644
--- a/plainbox/impl/execution.py
+++ b/plainbox/impl/execution.py
@@ -468,6 +468,11 @@ def get_execution_environment(job, environ, session_id, nest_dir):
     # Add per-session shared state directory
     env['PLAINBOX_SESSION_SHARE'] = WellKnownDirsHelper.session_share(
         session_id)
+    # Add the session directory
+    env['PLAINBOX_SESSION_DIR'] = WellKnownDirsHelper.session_dir(
+        session_id)
+    # Force the stdout and stderr streams to be unbuffered
+    env['PYTHONUNBUFFERED'] = "1"
 
     def set_if_not_none(envvar, source):
         """Update env if the source variable is not None"""
@@ -542,6 +547,8 @@ def get_differential_execution_environment(job, environ, session_id, nest_dir,
         for key, value in base_env.items():
             if key in copy_vars or key.startswith('SNAP'):
                 delta_env[key] = value
+    # Force the stdout and stderr streams to be unbuffered
+    delta_env['PYTHONUNBUFFERED'] = "1"
     return delta_env
 
 
diff --git a/plainbox/impl/session/assistant.py b/plainbox/impl/session/assistant.py
index c560828..253148e 100644
--- a/plainbox/impl/session/assistant.py
+++ b/plainbox/impl/session/assistant.py
@@ -613,6 +613,12 @@ class SessionAssistant:
         self._context.state.metadata.app_blob = updated_blob
         self._manager.checkpoint()
 
+    def save_launcher_file(self, launcher_text):
+        session_dir = WellKnownDirsHelper.session_dir(self.get_session_id())
+        launcher_file = os.path.join(session_dir, 'launcher')
+        with open(launcher_file, 'wt', encoding='UTF-8') as f:
+                f.write(launcher_text)
+
     @morris.signal
     def session_available(self, session_id):
         """
diff --git a/plainbox/impl/session/remote_assistant.py b/plainbox/impl/session/remote_assistant.py
index 96f5de4..b767722 100644
--- a/plainbox/impl/session/remote_assistant.py
+++ b/plainbox/impl/session/remote_assistant.py
@@ -31,6 +31,7 @@ from subprocess import CalledProcessError, check_output
 
 from plainbox.impl.config import Configuration
 from plainbox.impl.execution import UnifiedRunner
+from plainbox.impl.session import SessionMetaData
 from plainbox.impl.session.assistant import SessionAssistant
 from plainbox.impl.session.assistant import SA_RESTARTABLE
 from plainbox.impl.session.jobs import InhibitionCause
@@ -185,6 +186,9 @@ class RemoteSessionAssistant():
     def update_app_blob(self, app_blob):
         self._sa.update_app_blob(app_blob)
 
+    def save_launcher_file(self, launcher_text):
+        self._sa.save_launcher_file(launcher_text)
+
     def allowed_when(*states):
         def wrap(f):
             def fun(self, *args):
@@ -301,6 +305,7 @@ class RemoteSessionAssistant():
             'effective_normal_user': self._normal_user,
         }).encode("UTF-8")
         self._sa.update_app_blob(new_blob)
+        self._sa.save_launcher_file(configuration['launcher'])
         self._sa.configure_application_restart(self._cmd_callback)
 
         self._session_id = self._sa.get_session_id()
@@ -333,6 +338,11 @@ class RemoteSessionAssistant():
     def get_bootstrapping_todo_list(self):
         return self._sa.get_bootstrap_todo_list()
 
+    @allowed_when(Started)
+    def hand_pick_jobs(self, id_patterns):
+        self._sa.hand_pick_jobs(id_patterns)
+        self._state = TestsSelected
+
     def finish_bootstrap(self):
         self._sa.finish_bootstrap()
         self._state = Bootstrapped
@@ -646,19 +656,19 @@ class RemoteSessionAssistant():
             'extra_env': self.prepare_extra_env,
         }
         meta = self._sa.resume_session(session_id, runner_kwargs=runner_kwargs)
-        app_blob = json.loads(meta.app_blob.decode("UTF-8"))
-        launcher = app_blob['launcher']
-        self._launcher = Configuration.from_text(launcher, 'Remote launcher')
-        self._sa.use_alternate_configuration(self._launcher)
-
-        self._normal_user = app_blob.get(
-            'effective_normal_user', self._launcher.get_value(
-                'daemon', 'normal_user'))
-        _logger.info(
-            "normal_user after loading metadata: %r", self._normal_user)
-        test_plan_id = app_blob['testplan_id']
-        self._sa.select_test_plan(test_plan_id)
-        self._sa.bootstrap()
+        if SessionMetaData.FLAG_TESTPLANLESS not in meta.flags:
+            app_blob = json.loads(meta.app_blob.decode("UTF-8"))
+            launcher = app_blob['launcher']
+            self._launcher = Configuration.from_text(launcher, 'Remote launcher')
+            self._sa.use_alternate_configuration(self._launcher)
+            self._normal_user = app_blob.get(
+                'effective_normal_user', self._launcher.get_value(
+                    'daemon', 'normal_user'))
+            _logger.info(
+                "normal_user after loading metadata: %r", self._normal_user)
+            test_plan_id = app_blob['testplan_id']
+            self._sa.select_test_plan(test_plan_id)
+            self._sa.bootstrap()
         self._last_job = meta.running_job_name
 
         result_dict = {
diff --git a/plainbox/impl/session/restart.py b/plainbox/impl/session/restart.py
index b3c72cd..a79741b 100644
--- a/plainbox/impl/session/restart.py
+++ b/plainbox/impl/session/restart.py
@@ -255,7 +255,7 @@ class RemoteDebRestartStrategy(RemoteSnappyRestartStrategy):
         with open(self.session_resume_filename, 'wt') as f:
             f.write(session_id)
             os.fsync(f.fileno())
-        if cmd == self.service_name:
+        if cmd != self.service_name:
             subprocess.call(['systemctl', 'disable', self.service_name])