Add progress and handle abort for export disk

disk.py

@@ -11,7 +11,7 @@ from time import sleep
 from hashlib import md5
 import re
 
-from requests_toolbelt import MultipartEncoder
+from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
 import requests
 
 logger = logging.getLogger(__name__)
@@ -312,22 +312,34 @@ class Disk(object):
 
             self.size = Disk.get(self.dir, self.name).size
 
-    def export(self, format, exported_name, upload_link):
-        exported_path = self.get_path() + '.' + format
+    def export(self, task, disk_format, exported_name, upload_link, parent_id):
+        exported_path = self.get_path() + '.' + disk_format
         cmdline = ['qemu-img',
                    'convert',
-                   '-O', format,
+                   '-O', disk_format,
                    self.get_path(),
                    exported_path]
 
         subprocess.check_output(cmdline)
+        size = os.path.getsize(exported_path)
+
+        def update_state(monitor):
+            percent = monitor.bytes_read * 100 / size
+            if task.is_aborted():
+                raise AbortException()
+            task.update_state(
+                task_id=parent_id,
+                state=task.AsyncResult(parent_id).state,
+                meta={'percent': percent}
+            )
 
         with open(exported_path, 'rb') as exported_disk:
             try:
-                m = MultipartEncoder(
-                    {'data': (exported_name + '.' + format,
+                e = MultipartEncoder(
+                    {'data': (exported_name + '.' + disk_format,
                               exported_disk)}
                 )
+                m = MultipartEncoderMonitor(e, update_state)
                 response = requests.post(
                     upload_link,
                     data=m,

storagedriver.py

@@ -56,10 +56,17 @@ class import_disk(AbortableTask):
         }
 
 
-@celery.task()
-def export(disk_desc, format, exported_name, upload_link):
-    disk = Disk.deserialize(disk_desc)
-    disk.export(format, exported_name, upload_link)
+class export_disk(AbortableTask):
+    time_limit = 18000
+
+    def run(self, **kwargs):
+        disk_desc = kwargs["disk_desc"]
+        disk_format = kwargs["disk_format"]
+        exported_name = kwargs["exported_name"]
+        upload_link = kwargs["upload_link"]
+        parent_id = kwargs["task"]
+        disk = Disk.deserialize(disk_desc)
+        disk.export(self, disk_format, exported_name, upload_link, parent_id)
 
 
 @celery.task()