Merge pull request #615 from lago-project/fixing_export_issues

export: Some fixes

lago/cmd.py

@@ -344,15 +344,28 @@ def do_shutdown(prefix, vm_names, reboot, **kwargs):
     default='LagoInitFile',
     help='The name of the exported init file',
 )
+@lago.plugins.cli.cli_plugin_add_argument(
+    '--collect-only',
+    action='store_true',
+    help='Only output the disks that will be exported',
+)
+@lago.plugins.cli.cli_plugin_add_argument(
+    '--without-threads',
+    action='store_true',
+    help='If set, do not use threads',
+)
 @in_lago_prefix
 @with_logging
 def do_export(
     prefix, vm_names, standalone, dst_dir, compress, init_file_name,
-    out_format, **kwargs
+    out_format, collect_only, without_threads, **kwargs
 ):
-    prefix.export_vms(
-        vm_names, standalone, dst_dir, compress, init_file_name, out_format
+    output = prefix.export_vms(
+        vm_names, standalone, dst_dir, compress, init_file_name, out_format,
+        collect_only, not without_threads
     )
+    if collect_only:
+        print(out_format.format(output))
 
 
 @lago.plugins.cli.cli_plugin(

lago/export.py

@@ -16,11 +16,11 @@
 
 class DiskExportManager(object):
     """
-    ExportManager object is responsible on the export process of an image
-    from the current Lago prefix.
+    DiskExportManager object is responsible on the export process of
+    an image from the current Lago prefix.
 
-    DiskExportManger is the base class of specific ExportManagers.
-    Each specific ExportManger is responsible on the export process of an
+    DiskExportManger is the base class of specific DiskExportManger.
+    Each specific DiskExportManger is responsible on the export process of an
     image with a specific type (e.g template, file...)
 
     Attributes:
@@ -32,7 +32,8 @@ class DiskExportManager(object):
             as found in workdir/current/virt/VM-NAME
         exported_metadata(dict): A copy of the source disk metadata, this
             dict should be updated with new values during the export process.
-            do_compress(bool): If true, apply compression to the exported disk.
+        do_compress(bool): If true, apply compression to the exported
+            disk.
     """
 
     __metaclass__ = ABCMeta
@@ -65,7 +66,8 @@ def get_instance_by_type(dst, disk, do_compress, *args, **kwargs):
             disk (dict): Disk attributes
                 (of the disk that should be exported) as
                 found in workdir/current/virt/VM-NAME
-            do_compress(bool): If true, apply compression to the exported disk.
+            do_compress(bool): If true, apply compression to the exported
+                disk.
         Returns
             An instance of a subclass of DiskExportManager which
             matches the disk type.
@@ -185,9 +187,18 @@ def rebase(self):
                 # this identifier will be used later by Lago in order
                 # to resolve and download the base image
                 parent = self.src_qemu_info[0]['backing-filename']
-                parent = './{}'.format(
-                    os.path.basename(parent.split(':', 1)[1])
-                )
+
+                # Hack for working with lago images naming convention
+                # For example: /var/lib/lago/store/phx_repo:el7.3-base:v1
+                # Extract only the image name and the version
+                # (in the example el7.3-base:v1)
+                parent = os.path.basename(parent)
+                try:
+                    parent = parent.split(':', 1)[1]
+                except IndexError:
+                    pass
+
+                parent = './{}'.format(parent)
                 utils.qemu_rebase(
                     target=self.dst, backing_file=parent, safe=False
                 )
@@ -243,14 +254,98 @@ def export(self):
                     shutil.rmtree, self.dst, ignore_errors=True
                 )
                 self.copy()
-                self.sparse()
+                if not self.disk['format'] == 'iso':
+                    self.sparse()
                 self.calc_sha('sha1')
                 self.update_lago_metadata()
                 self.write_lago_metadata()
                 self.compress()
                 rollback.clear()
 
 
+class VMExportManager(object):
+    """
+    VMExportManager object is responsible on the export process of a list of
+    disks.
+
+    Attributes:
+        disks (list of dicts): Disks to export.
+        dst (str): Where to place the exported disks.
+        compress(bool): If True compress each exported disk.
+        with_threads(bool): If True, run the export in parallel
+        *args(list): Extra args, will be passed to each
+            DiskExportManager
+        **kwargs(dict): Extra args, will be passed to each
+            DiskExportManager
+
+    """
+
+    def __init__(
+        self, disks, dst, compress, with_threads=True, *args, **kwargs
+    ):
+        self._disks = disks
+        self._dst = os.path.expandvars(os.path.realpath(dst))
+        self._compress = compress
+        self._with_threads = with_threads
+        self._args = args
+        self._kwargs = kwargs
+
+    def _collect(self):
+        """
+        Returns:
+            (generator of dicts): The disks that needed to be exported
+        """
+        return (disk for disk in self._disks if not disk.get('skip-export'))
+
+    def collect_paths(self):
+        """
+        Returns:
+            (list of str): The path of the disks that will be exported.
+        """
+        return [os.path.expandvars(disk['path']) for disk in self._collect()]
+
+    def exported_disks_paths(self):
+        """
+        Returns:
+            (list of str): The path of the exported disks.
+        """
+        return [
+            os.path.join(self._dst, os.path.basename(disk['path']))
+            for disk in self._collect()
+        ]
+
+    def _get_export_mgr(self):
+        """
+        Returns:
+            (DiskExportManager): Handler for each disk
+        """
+        return (
+            DiskExportManager.get_instance_by_type(
+                dst=self._dst,
+                disk=disk,
+                do_compress=self._compress,
+                *self._args,
+                **self._kwargs
+            ) for disk in self._collect()
+        )
+
+    def export(self):
+        """
+        Run the export process
+        Returns:
+            (list of str): The path of the exported disks.
+        """
+        if self._with_threads:
+            utils.invoke_different_funcs_in_parallel(
+                *map(lambda mgr: mgr.export, self._get_export_mgr())
+            )
+        else:
+            for mgr in self._get_export_mgr():
+                mgr.export()
+
+        return self.exported_disks_paths()
+
+
 HANDLERS = {
     'file': FileExportManager,
     'empty': FileExportManager,

lago/plugins/vm.py

@@ -387,13 +387,21 @@ def extract_paths_dead(self, paths, *args, **kwargs):
         return self.provider.extract_paths_dead(paths, *args, **kwargs)
 
     def export_disks(
-        self, standalone=True, dst_dir=None, compress=False, *args, **kwargs
+        self,
+        standalone=True,
+        dst_dir=None,
+        compress=False,
+        collect_only=False,
+        with_threads=True,
+        *args,
+        **kwargs
     ):
         """
         Thin method that just uses the provider
         """
         return self.provider.export_disks(
-            standalone, dst_dir, compress, *args, **kwargs
+            standalone, dst_dir, compress, collect_only, with_threads, *args,
+            **kwargs
         )
 
     def copy_to(self, local_path, remote_path, recursive=True):

lago/prefix.py

@@ -31,6 +31,7 @@
 import warnings
 import pkg_resources
 from os.path import join
+from plugins.output import YAMLOutFormatPlugin
 
 import xmltodict
 
@@ -1220,13 +1221,47 @@ def build(self, conf):
 
         utils.invoke_in_parallel(build.Build.build, builders)
 
+    @sdk_utils.expose
     def export_vms(
-        self, vms_names, standalone, export_dir, compress, init_file_name,
-        out_format
+        self,
+        vms_names=None,
+        standalone=False,
+        export_dir='.',
+        compress=False,
+        init_file_name='LagoInitFile',
+        out_format=YAMLOutFormatPlugin(),
+        collect_only=False,
+        with_threads=True,
     ):
-        self.virt_env.export_vms(
+        """
+        Export vm images disks and init file.
+        The exported images and init file can be used to recreate
+        the environment.
+
+        Args:
+            vms_names(list of str): Names of the vms to export, if None
+                export all the vms in the env (default=None)
+            standalone(bool): If false, export a layered image
+                (default=False)
+            export_dir(str): Dir to place the exported images and init file
+            compress(bool): If True compress the images with xz
+                  (default=False)
+            init_file_name(str): The name of the exported init file
+                (default='LagoInitfile')
+            out_format(:class:`lago.plugins.output.OutFormatPlugin`):
+                The type of the exported init file (the default is yaml)
+            collect_only(bool): If True, return only a mapping from vm name
+                to the disks that will be exported. (default=False)
+            with_threads(bool): If True, run the export in parallel
+                (default=True)
+
+        Returns
+            Unless collect_only == True, a mapping between vms' disks.
+
+        """
+        return self.virt_env.export_vms(
             vms_names, standalone, export_dir, compress, init_file_name,
-            out_format
+            out_format, collect_only, with_threads
         )
 
     @sdk_utils.expose

lago/providers/libvirt/vm.py

@@ -344,36 +344,48 @@ def extract_paths_dead(self, paths, ignore_nopath):
             ignore_nopath=ignore_nopath
         )
 
-    def export_disks(self, standalone, dst_dir, compress, *args, **kwargs):
+    def export_disks(
+        self,
+        standalone,
+        dst_dir,
+        compress,
+        collect_only=False,
+        with_threads=True,
+        *args,
+        **kwargs
+    ):
         """
-        Exports all the disks of self.
-        For each disk type, handler function should be added.
+        Export all the disks of self.
 
         Args:
             standalone (bool): if true, merge the base images and the layered
-             image into a new file (Supported only in qcow2 format)
+                image into a new file (Supported only in qcow2 format)
             dst_dir (str): dir to place the exported disks
             compress(bool): if true, compress each disk.
+            collect_only(bool): If true, return only a dict which maps between
+                the name of the vm to the paths of the disks that will be
+                exported (don't export anything).
+            with_threads(bool): If True, export disks in parallel
 
+        Returns:
+            (dict): which maps between the name of the vm to the paths of
+            the disks that will be exported
         """
-        formats_to_exclude = {'iso'}
-
-        export_managers = [
-            export.DiskExportManager.get_instance_by_type(
-                dst=dst_dir,
-                disk=disk,
-                do_compress=compress,
-                standalone=standalone,
-                *args,
-                **kwargs
-            ) for disk in self.vm.disks
-            if disk.get('format') not in formats_to_exclude
-        ]
-
-        utils.invoke_different_funcs_in_parallel(
-            *map(lambda manager: manager.export, export_managers)
+        vm_export_mgr = export.VMExportManager(
+            disks=self.vm.disks,
+            dst=dst_dir,
+            compress=compress,
+            with_threads=with_threads,
+            standalone=standalone,
+            *args,
+            **kwargs
         )
 
+        if collect_only:
+            return {self.vm.name(): vm_export_mgr.collect_paths()}
+        else:
+            return {self.vm.name(): vm_export_mgr.export()}
+
     @vm_plugin.check_defined
     def interactive_console(self):
         """

lago/utils.py

@@ -97,13 +97,13 @@ def join_all(self, raise_exceptions=True):
 def invoke_in_parallel(func, *args_sequences):
     vt = VectorThread(func_vector(func, zip(*args_sequences)))
     vt.start_all()
-    vt.join_all()
+    return vt.join_all()
 
 
 def invoke_different_funcs_in_parallel(*funcs):
     vt = VectorThread(funcs)
     vt.start_all()
-    vt.join_all()
+    return vt.join_all()
 
 
 _CommandStatus = collections.namedtuple(