migration: Support multiple hosts migration

Signed-off-by: Yongxue Hong <[email protected]>

avocado_vt/plugins/vt_bootstrap.py

@@ -142,6 +142,15 @@ def configure(self, parser):
                 "generating the host configuration entry."
             ),
         )
+        parser.add_argument(
+            "--vt-cluster-config",
+            action="store",
+            metavar="CLUSTER_CONFIG",
+            help=(
+                "The cluster config json file to be used when "
+                "generating the cluster hosts configuration entry."
+            ),
+        )
 
     def run(self, config):
         try:

avocado_vt/plugins/vt_cluster.py

@@ -0,0 +1,53 @@
+import logging
+import os
+import sys
+
+from avocado.core import exit_codes
+from avocado.core.plugin_interfaces import JobPostTests as Post
+from avocado.core.plugin_interfaces import JobPreTests as Pre
+from avocado.utils.stacktrace import log_exc_info
+
+from virttest.vt_cluster import cluster, node_metadata
+
+
+class ClusterCreationError(Exception):
+    """
+    Represents any error situation when attempting to create a cluster.
+    """
+
+    pass
+
+
+class VTCluster(Pre, Post):
+
+    name = "vt-cluster"
+    description = "Avocado-VT Cluster Pre/Post"
+
+    def __init__(self, **kwargs):
+        self._log = logging.getLogger("avocado.app")
+
+    def pre_tests(self, job):
+        if cluster.get_all_nodes():
+            try:
+                for node in cluster.get_all_nodes():
+                    node.start_agent_server()
+                node_metadata.load_metadata()
+
+            except Exception as detail:
+                msg = "Failure trying to set Avocado-VT job env: %s" % detail
+                self._log.error(msg)
+                log_exc_info(sys.exc_info(), self._log.name)
+                sys.exit(exit_codes.AVOCADO_JOB_FAIL | job.exitcode)
+
+    def post_tests(self, job):
+        if cluster.get_all_nodes():
+            cluster_dir = os.path.join(job.logdir, "cluster")
+            for node in cluster.get_all_nodes():
+                try:
+                    node_dir = os.path.join(cluster_dir, node.name)
+                    os.makedirs(node_dir)
+                    node.upload_agent_log(node_dir)
+                    node.stop_agent_server()
+                except Exception:
+                    pass
+            node_metadata.unload_metadata()

avocado_vt/test.py

@@ -36,6 +36,7 @@
     version,
 )
 from virttest._wrappers import load_source
+from virttest.vt_cluster import cluster, logger, selector
 
 # avocado-vt no longer needs autotest for the majority of its functionality,
 # except by:
@@ -115,6 +116,10 @@ def __init__(self, **kwargs):
         utils_logfile.set_log_file_dir(self.logdir)
         self.__status = None
         self.__exc_info = None
+        self._cluster_partition = None
+        self._logger_server = logger.LoggerServer(
+            cluster.logger_server_host, cluster.logger_server_port, self.log
+        )
 
     @property
     def params(self):
@@ -262,6 +267,10 @@ def _runTest(self):
 
         try:
             try:
+                self._init_partition()
+                self._setup_partition()
+                self._logger_server.start()
+                self._start_logger_client()
                 try:
                     # Pre-process
                     try:
@@ -321,6 +330,14 @@ def _runTest(self):
                         or params.get("env_cleanup", "no") == "yes"
                     ):
                         env.destroy()  # Force-clean as it can't be stored
+                    self._stop_logger_client()
+                    self._logger_server.stop()
+                    self._clear_partition()
+                    if (
+                        self._safe_env_save(env)
+                        or params.get("env_cleanup", "no") == "yes"
+                    ):
+                        env.destroy()  # Force-clean as it can't be stored
 
         except Exception as e:
             if params.get("abort_on_error") != "yes":
@@ -345,3 +362,55 @@ def _runTest(self):
                 raise exceptions.JobError("Abort requested (%s)" % e)
 
         return test_passed
