Added RPC handler on hypervisor for volume export/import.

ccnode/hypervisor/__init__.py

@@ -13,6 +13,7 @@ from ccnode.hypervisor.lib import (
 )
 from ccnode.hypervisor.domains import VirtualMachine
 from ccnode.exc import UndefinedDomain, PoolStorageError
+from ccnode.hypervisor.jobs import ImportVolume, ExportVolume
 
 
 logger = logging.getLogger(__name__)
@@ -104,6 +105,11 @@ class Handler(HostHandler):
         self.main.reset_handler('vm_start', self.vm_start)
         self.main.reset_handler('vm_suspend', self.vm_suspend)
         self.main.reset_handler('vm_resume', self.vm_resume)
+        self.main.reset_handler('vol_create', self.vol_create)
+        self.main.reset_handler('vol_delete', self.vol_delete)
+        self.main.reset_handler('vol_import', self.vol_import)
+        self.main.reset_handler('vol_import_wait', self.vol_import_wait)
+        self.main.reset_handler('vol_export', self.vol_export)
 
         # if everything went fine, unregister the timer
         self.timer.stop()
@@ -145,6 +151,11 @@ class Handler(HostHandler):
         self.main.remove_handler('vm_start')
         self.main.remove_handler('vm_suspend')
         self.main.remove_handler('vm_resume')
+        self.main.remove_handler('vol_create')
+        self.main.remove_handler('vol_delete')
+        self.main.remove_handler('vol_import')
+        self.main.remove_handler('vol_import_wait')
+        self.main.remove_handler('vol_export')
         # launch connection timer
         self.timer.start()
 
@@ -263,6 +274,76 @@ class Handler(HostHandler):
             logger.exception('Error while deleting volume')
             raise
 
+    def vol_import(self, pool, name):
+        """
+        :param pool: pool name where the volume is
+        :param name: name of the volume
+        """
+        logger.debug('Volume import pool = %s, volume = %s', pool, name)
+        try:
+            pool = self.hypervisor.storage.get_storage(pool)
+            if pool is None:
+                raise Exception('Pool storage does not exist')  # TODO exc
+
+            volume = pool.volumes.get(name)
+            if volume is None:
+                raise Exception('Volume does not exist')
+
+            # create the job
+            job = self.main.job_manager.create(ImportVolume, volume)
+            job.start()
+        except Exception:
+            logger.exception('Error while starting import job')
+            raise
+
+        return dict(id=job.id, port=job.port)
+
+    def vol_import_wait(self, job_id):
+        """Block until completion of the given job id."""
+        job = self.main.job_manager.get(job_id)
+        logger.debug('Waiting for import job to terminate')
+        job.join()
+        logger.debug('Import job terminated')
+
+        return dict(id=job.id, log='', checksum=job.checksum)
+
+    def vol_import_cancel(self, job_id):
+        """Cancel import job."""
+        logger.debug('Cancel import job')
+        self.main.job_manager.cancel(job_id)
+
+    def vol_export(self, pool, name, raddr, rport):
+        """
+        :param pool: pool name where the volume is
+        :param name: name of the volume
+        :param raddr: IP address of the destination to send the volume to
+        :param rport: TCP port of the destination
+        """
+        pool = self.hypervisor.storage.get_storage(pool)
+
+        if pool is None:
+            raise Exception('Pool storage does not exist')
+
+        volume = pool.volumes.get(name)
+
+        if volume is None:
+            raise Exception('Volume does not exist')
+
+        try:
+            job = self.main.job_manager.create(ExportVolume, volume, raddr, rport)
+            # we don't run the job in a background thread thus exceptions are
+            # sent to the sjRPC
+            job.start_current()
+            # otherwise we would do
+            # job.start()
+            # job.join()
+        except Exception:
+            logger.exception('Error while exporting volume')
+            raise
+
+        logger.debug('Export volume successfull')
+        return dict(id=job.id, log='', checksum=job.checksum)
+
 
 class Hypervisor(object):
     """Container for all hypervisor related state."""