Add storage and image management support

Signed-off-by: Zhenchao Liu <zhencliu@redhat.com>

Author: Zhenchao Liu

avocado_vt/plugins/vt_cluster.py

@@ -6,7 +6,10 @@
 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
+from virttest.vt_imgr import imgr
+from virttest.vt_resmgr import resmgr
 
 
 class ClusterSetupError(Exception):
@@ -50,21 +53,17 @@ def _pre_node_setup():
     def _pre_mgr_setup():
         try:
             # Pre-setup the cluster manager
-            # e.g:
-            # startup_resmgr()
-            # vt_imgr.startup()
-            pass
+            resmgr.startup()
+            imgr.startup()
         except Exception as err:
             raise ClusterManagerSetupError(err)
 
     @staticmethod
     def _post_mgr_cleanup():
         try:
             # Post-cleanup the cluster manager
-            # e.g:
-            # teardown_resmgr()
-            # vt_imgr.teardown()
-            pass
+            imgr.teardown()
+            resmgr.teardown()
         except Exception as err:
             raise ClusterManagerCleanupError(err)
 

virttest/bootstrap.py

@@ -12,6 +12,7 @@
 from avocado.utils import process
 
 from virttest.vt_cluster import cluster, node
+from virttest.vt_resmgr import resmgr
 
 from . import arch, asset, cartesian_config, data_dir, defaults, utils_selinux
 from .compat import get_opt
@@ -895,6 +896,10 @@ def _register_hosts(hosts_configs):
             LOG.debug("Host %s registered", host)
 
 
+def _initialize_managers(pools_params):
+    resmgr.setup(pools_params)
+
+
 def _config_master_server(master_config):
     """Configure the master server."""
     if master_config:
@@ -1084,6 +1089,7 @@ def bootstrap(options, interactive=False):
         cluster_config = _load_cluster_config(vt_cluster_config)
         _register_hosts(cluster_config.get("hosts"))
         _config_master_server(cluster_config.get("master"))
+        _initialize_managers(cluster_config.get("pools"))
 
     LOG.info("")
     LOG.info("VT-BOOTSTRAP FINISHED")

virttest/env_process.py

@@ -68,6 +68,7 @@
 from virttest.test_setup.storage import StorageConfig
 from virttest.test_setup.verify import VerifyHostDMesg
 from virttest.test_setup.vms import ProcessVMOff, UnrequestedVMHandler
+from virttest.vt_imgr import imgr
 
 utils_libvirtd = lazy_import("virttest.utils_libvirtd")
 virsh = lazy_import("virttest.virsh")
@@ -112,42 +113,72 @@
 LOG = logging.getLogger("avocado." + __name__)
 
 
-def preprocess_image(test, params, image_name, vm_process_status=None):
+def preprocess_image(test, params, image_name, vm_process_status=None, vm_name=None):
     """
     Preprocess a single QEMU image according to the instructions in params.
 
     :param test: Autotest test object.
     :param params: A dict containing image preprocessing parameters.
     :param vm_process_status: This is needed in postprocess_image. Add it here
                               only for keep it work with process_images()
+    :param vm_name: vm tag defined in 'vms'
     :note: Currently this function just creates an image if requested.
     """
+    # FIXME:
+    image_id = None
+    if params.get_boolean("multihost"):
+        params = params.copy()
+        # 'images' is the vm's param, e.g. 'images_vm1', so when the 'vms' is
+        # set, the images defined in 'images' should be used by the vm
+        params[f"image_owner_{image_name}"] = vm_name
+        image_config = imgr.define_logical_image_config(image_name, params)
+        image_id = imgr.create_logical_image_object(image_config)
+
+    params = params.object_params(image_name)
     base_dir = params.get("images_base_dir", data_dir.get_data_dir())
 
     if not storage.preprocess_image_backend(base_dir, params, image_name):
         LOG.error("Backend can't be prepared correctly.")
 
-    image_filename = storage.get_image_filename(params, base_dir)
+    image_filename = None
+    if not params.get_boolean("multihost"):
+        image_filename = storage.get_image_filename(params, base_dir)
 
     create_image = False
     if params.get("force_create_image") == "yes":
         create_image = True
-    elif params.get("create_image") == "yes" and not storage.file_exists(
-        params, image_filename
-    ):
-        create_image = True
+    elif params.get("create_image") == "yes":
+        # FIXME: check all volumes allocated
+        if params.get_boolean("multihost"):
+            volume = imgr.get_logical_image_info(
+                image_id, request=f"spec.images.{image_name}.spec.volume.meta"
+            )
+            create_image = True if not volume["meta"]["allocated"] else False
+        else:
+            create_image = (
+                True if not storage.file_exists(params, image_filename) else False
+            )
+    else:
+        # FIXME: sync all volumes configurations
+        if params.get_boolean("multihost"):
+            imgr.get_logical_image_info(image_id)
+        # TODO: check if file allocated
 
     if params.get("backup_image_before_testing", "no") == "yes":
