Merge branch 'export_progress' into 'master'

Export and import disk progress

See merge request !419

circle/dashboard/forms.py

@@ -824,7 +824,7 @@ class VmCreateDiskForm(OperationForm):
 
 
 
 
 class VmDiskExportForm(OperationForm):
 class VmDiskExportForm(OperationForm):
-    format = forms.ChoiceField(
+    disk_format = forms.ChoiceField(
         choices=Disk.EXPORT_FORMATS,
         choices=Disk.EXPORT_FORMATS,
         label=_('Format'))
         label=_('Format'))
 
 
@@ -850,7 +850,7 @@ class VmDiskExportForm(OperationForm):
                     HTML(_("<label>Disk:</label> %s") % escape(self.disk)),
                     HTML(_("<label>Disk:</label> %s") % escape(self.disk)),
                     css_class="form-group",
                     css_class="form-group",
                 ),
                 ),
-                Field('disk'), Field('format')
+                Field('disk'), Field('disk_format')
             )
             )
         return helper
         return helper
 
 

circle/storage/models.py

@@ -123,8 +123,8 @@ class Disk(TimeStampedModel):
     TYPES = [('qcow2-norm', 'qcow2 normal'), ('qcow2-snap', 'qcow2 snapshot'),
     TYPES = [('qcow2-norm', 'qcow2 normal'), ('qcow2-snap', 'qcow2 snapshot'),
              ('iso', 'iso'), ('raw-ro', 'raw read-only'), ('raw-rw', 'raw')]
              ('iso', 'iso'), ('raw-ro', 'raw read-only'), ('raw-rw', 'raw')]
     BUS_TYPES = (('virtio', 'virtio'), ('ide', 'ide'), ('scsi', 'scsi'))
     BUS_TYPES = (('virtio', 'virtio'), ('ide', 'ide'), ('scsi', 'scsi'))
-    EXPORT_FORMATS = (('vmdk', _('VMware disk image')),
-                      ('qcow2', _('QEMU disk image')),
+    EXPORT_FORMATS = (('qcow2', _('QEMU disk image')),
+                      ('vmdk', _('VMware disk image')),
                       ('vdi', _('VirtualBox disk image')),
                       ('vdi', _('VirtualBox disk image')),
                       ('vpc', _('HyperV disk image')))
                       ('vpc', _('HyperV disk image')))
     name = CharField(blank=True, max_length=100, verbose_name=_("name"))
     name = CharField(blank=True, max_length=100, verbose_name=_("name"))
@@ -431,6 +431,19 @@ class Disk(TimeStampedModel):
         return disk
         return disk
 
 
     @classmethod
     @classmethod
+    def _run_abortable_task(cls, remote, task):
+        while True:
+            try:
+                result = remote.get(timeout=5)
+                break
+            except TimeoutError as e:
+                if task is not None and task.is_aborted():
+                    AbortableAsyncResult(remote.id).abort()
+                    raise humanize_exception(ugettext_noop(
+                        "Operation aborted by user."), e)
+        return result
+
+    @classmethod
     def download(cls, url, task, user=None, **params):
     def download(cls, url, task, user=None, **params):
         """Create disk object and download data from url synchronusly.
         """Create disk object and download data from url synchronusly.
 
 
@@ -455,15 +468,7 @@ class Disk(TimeStampedModel):
             kwargs={'url': url, 'parent_id': task.request.id,
             kwargs={'url': url, 'parent_id': task.request.id,
                     'disk': disk.get_disk_desc()},
                     'disk': disk.get_disk_desc()},
             queue=queue_name)
             queue=queue_name)
-        while True:
-            try:
-                result = remote.get(timeout=5)
-                break
-            except TimeoutError as e:
-                if task is not None and task.is_aborted():
-                    AbortableAsyncResult(remote.id).abort()
-                    raise humanize_exception(ugettext_noop(
-                        "Operation aborted by user."), e)
+        result = cls._run_abortable_task(remote, task)
         disk.size = result['size']
         disk.size = result['size']
         disk.type = result['type']
         disk.type = result['type']
         disk.checksum = result.get('checksum', None)
         disk.checksum = result.get('checksum', None)
@@ -472,28 +477,40 @@ class Disk(TimeStampedModel):
         return disk
         return disk
 
 
     @classmethod
     @classmethod
-    def import_disk(cls, user, name, download_link, timeout=3600):
+    def import_disk(cls, user, name, download_link, task):
         params = {'name': name,
         params = {'name': name,
                   'type': 'qcow2-norm'}
                   'type': 'qcow2-norm'}
