Merge branch 'ssh_export_import' into 'master'

Export and import disks using SSH

See merge request !424

circle/dashboard/store_api.py

@@ -47,7 +47,7 @@ class NoStoreException(StoreApiException):
 
 
 class Store(object):
 class Store(object):
 
 
-    def __init__(self, user, default_timeout=0.5):
+    def __init__(self, user, default_timeout=5):
         self.store_url = settings.STORE_URL
         self.store_url = settings.STORE_URL
         if not self.store_url:
         if not self.store_url:
             raise NoStoreException
             raise NoStoreException
@@ -127,6 +127,21 @@ class Store(object):
         r = self._request_cmd("UPLOAD", PATH=path)
         r = self._request_cmd("UPLOAD", PATH=path)
         return r.json()['LINK']
         return r.json()['LINK']
 
 
+    def request_ssh_download(self, path):
+        r = self._request_cmd("SSH_DOWNLOAD", PATH=path)
+        return r.json()['LINK'], r.json()['PORT']
+
+    def request_ssh_upload(self):
+        r = self._request_cmd("SSH_UPLOAD")
+        return r.json()['LINK'], r.json()['PORT']
+
+    def ssh_upload_finished(self, uploaded_name, path):
+        self._request_cmd(
+            "SSH_UPLOAD_FINISHED",
+            FILENAME=uploaded_name,
+            PATH=path,
+        )
+
     def remove(self, path):
     def remove(self, path):
         self._request_cmd("REMOVE", PATH=path)
         self._request_cmd("REMOVE", PATH=path)
 
 

circle/storage/models.py

@@ -477,7 +477,7 @@ class Disk(TimeStampedModel):
         return disk
         return disk
 
 
     @classmethod
     @classmethod
-    def import_disk(cls, user, name, download_link, task):
+    def import_disk(cls, user, name, download_link, port, task):
         params = {'name': name,
         params = {'name': name,
                   'type': 'qcow2-norm'}
                   'type': 'qcow2-norm'}
         disk = cls.__create(user=user, params=params)
         disk = cls.__create(user=user, params=params)
@@ -486,7 +486,7 @@ class Disk(TimeStampedModel):
             kwargs={
             kwargs={
                 "disk_desc": disk.get_disk_desc(),
                 "disk_desc": disk.get_disk_desc(),
                 "url": download_link,
                 "url": download_link,
-                "task": task.request.id
+                "port": port
             },
             },
             queue=queue_name
             queue=queue_name
         )
         )
@@ -497,18 +497,17 @@ class Disk(TimeStampedModel):
         disk.save()
         disk.save()
         return disk
         return disk
 
 
-    def export(self, exported_name, disk_format, upload_link, task):
+    def export(self, disk_format, upload_link, port, task):
         queue_name = self.get_remote_queue_name('storage', priority='slow')
         queue_name = self.get_remote_queue_name('storage', priority='slow')
         remote = storage_tasks.export_disk.apply_async(
         remote = storage_tasks.export_disk.apply_async(
             kwargs={
             kwargs={
                 "disk_desc": self.get_disk_desc(),
                 "disk_desc": self.get_disk_desc(),
                 "disk_format": disk_format,
                 "disk_format": disk_format,
-                "exported_name": exported_name,
                 "upload_link": upload_link,
                 "upload_link": upload_link,
-                "task": task.request.id
+                "port": port
             },
             },
             queue=queue_name)
             queue=queue_name)
-        self._run_abortable_task(remote, task)
+        return 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

@@ -39,12 +39,12 @@ def download(disk_desc, url):
 
 
 
 
 @celery.task(name='storagedriver.import_disk')
 @celery.task(name='storagedriver.import_disk')
-def import_disk(disk_desc, url):
+def import_disk(disk_desc, url, port):
     pass
     pass
 
 
 
 
 @celery.task(name='storagedriver.export_disk')
 @celery.task(name='storagedriver.export_disk')
-def export_disk(disk_desc, format):
+def export_disk(disk_desc, disk_format, url, port):
     pass
     pass
 
 
 
 

circle/vm/operations.py

@@ -352,7 +352,6 @@ class ImportDiskOperation(InstanceOperation):
                     '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
     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'
@@ -366,8 +365,8 @@ class ImportDiskOperation(InstanceOperation):
 
 
     def _operation(self, user, name, disk_path, task):
     def _operation(self, user, name, disk_path, task):
         store = Store(user)
         store = Store(user)
-        download_link = store.request_download(disk_path)
-        disk = Disk.import_disk(user, name, download_link, task)
+        download_link, port = store.request_ssh_download(disk_path)
+        disk = Disk.import_disk(user, name, download_link, port, task)
         self.instance.disks.add(disk)
         self.instance.disks.add(disk)
 
 
 
 
@@ -377,7 +376,6 @@ class ExportDiskOperation(InstanceOperation):
     name = _('export disk')
     name = _('export disk')
     description = _('Export disk to the selected format.')
     description = _('Export disk to the selected format.')
     abortable = True
     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'
@@ -391,8 +389,9 @@ class ExportDiskOperation(InstanceOperation):
 
 
     def _operation(self, user, disk, exported_name, disk_format, task):
     def _operation(self, user, disk, exported_name, disk_format, task):
         store = Store(user)
         store = Store(user)
-        upload_link = store.request_upload('/')
-        disk.export(exported_name, disk_format, upload_link, task)
+        upload_link, port = store.request_ssh_upload()
+        file_name = disk.export(disk_format, upload_link, port, task)
+        store.ssh_upload_finished(file_name, exported_name + '.' + disk_format)
 
 
 
 
 @register_operation
 @register_operation