+        # FIXME: add backup_image
         image = qemu_storage.QemuImg(params, base_dir, image_name)
         image.backup_image(params, base_dir, "backup", True, True)
     if create_image:
-        if storage.file_exists(params, image_filename):
-            # As rbd image can not be covered, so need remove it if we need
-            # force create a new image.
-            storage.file_remove(params, image_filename)
-        image = qemu_storage.QemuImg(params, base_dir, image_name)
-        LOG.info("Create image on %s." % image.storage_type)
-        image.create(params)
+        if params.get_boolean("multihost"):
+            imgr.update_logical_image(image_id, {"create": {}})
+        else:
+            if storage.file_exists(params, image_filename):
+                # As rbd image can not be covered, so need remove it if we need
+                # force create a new image.
+                storage.file_remove(params, image_filename)
+            image = qemu_storage.QemuImg(params, base_dir, image_name)
+            LOG.info("Create image on %s." % image.storage_type)
+            image.create(params)
 
 
 def preprocess_fs_source(test, params, fs_name, vm_process_status=None):
@@ -432,7 +463,7 @@ def preprocess_vm(test, params, env, name):
             )
 
 
-def check_image(test, params, image_name, vm_process_status=None):
+def check_image(test, params, image_name, vm_process_status=None, vm_name=None):
     """
     Check a single QEMU image according to the instructions in params.
 
@@ -441,6 +472,7 @@ def check_image(test, params, image_name, vm_process_status=None):
     :param vm_process_status: (optional) vm process status like running, dead
                               or None for no vm exist.
     """
+    params = params.object_params(image_name)
     clone_master = params.get("clone_master", None)
     base_dir = data_dir.get_data_dir()
     check_image_flag = params.get("check_image") == "yes"
@@ -511,7 +543,7 @@ def check_image(test, params, image_name, vm_process_status=None):
                 raise e
 
 
-def postprocess_image(test, params, image_name, vm_process_status=None):
+def postprocess_image(test, params, image_name, vm_process_status=None, vm_name=None):
     """
     Postprocess a single QEMU image according to the instructions in params.
 
@@ -528,6 +560,16 @@ def postprocess_image(test, params, image_name, vm_process_status=None):
         )
         return
 
+    # FIXME: multihost
+    image_id = None
+    if params.get_boolean("multihost"):
+        image_id = imgr.query_logical_image(image_name, vm_name)
+        if image_id is None:
+            LOG.warning(f"Cannot find the image {image_name}")
+            image_config = imgr.define_logical_image_config(image_name, params)
+            image_id = imgr.create_logical_image_object(image_config)
+    params = params.object_params(image_name)
+
     restored, removed = (False, False)
     clone_master = params.get("clone_master", None)
     base_dir = params.get("images_base_dir", data_dir.get_data_dir())
@@ -585,10 +627,18 @@ def postprocess_image(test, params, image_name, vm_process_status=None):
         )
         LOG.info("Remove image on %s." % image.storage_type)
         if clone_master is None:
-            image.remove()
+            if params.get_boolean("multihost"):
+                imgr.update_logical_image(image_id, {"destroy": {}})
+                imgr.destroy_logical_image_object(image_id)
+            else:
+                image.remove()
         elif clone_master == "yes":
             if image_name in params.get("master_images_clone").split():
-                image.remove()
+                if params.get_boolean("multihost"):
+                    imgr.update_logical_image(image_id, {"destroy": {}})
+                    imgr.destroy_logical_image_object(image_id)
+                else:
+                    image.remove()
 
 
 def postprocess_fs_source(test, params, fs_name, vm_process_status=None):
@@ -742,7 +792,16 @@ class _CreateImages(threading.Thread):
     in self.exc_info
     """
 
-    def __init__(self, image_func, test, images, params, exit_event, vm_process_status):
+    def __init__(
+        self,
+        image_func,
+        test,
+        images,
+        params,
+        exit_event,
+        vm_process_status,
+        vm_name=None,
+    ):
         threading.Thread.__init__(self)
         self.image_func = image_func
         self.test = test
@@ -751,6 +810,7 @@ def __init__(self, image_func, test, images, params, exit_event, vm_process_stat
         self.exit_event = exit_event
         self.exc_info = None
         self.vm_process_status = vm_process_status
+        self.vm_name = vm_name
 
     def run(self):
         try:
@@ -761,13 +821,14 @@ def run(self):
                 self.params,
                 self.exit_event,
                 self.vm_process_status,
+                self.vm_name,
             )
         except Exception:
             self.exc_info = sys.exc_info()
             self.exit_event.set()
 
 