+
+    def _init_partition(self):
+        self._cluster_partition = cluster.create_partition()
+
+    def _setup_partition(self):
+        for node in self.params.objects("nodes"):
+            node_params = self.params.object_params(node)
+            node_selectors = node_params.get("node_selectors")
+            _node = selector.select_node(cluster.free_nodes, node_selectors)
+            if not _node:
+                raise selector.SelectorError(
+                    f'No available host for node "{node}" with "{node_selectors}"'
+                )
+            _node.tag = node
+            self._cluster_partition.add_node(_node)
+
+    def _clear_partition(self):
+        cluster_dir = os.path.join(self.resultsdir, "cluster")
+        if self._cluster_partition.nodes:
+            for node in self._cluster_partition.nodes:
+                node_dir = os.path.join(cluster_dir, node.tag)
+                os.makedirs(node_dir)
+                node.upload_service_log(node_dir)
+            cluster.clear_partition(self._cluster_partition)
+        self._cluster_partition = None
+
+    def _start_logger_client(self):
+        if self._cluster_partition.nodes:
+            for node in self._cluster_partition.nodes:
+                try:
+                    node.proxy.api.start_logger_client(
+                        cluster.logger_server_host, cluster.logger_server_port
+                    )
+                except ModuleNotFoundError:
+                    pass
+
+    def _stop_logger_client(self):
+        if self._cluster_partition.nodes:
+            for node in self._cluster_partition.nodes:
+                try:
+                    node.proxy.api.stop_logger_client()
+                except ModuleNotFoundError:
+                    pass
+
+    @property
+    def nodes(self):
+        return self._cluster_partition.nodes
+
+    def get_node(self, node_tag):
+        for node in self._cluster_partition.nodes:
+            if node_tag == node.tag:
+                return node

examples/tests/vt_node_test.cfg

@@ -0,0 +1,11 @@
+- vt_node_test:
+    type = vt_node_test
+    start_vm = no
+    not_preprocess = yes
+    nodes = node1 node2 node3
+    node_selectors_node1 = [{"key": "cpu_vendor_id", "operator": "eq", "values": "AuthenticAMD"},
+    node_selectors_node1 += {"key": "hostname", "operator": "contains", "values": "redhat.com"}]
+    node_selectors_node2 = [{"key": "cpu_vendor_id", "operator": "==", "values": "AuthenticAMD"},
+    node_selectors_node2 += {"key": "hostname", "operator": "contains", "values": "redhat.com"}]
+    node_selectors_node3 = [{"key": "cpu_vendor_id", "operator": "==", "values": "AuthenticAMD"},
+    node_selectors_node3 += {"key": "hostname", "operator": "contains", "values": "redhat.com"}]

examples/tests/vt_node_test.py

@@ -0,0 +1,14 @@
+"""
+Simple vt node handling test.
+
+Please put the configuration file vt_node_test.cfg into $tests/cfg/ directory.
+
+"""
+
+
+def run(test, params, env):
+    for node in test.nodes:
+        test.log.info("========Start test on %s========", node.name)
+        node.proxy.unittest.hello.say()
+        node.proxy.unittest.testcase.vm.boot_up()
+        test.log.info("========End test========")

setup.py

