vm: move Instance.shutdown_vm to ShutdownOperation

circle/vm/models/instance.py

@@ -22,8 +22,6 @@ from importlib import import_module
 from logging import getLogger
 from warnings import warn
 
-from celery.exceptions import TimeoutError
-from celery.contrib.abortable import AbortableAsyncResult
 import django.conf
 from django.contrib.auth.models import User
 from django.core import signing
@@ -43,7 +41,7 @@ from taggit.managers import TaggableManager
 
 from acl.models import AclBase
 from common.models import (
-    create_readable, HumanReadableException, humanize_exception
+    create_readable, HumanReadableException,
 )
 from common.operations import OperatedMixin
 from ..tasks import vm_tasks, agent_tasks
@@ -769,22 +767,6 @@ class Instance(AclBase, VirtualMachineDescModel, StatusModel, OperatedMixin,
                                             queue=queue_name
                                             ).get(timeout=timeout)
 
-    def shutdown_vm(self, task=None, step=5):
-        queue_name = self.get_remote_queue_name('vm', 'slow')
-        logger.debug("RPC Shutdown at queue: %s, for vm: %s.", queue_name,
-                     self.vm_name)
-        remote = vm_tasks.shutdown.apply_async(kwargs={'name': self.vm_name},
-                                               queue=queue_name)
-
-        while True:
-            try:
-                return remote.get(timeout=step)
-            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)
-
     def delete_mem_dump(self, timeout=15):
         queue_name = self.mem_dump['datastore'].get_remote_queue_name(
             'storage', 'fast')

circle/vm/operations.py

@@ -46,6 +46,7 @@ from .models import (
 from .tasks import agent_tasks, local_agent_tasks, vm_tasks
 
 from dashboard.store_api import Store, NoStoreException
+from storage.tasks import storage_tasks
 
 logger = getLogger(__name__)
 
@@ -65,6 +66,24 @@ class RemoteOperationMixin(object):
         self._get_remote_queue()
 
 
+class AbortableRemoteOperationMixin(object):
+    remote_step = property(lambda self: self.remote_timeout / 10)
+
+    def _operation(self, task, **kwargs):
+        args = self._get_remote_args(**kwargs),
+        remote = self.task.apply_async(
+            args=args, queue=self._get_remote_queue())
+        for i in xrange(int(self.remote_timeout / self.remote_step)):
+            try:
+                return remote.get(timeout=self.remote_step)
+            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 remote.get(timeout=self.remote_step)
+
+
 class InstanceOperation(Operation):
     acl_level = 'owner'
     async_operation = abortable_async_instance_operation
@@ -656,7 +675,8 @@ class SaveAsTemplateOperation(InstanceOperation):
 
 
 @register_operation
-class ShutdownOperation(InstanceOperation):
+class ShutdownOperation(AbortableRemoteOperationMixin,
+                        RemoteInstanceOperation):
     id = 'shutdown'
     name = _("shutdown")
     description = _("Try to halt virtual machine by a standard ACPI signal, "
@@ -667,9 +687,12 @@ class ShutdownOperation(InstanceOperation):
     required_perms = ()
     accept_states = ('RUNNING', )
     resultant_state = 'STOPPED'
+    task = vm_tasks.shutdown
+    remote_queue = ("vm", "slow")
+    timeout = 120
 
-    def _operation(self, task=None):
-        self.instance.shutdown_vm(task=task)
+    def _operation(self, task):
+        super(ShutdownOperation, self)._operation(task=task)
         self.instance.yield_node()
 
     def on_abort(self, activity, error):