-        disk = cls.create(user, **params)
+        disk = cls.__create(user=user, params=params)
         queue_name = disk.get_remote_queue_name('storage', priority='slow')
         queue_name = disk.get_remote_queue_name('storage', priority='slow')
         remote = storage_tasks.import_disk.apply_async(
         remote = storage_tasks.import_disk.apply_async(
-            args=[disk.get_disk_desc(), download_link],
+            kwargs={
+                "disk_desc": disk.get_disk_desc(),
+                "url": download_link,
+                "task": task.request.id
+            },
             queue=queue_name
             queue=queue_name
         )
         )
-        disk_size = remote.get(timeout=timeout)
-        disk.size = disk_size
+        result = cls._run_abortable_task(remote, task)
+        disk.size = result["size"]
+        disk.checksum = result["checksum"]
         disk.is_ready = True
         disk.is_ready = True
         disk.save()
         disk.save()
         return disk
         return disk
 
 
-    def export(self, format, upload_link, timeout=3600):
+    def export(self, disk_format, upload_link, task):
         exported_name = self.name if self.name != '' else self.filename
         exported_name = self.name if self.name != '' else self.filename
 
 
         queue_name = self.get_remote_queue_name('storage', priority='slow')
         queue_name = self.get_remote_queue_name('storage', priority='slow')
-        storage_tasks.export.apply_async(
-            args=[self.get_disk_desc(), format, exported_name, upload_link],
-            queue=queue_name).get(timeout=timeout)
+        remote = storage_tasks.export_disk.apply_async(
+            kwargs={
+                "disk_desc": self.get_disk_desc(),
+                "disk_format": disk_format,
+                "exported_name": exported_name,
+                "upload_link": upload_link,
+                "task": task.request.id
+            },
+            queue=queue_name)
+        self._run_abortable_task(remote, task)
 
 
     def destroy(self, user=None, task_uuid=None):
     def destroy(self, user=None, task_uuid=None):
         if self.destroyed:
         if self.destroyed:

circle/storage/tasks/storage_tasks.py

@@ -43,8 +43,8 @@ def import_disk(disk_desc, url):
     pass
     pass
 
 
 
 
-@celery.task(name='storagedriver.export')
-def export(disk_desc, format):
+@celery.task(name='storagedriver.export_disk')
+def export_disk(disk_desc, format):
     pass
     pass
 
 
 
 

circle/vm/operations.py

@@ -351,6 +351,8 @@ class ImportDiskOperation(InstanceOperation):
     description = _('Import and attach a disk image to the virtual machine '
     description = _('Import and attach a disk image to the virtual machine '
                     'from the user store. The disk image has to be in the '
                     'from the user store. The disk image has to be in the '
                     'root directory of the store.')
                     'root directory of the store.')
+    abortable = True
+    has_percentage = True
     required_perms = ('storage.import_disk',)
     required_perms = ('storage.import_disk',)
     accept_states = ('STOPPED', 'PENDING', 'RUNNING')
     accept_states = ('STOPPED', 'PENDING', 'RUNNING')
     async_queue = 'localhost.man.slow'
     async_queue = 'localhost.man.slow'
@@ -362,10 +364,10 @@ class ImportDiskOperation(InstanceOperation):
         except NoStoreException:
         except NoStoreException:
             raise PermissionDenied
             raise PermissionDenied
 
 
-    def _operation(self, user, name, disk_path):
+    def _operation(self, user, name, disk_path, task):
         store = Store(user)
         store = Store(user)
         download_link = store.request_download(disk_path)
         download_link = store.request_download(disk_path)
-        disk = Disk.import_disk(user, name, download_link)
+        disk = Disk.import_disk(user, name, download_link, task)
         self.instance.disks.add(disk)
         self.instance.disks.add(disk)
 
 
 
 
@@ -374,6 +376,8 @@ class ExportDiskOperation(InstanceOperation):
     id = 'export_disk'
     id = 'export_disk'
     name = _('export disk')
     name = _('export disk')
     description = _('Export disk to the selected format.')
     description = _('Export disk to the selected format.')
+    abortable = True
+    has_percentage = True
     required_perms = ('storage.export_disk',)
     required_perms = ('storage.export_disk',)
     accept_states = ('STOPPED',)
     accept_states = ('STOPPED',)
     async_queue = 'localhost.man.slow'
     async_queue = 'localhost.man.slow'
@@ -385,10 +389,10 @@ class ExportDiskOperation(InstanceOperation):
         except NoStoreException:
         except NoStoreException:
             raise PermissionDenied
             raise PermissionDenied
 
 
-    def _operation(self, user, disk, format):
+    def _operation(self, user, disk, disk_format, task):
         store = Store(user)
         store = Store(user)
         upload_link = store.request_upload('/')
         upload_link = store.request_upload('/')
-        disk.export(format, upload_link)
+        disk.export(disk_format, upload_link, task)
 
 
 
 
 @register_operation
 @register_operation