@@ -82,6 +82,7 @@ def run(self):
             ],
             "avocado.plugins.result_events": [
                 "vt-joblock = avocado_vt.plugins.vt_joblock:VTJobLock",
+                "vt-cluster = avocado_vt.plugins.vt_cluster:VTCluster",
             ],
             "avocado.plugins.init": [
                 "vt-init = avocado_vt.plugins.vt_init:VtInit",

virttest/bootstrap.py

@@ -1,4 +1,5 @@
 import glob
+import json
 import logging
 import os
 import re
@@ -10,6 +11,8 @@
 from avocado.utils import path as utils_path
 from avocado.utils import process
 
+from virttest.vt_cluster import cluster, node
+
 from . import arch, asset, cartesian_config, data_dir, defaults, utils_selinux
 from .compat import get_opt
 
@@ -875,6 +878,33 @@ def verify_selinux(datadir, imagesdir, isosdir, tmpdir, interactive, selinux=Fal
             LOG.info("Corrected contexts on %d files/dirs", len(changes))
 
 
+def _load_cluster_config(cluster_config):
+    """Load the cluster config"""
+    with open(cluster_config, "r") as config:
+        return json.load(config)
+
+
+def _register_hosts(hosts_configs):
+    """Register the configs of the hosts into the cluster."""
+    if hosts_configs:
+        for host, host_params in hosts_configs.items():
+            _node = node.Node(host_params, host)
+            _node.setup_agent_env()
+            cluster.register_node(_node.name, _node)
+            LOG.debug("Host %s registered", host)
+
+
+def _config_master_server(master_config):
+    """Configure the master server."""
+    if master_config:
+        logger_server_host = master_config.get("logger_server_host")
+        if logger_server_host:
+            cluster.assign_logger_server_host(logger_server_host)
+        logger_server_port = master_config.get("logger_server_port")
+        if logger_server_port:
+            cluster.assign_logger_server_port(logger_server_port)
+
+
 def bootstrap(options, interactive=False):
     """
     Common virt test assistant module.
@@ -1042,6 +1072,18 @@ def bootstrap(options, interactive=False):
             else:
                 LOG.debug("Module %s loaded", module)
 
+    # Setup the cluster environment.
+    vt_cluster_config = get_opt(options, "vt_cluster_config")
+    if vt_cluster_config:
+        LOG.info("")
+        step += 1
+        LOG.info(
+            "%d - Setting up the cluster environment via %s", step, vt_cluster_config
+        )
+        cluster_config = _load_cluster_config(vt_cluster_config)
+        _register_hosts(cluster_config.get("hosts"))
+        _config_master_server(cluster_config.get("master"))
+
     LOG.info("")
     LOG.info("VT-BOOTSTRAP FINISHED")
     LOG.debug("You may take a look at the following online docs for more info:")

virttest/vt_agent/__init__.py

@@ -0,0 +1,10 @@
+import os
+import sys
+
+BASE_DIR = os.path.dirname(__file__)
+LOG_DIR = os.path.join(BASE_DIR, "log")
+AGENT_LOG_FILENAME = os.path.join(LOG_DIR, "agent.log")
+SERVICE_LOG_FILENAME = os.path.join(LOG_DIR, "service.log")
+LOG_FORMAT = "%(asctime)s %(name)s %(levelname)-5.5s| %(message)s"
+
+sys.path.append(BASE_DIR)

virttest/vt_agent/__main__.py

@@ -0,0 +1,40 @@
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program 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 LICENSE for more details.
+#
+# Copyright: Red Hat Inc. 2022
+# Authors: Yongxue Hong <[email protected]>
+
+"""
+Main entry point when called by 'python -m'.
+"""
+
+import os
+import shutil
+
+from . import LOG_DIR
+from .app.args import init_arguments
+from .app.cmd import run
+from .core.logger import init_logger
+
+args = init_arguments()
+
+try:
+    shutil.rmtree(LOG_DIR)
+    os.remove(args.pid_file)
+except (FileNotFoundError, OSError):
+    pass
+
+os.makedirs(LOG_DIR)
+
+root_logger = init_logger()
+
+if __name__ == "__main__":
+    run(args.host, args.port, args.pid_file)

virttest/vt_agent/api.py

@@ -0,0 +1,71 @@
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program 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 LICENSE for more details.
+#
+# Copyright: Red Hat Inc. 2022
+# Authors: Yongxue Hong <[email protected]>
+
+import logging.handlers
+import os
+import signal
+
+from . import AGENT_LOG_FILENAME, LOG_FORMAT, SERVICE_LOG_FILENAME
+
+LOG = logging.getLogger("avocado.agent." + __name__)
+
+
+def quit():
+    """Quit the agent server."""
+    pid = os.getpid()
+    LOG.info("Quit the server daemon(PID:%s).", pid)
+    os.kill(pid, signal.SIGKILL)
+
+
+def is_alive():
+    """Check whether the agent server is alive."""
+    LOG.info("The server daemon is alive.")
+    return True
+
+
+def start_logger_client(host, port):
+    """Start the agent logger client"""
+    try:
+        os.remove(SERVICE_LOG_FILENAME)
+    except FileNotFoundError:
+        pass
+
+    logger = logging.getLogger("avocado.service")
+    logger.setLevel(logging.DEBUG)
+    file_handler = logging.FileHandler(filename=SERVICE_LOG_FILENAME)
+    file_handler.setFormatter(logging.Formatter(fmt=LOG_FORMAT))
+    logger.addHandler(file_handler)
+
+    LOG.info("Start the logger client.")
+    socket_handler = logging.handlers.SocketHandler(host, port)
+    socket_handler.setLevel(logging.DEBUG)
+    logger.addHandler(socket_handler)
+
+
+def stop_logger_client():
+    """Stop the agent logger client."""
+    LOG.info("Stop the logger client.")
+    for handler in logging.getLogger("avocado.service").handlers:
+        handler.close()
+    logging.getLogger("avocado.service").handlers.clear()
+
+
+def get_agent_log_filename():
+    """Get the filename of the agent log."""
+    return AGENT_LOG_FILENAME
+
+
+def get_service_log_filename():
+    """Get the filename of the service log."""
+    return SERVICE_LOG_FILENAME