-def process_images(image_func, test, params, vm_process_status=None):
+def process_images(image_func, test, params, vm_process_status=None, vm_name=None):
     """
     Wrapper which chooses the best way to process images.
 
@@ -780,11 +841,20 @@ def process_images(image_func, test, params, vm_process_status=None):
     images = params.objects("images")
     if len(images) > 20:  # Lets do it in parallel
         _process_images_parallel(
-            image_func, test, params, vm_process_status=vm_process_status
+            image_func,
+            test,
+            params,
+            vm_process_status=vm_process_status,
+            vm_name=vm_name,
         )
     else:
         _process_images_serial(
-            image_func, test, images, params, vm_process_status=vm_process_status
+            image_func,
+            test,
+            images,
+            params,
+            vm_process_status=vm_process_status,
+            vm_name=vm_name,
         )
 
 
@@ -804,7 +874,13 @@ def process_fs_sources(fs_source_func, test, params, vm_process_status=None):
 
 
 def _process_images_serial(
-    image_func, test, images, params, exit_event=None, vm_process_status=None
+    image_func,
+    test,
+    images,
+    params,
+    exit_event=None,
+    vm_process_status=None,
+    vm_name=None,
 ):
     """
     Original process_image function, which allows custom set of images
@@ -817,14 +893,17 @@ def _process_images_serial(
                               or None for no vm exist.
     """
     for image_name in images:
-        image_params = params.object_params(image_name)
-        image_func(test, image_params, image_name, vm_process_status)
+        # image_params = params.object_params(image_name)
+        # image_func(test, image_params, image_name, vm_process_status)
+        image_func(test, params, image_name, vm_process_status, vm_name)
         if exit_event and exit_event.is_set():
             LOG.error("Received exit_event, stop processing of images.")
             break
 
 
-def _process_images_parallel(image_func, test, params, vm_process_status=None):
+def _process_images_parallel(
+    image_func, test, params, vm_process_status=None, vm_name=None
+):
     """
     The same as _process_images but in parallel.
     :param image_func: Process function
@@ -840,7 +919,9 @@ def _process_images_parallel(image_func, test, params, vm_process_status=None):
     for i in xrange(no_threads):
         imgs = images[i::no_threads]
         threads.append(
-            _CreateImages(image_func, test, imgs, params, exit_event, vm_process_status)
+            _CreateImages(
+                image_func, test, imgs, params, exit_event, vm_process_status, vm_name
+            )
         )
         threads[-1].start()
 
@@ -895,7 +976,9 @@ def _call_image_func():
                     unpause_vm = True
                     vm_params["skip_cluster_leak_warn"] = "yes"
                 try:
-                    process_images(image_func, test, vm_params, vm_process_status)
+                    process_images(
+                        image_func, test, vm_params, vm_process_status, vm_name
+                    )
                 finally:
                     if unpause_vm:
                         vm.resume()

virttest/staging/service.py

@@ -604,7 +604,7 @@ def __init__(
     #     :param runlevel: sys_v runlevel to set as default in inittab
     #     :type runlevel: str
     #     """
-    #     raise NotImplemented
+    #     raise NotImplementedError
 
 
 def convert_sysv_runlevel(level):

virttest/test_setup/storage.py

@@ -26,7 +26,9 @@ def setup(self):
             self.params["image_raw_device"] = "yes"
             self.env.register_lvmdev("lvm_%s" % self.params["main_vm"], lvmdev)
 
-        if self.params.get("storage_type") == "nfs":
+        if self.params.get("storage_type") == "nfs" and self.params.get_boolean(
+            "setup_local_nfs"
+        ):
             selinux_local = self.params.get("set_sebool_local", "yes") == "yes"
             selinux_remote = self.params.get("set_sebool_remote", "no") == "yes"
             image_nfs = Nfs(self.params)
@@ -89,7 +91,9 @@ def cleanup(self):
             finally:
                 self.env.unregister_lvmdev("lvm_%s" % self.params["main_vm"])
 
-        if self.params.get("storage_type") == "nfs":
+        if self.params.get("storage_type") == "nfs" and self.params.get_boolean(
+            "setup_local_nfs"
+        ):
             migration_setup = self.params.get("migration_setup", "no") == "yes"
             image_nfs = Nfs(self.params)
             image_nfs.cleanup()

virttest/vt_agent/managers/__init__.py

@@ -1,5 +1,9 @@
 from .connect import ConnectManager
 from .console import ConsoleManager
+from .image_manager import ImageHandlerManager
+from .resource_backing_manager import ResourceBackingManager
 
 connect_mgr = ConnectManager()
 console_mgr = ConsoleManager()
+resbacking_mgr = ResourceBackingManager()
+image_handler_mgr = ImageHandlerManager()