Merge branch 'issue-69'

Conflicts:
	circle/dashboard/views.py

circle/common/models.py

 from collections import deque
+from contextlib import contextmanager
 from hashlib import sha224
 from logging import getLogger
 from time import time
@@ -32,6 +33,18 @@ def activitycontextimpl(act, on_abort=None, on_commit=None):
         act.finish(succeeded=True, event_handler=on_commit)
 
 
+activity_context = contextmanager(activitycontextimpl)
+
+
+activity_code_separator = '.'
+
+
+def join_activity_code(*args):
+    """Join the specified parts into an activity code.
+    """
+    return activity_code_separator.join(args)
+
+
 class ActivityModel(TimeStampedModel):
     activity_code = CharField(max_length=100, verbose_name=_('activity code'))
     parent = ForeignKey('self', blank=True, null=True, related_name='children')

circle/common/operations.py

+from logging import getLogger
+
+from .models import activity_context
+
+from django.core.exceptions import PermissionDenied
+
+
+logger = getLogger(__name__)
+
+
+class Operation(object):
+    """Base class for VM operations.
+    """
+    async_queue = 'localhost.man'
+    required_perms = ()
+
+    def __call__(self, **kwargs):
+        return self.call(**kwargs)
+
+    def __init__(self, subject):
+        """Initialize a new operation bound to the specified subject.
+        """
+        self.subject = subject
+
+    def __unicode__(self):
+        return self.name
+
+    def __prelude(self, kwargs):
+        """This method contains the shared prelude of call and async.
+        """
+        skip_checks = kwargs.setdefault('system', False)
+        user = kwargs.setdefault('user', None)
+        parent_activity = kwargs.pop('parent_activity', None)
+
+        if not skip_checks:
+            self.check_auth(user)
+        self.check_precond()
+        return self.create_activity(parent=parent_activity, user=user)
+
+    def _exec_op(self, activity, user, **kwargs):
+        """Execute the operation inside the specified activity's context.
+        """
+        with activity_context(activity, on_abort=self.on_abort,
+                              on_commit=self.on_commit):
+            return self._operation(activity, user, **kwargs)
+
+    def _operation(self, activity, user, system, **kwargs):
+        """This method is the operation's particular implementation.
+
+        Deriving classes should implement this method.
+        """
+        raise NotImplementedError
+
+    def async(self, **kwargs):
+        """Execute the operation asynchronously.
+
+        Only a quick, preliminary check is ran before creating the associated
+        activity and queuing the job.
+
+        The returned value is the handle for the asynchronous job.
+
+        For more information, check the synchronous call's documentation.
+        """
+        logger.info("%s called asynchronously with the following parameters: "
+                    "%r", self.__class__.__name__, kwargs)
+        activity = self.__prelude(kwargs)
+        return self.async_operation.apply_async(args=(self.id,
+                                                      self.subject.pk,
+                                                      activity.pk),
+                                                kwargs=kwargs,
+                                                queue=self.async_queue)
+
+    def call(self, **kwargs):
+        """Execute the operation (synchronously).
+
+        Anticipated keyword arguments:
+        * parent_activity: Parent activity for the operation. If this argument
+                           is present, the operation's activity will be created
+                           as a child activity of it.
+        * system: Indicates that the operation is invoked by the system, not a
+                  User. If this argument is present and has a value of True,
+                  then authorization checks are skipped.
+        * user: The User invoking the operation. If this argument is not
+                present, it'll be provided with a default value of None.
+        """
+        logger.info("%s called (synchronously) with the following parameters: "
+                    "%r", self.__class__.__name__, kwargs)
+        activity = self.__prelude(kwargs)
+        return self._exec_op(activity=activity, **kwargs)
+
+    def check_precond(self):
+        pass
+
+    def check_auth(self, user):
+        if not user.has_perms(self.required_perms):
+            raise PermissionDenied("%s doesn't have the required permissions."
+                                   % user)
+
+    def create_activity(self, parent, user):
+        raise NotImplementedError
+
+    def on_abort(self, activity, error):
+        """This method is called when the operation aborts (i.e. raises an
+        exception).
+        """
+        pass
+
+    def on_commit(self, activity):
+        """This method is called when the operation executes successfully.
+        """
+        pass
+
+
+operation_registry_name = '_ops'
+
+
+class OperatedMixin(object):
+    def __getattr__(self, name):
+        # NOTE: __getattr__ is only called if the attribute doesn't already
+        # exist in your __dict__
+        cls = self.__class__
+        ops = getattr(cls, operation_registry_name, {})
+        op = ops.get(name)
+        if op:
+            return op(self)
+        else:
+            raise AttributeError("%r object has no attribute %r" %
+                                 (self.__class__.__name__, name))
+
+
+def register_operation(target_cls, op_cls, op_id=None):
+    """Register the specified operation with the target class.
+
+    You can optionally specify an ID to be used for the registration;
+    otherwise, the operation class' 'id' attribute will be used.
+    """
+    if op_id is None:
+        op_id = op_cls.id
+
+    if not issubclass(target_cls, OperatedMixin):
+        raise TypeError("%r is not a subclass of %r" %
+                        (target_cls.__name__, OperatedMixin.__name__))
+
+    if not hasattr(target_cls, operation_registry_name):
+        setattr(target_cls, operation_registry_name, dict())
+
+    getattr(target_cls, operation_registry_name)[op_id] = op_cls

circle/common/tests/test_operations.py

+from mock import MagicMock, patch
+
+from django.test import TestCase
+
+from ..operations import Operation
+
+
+class OperationTestCase(TestCase):
+    def test_activity_created_before_async_job(self):
+        class AbortEx(Exception):
+            pass
+
+        op = Operation(MagicMock())
+        op.activity_code_suffix = 'test'
+        op.id = 'test'
+        op.async_operation = MagicMock(
+            apply_async=MagicMock(side_effect=AbortEx))
+
+        with patch.object(Operation, 'check_precond'):
+            with patch.object(Operation, 'create_activity') as create_act:
+                try:
+                    op.async(system=True)
+                except AbortEx:
+                    self.assertTrue(create_act.called)
+
+    def test_check_precond_called_before_create_activity(self):
+        class AbortEx(Exception):
+            pass
+
+        op = Operation(MagicMock())
+        op.activity_code_suffix = 'test'
+        op.id = 'test'
+        with patch.object(Operation, 'create_activity', side_effect=AbortEx):
+            with patch.object(Operation, 'check_precond') as chk_pre:
+                try:
+                    op.call(system=True)
+                except AbortEx:
+                    self.assertTrue(chk_pre.called)
+
+    def test_auth_check_on_non_system_call(self):
+        op = Operation(MagicMock())
+        op.activity_code_suffix = 'test'
+        op.id = 'test'
+        user = MagicMock()
+        with patch.object(Operation, 'check_auth') as check_auth:
+            with patch.object(Operation, 'check_precond'), \
+                    patch.object(Operation, 'create_activity'), \
+                    patch.object(Operation, '_exec_op'):
+                op.call(user=user)
+            check_auth.assert_called_with(user)
+
+    def test_no_auth_check_on_system_call(self):
+        op = Operation(MagicMock())
+        op.activity_code_suffix = 'test'
+        op.id = 'test'
+        with patch.object(Operation, 'check_auth', side_effect=AssertionError):
+            with patch.object(Operation, 'check_precond'), \
+                    patch.object(Operation, 'create_activity'), \
+                    patch.object(Operation, '_exec_op'):
+                op.call(system=True)

circle/dashboard/tests/test_views.py

@@ -6,6 +6,7 @@ from django.core.urlresolvers import reverse
 from django.contrib.auth.models import Permission
 
 from vm.models import Instance, InstanceTemplate, Lease, Node
+from vm.operations import WakeUpOperation
 from ..models import Profile
 from ..views import VmRenewView
 from storage.models import Disk
@@ -487,7 +488,7 @@ class VmDetailTest(LoginMixin, TestCase):
     def test_permitted_wake_up_wrong_state(self):
         c = Client()
         self.login(c, "user2")
-        with patch.object(Instance, 'wake_up_async') as mock_method:
+        with patch.object(WakeUpOperation, 'async') as mock_method:
             inst = Instance.objects.get(pk=1)
             mock_method.side_effect = inst.wake_up
             inst.manual_state_change('RUNNING')
@@ -501,7 +502,7 @@ class VmDetailTest(LoginMixin, TestCase):
         c = Client()
         self.login(c, "user2")
         with patch.object(Instance, 'select_node', return_value=None):
-            with patch.object(Instance, 'wake_up_async') as new_wake_up:
+            with patch.object(WakeUpOperation, 'async') as new_wake_up:
                 with patch('vm.tasks.vm_tasks.wake_up.apply_async') as wuaa:
                     inst = Instance.objects.get(pk=1)
                     new_wake_up.side_effect = inst.wake_up

circle/dashboard/views.py

@@ -423,7 +423,7 @@ class VmDetailView(CheckedDetailView):
         new_name = "Saved from %s (#%d) at %s" % (
             self.object.name, self.object.pk, date
         )
-        self.object.save_as_template_async(name=new_name,
+        self.object.save_as_template.async(name=new_name,
                                            user=request.user)
         messages.success(request, _("Saving instance as template!"))
         return redirect("%s#activity" % self.object.get_absolute_url())
@@ -433,7 +433,7 @@ class VmDetailView(CheckedDetailView):
         if not self.object.has_level(request.user, 'owner'):
             raise PermissionDenied()
 
-        self.object.shutdown_async(request.user)
+        self.object.shutdown.async(user=request.user)
         return redirect("%s#activity" % self.object.get_absolute_url())
 
     def __sleep(self, request):
@@ -441,7 +441,7 @@ class VmDetailView(CheckedDetailView):
         if not self.object.has_level(request.user, 'owner'):
             raise PermissionDenied()
 
-        self.object.sleep_async(request.user)
+        self.object.sleep.async(user=request.user)
         return redirect("%s#activity" % self.object.get_absolute_url())
 
     def __wake_up(self, request):
@@ -449,7 +449,7 @@ class VmDetailView(CheckedDetailView):
         if not self.object.has_level(request.user, 'owner'):
             raise PermissionDenied()
 
-        self.object.wake_up_async(request.user)
+        self.object.wake_up.async(user=request.user)
         return redirect("%s#activity" % self.object.get_absolute_url())
 
     def __deploy(self, request):
@@ -457,7 +457,7 @@ class VmDetailView(CheckedDetailView):
         if not self.object.has_level(request.user, 'owner'):
             raise PermissionDenied()
 
-        self.object.deploy_async(request.user)
+        self.object.deploy.async(user=request.user)
         return redirect("%s#activity" % self.object.get_absolute_url())
 
     def __reset(self, request):
@@ -465,7 +465,7 @@ class VmDetailView(CheckedDetailView):
         if not self.object.has_level(request.user, 'owner'):
             raise PermissionDenied()
 
-        self.object.reset_async(request.user)
+        self.object.reset.async(user=request.user)
         return redirect("%s#activity" % self.object.get_absolute_url())
 
     def __reboot(self, request):
@@ -473,7 +473,7 @@ class VmDetailView(CheckedDetailView):
         if not self.object.has_level(request.user, 'owner'):
             raise PermissionDenied()
 
-        self.object.reboot_async(request.user)
+        self.object.reboot.async(user=request.user)
         return redirect("%s#activity" % self.object.get_absolute_url())
 
     def __shut_off(self, request):
@@ -481,7 +481,7 @@ class VmDetailView(CheckedDetailView):
         if not self.object.has_level(request.user, 'owner'):
             raise PermissionDenied()
 
-        self.object.shut_off_async(request.user)
+        self.object.shut_off.async(user=request.user)
         return redirect("%s#activity" % self.object.get_absolute_url())
 
 
@@ -1148,7 +1148,7 @@ class VmCreate(LoginRequiredMixin, TemplateView):
 
     def __deploy(self, request, instances, *args, **kwargs):
         for i in instances:
-            i.deploy_async(user=request.user)
+            i.deploy.async(user=request.user)
 
         if len(instances) > 1:
             messages.success(request, _("Successfully created %d VMs!" %
@@ -1286,7 +1286,7 @@ class VmDelete(LoginRequiredMixin, DeleteView):
         if not object.has_level(request.user, 'owner'):
             raise PermissionDenied()
 
-        object.destroy_async(user=request.user)
+        object.destroy.async(user=request.user)
         success_url = self.get_success_url()
         success_message = _("VM successfully deleted!")
 
@@ -1466,7 +1466,7 @@ class NodeFlushView(LoginRequiredMixin, SuperuserRequiredMixin, DetailView):
 
     def __flush(self, request):
         self.object = self.get_object()
-        self.object.flush_async(user=request.user)
+        self.object.flush.async(user=request.user)
         success_message = _("Node successfully flushed!")
         messages.success(request, success_message)
         return redirect(self.get_success_url())
@@ -1537,7 +1537,7 @@ class VmMassDelete(LoginRequiredMixin, View):
                     raise PermissionDenied()  # no need for rollback or proper
                                             # error message, this can't
                                             # normally happen.
-                i.destroy_async(request.user)
+                i.destroy.async(user=request.user)
                 names.append(i.name)
 
         success_message = _("Mass delete complete, the following VMs were "
@@ -2066,7 +2066,7 @@ class VmMigrateView(SuperuserRequiredMixin, TemplateView):
 
         if node:
             node = Node.objects.get(pk=node)
-            vm.migrate_async(to_node=node, user=self.request.user)
+            vm.migrate.async(to_node=node, user=self.request.user)
         else:
             messages.error(self.request, _("You didn't select a node!"))
 

circle/vm/__init__.py

+# This import is responsible for running the operations' registration code.
+from . import operations  # noqa

circle/vm/models/activity.py

@@ -7,7 +7,11 @@ from django.db.models import CharField, ForeignKey
 from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
 
-from common.models import ActivityModel, activitycontextimpl
+from common.models import (
+    ActivityModel, activitycontextimpl, join_activity_code,
+)
+
+
 logger = getLogger(__name__)
 
 
@@ -24,6 +28,7 @@ class ActivityInProgressError(Exception):
 
 
 class InstanceActivity(ActivityModel):
+    ACTIVITY_CODE_BASE = join_activity_code('vm', 'Instance')
     instance = ForeignKey('Instance', related_name='activity_log',
                           help_text=_('Instance this activity works on.'),
                           verbose_name=_('instance'))
@@ -65,9 +70,10 @@ class InstanceActivity(ActivityModel):
         if concurrency_check and active_activities.exists():
             raise ActivityInProgressError(active_activities[0])
 
-        act = cls(activity_code='vm.Instance.' + code_suffix,
-                  instance=instance, parent=None, resultant_state=None,
-                  started=timezone.now(), task_uuid=task_uuid, user=user)
+        activity_code = join_activity_code(cls.ACTIVITY_CODE_BASE, code_suffix)
+        act = cls(activity_code=activity_code, instance=instance, parent=None,
+                  resultant_state=None, started=timezone.now(),
+                  task_uuid=task_uuid, user=user)
         act.save()
         return act
 
@@ -78,7 +84,7 @@ class InstanceActivity(ActivityModel):
             raise ActivityInProgressError(active_children[0])
 
         act = InstanceActivity(
-            activity_code=self.activity_code + '.' + code_suffix,
+            activity_code=join_activity_code(self.activity_code, code_suffix),
             instance=self.instance, parent=self, resultant_state=None,
             started=timezone.now(), task_uuid=task_uuid, user=self.user)
         act.save()
@@ -109,6 +115,7 @@ def instance_activity(code_suffix, instance, on_abort=None, on_commit=None,
 
 
 class NodeActivity(ActivityModel):
+    ACTIVITY_CODE_BASE = join_activity_code('vm', 'Node')
     node = ForeignKey('Node', related_name='activity_log',
                       help_text=_('Node this activity works on.'),
                       verbose_name=_('node'))
@@ -131,15 +138,15 @@ class NodeActivity(ActivityModel):
 
     @classmethod
     def create(cls, code_suffix, node, task_uuid=None, user=None):
-        act = cls(activity_code='vm.Node.' + code_suffix,
-                  node=node, parent=None, started=timezone.now(),
-                  task_uuid=task_uuid, user=user)
+        activity_code = join_activity_code(cls.ACTIVITY_CODE_BASE, code_suffix)
+        act = cls(activity_code=activity_code, node=node, parent=None,
+                  started=timezone.now(), task_uuid=task_uuid, user=user)
         act.save()
         return act
 
     def create_sub(self, code_suffix, task_uuid=None):
         act = NodeActivity(
-            activity_code=self.activity_code + '.' + code_suffix,
+            activity_code=join_activity_code(self.activity_code, code_suffix),
             node=self.node, parent=self, started=timezone.now(),
             task_uuid=task_uuid, user=self.user)
         act.save()

circle/vm/models/instance.py

 from __future__ import absolute_import, unicode_literals
 from datetime import timedelta
-from logging import getLogger
 from importlib import import_module
+from logging import getLogger
+from string import ascii_lowercase
 from warnings import warn
-import string
 
 import django.conf
 from django.contrib.auth.models import User
@@ -16,14 +16,14 @@ from django.dispatch import Signal
 from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
 
-from celery.exceptions import TimeLimitExceeded
 from model_utils import Choices
 from model_utils.models import TimeStampedModel, StatusModel
 from taggit.managers import TaggableManager
 
 from acl.models import AclBase
+from common.operations import OperatedMixin
 from storage.models import Disk
-from ..tasks import local_tasks, vm_tasks, agent_tasks
+from ..tasks import vm_tasks, agent_tasks
 from .activity import (ActivityInProgressError, instance_activity,
                        InstanceActivity)
 from .common import BaseResourceConfigModel, Lease
@@ -162,7 +162,7 @@ class InstanceTemplate(AclBase, VirtualMachineDescModel, TimeStampedModel):
         return ('dashboard.views.template-detail', None, {'pk': self.pk})
 
 
-class Instance(AclBase, VirtualMachineDescModel, StatusModel,
+class Instance(AclBase, VirtualMachineDescModel, StatusModel, OperatedMixin,
                TimeStampedModel):
 
     """Virtual machine instance.
@@ -727,479 +727,136 @@ class Instance(AclBase, VirtualMachineDescModel, StatusModel,
         """
         return scheduler.select_node(self, Node.objects.all())
 
-    def __schedule_vm(self, act):
-        """Schedule the virtual machine as part of a higher level activity.
-
-        :param act: Parent activity.
+    def deploy_disks(self):
+        """Deploy all associated disks.
         """
-        # Find unused port for VNC
-        if self.vnc_port is None:
-            self.vnc_port = find_unused_vnc_port()
-
-        # Schedule
-        if self.node is None:
-            self.node = self.select_node()
-
-        self.save()
-
-    def __deploy_vm(self, act, timeout=15):
-        """Deploy the virtual machine.
-
-        :param self: The virtual machine.
+        devnums = list(ascii_lowercase)  # a-z
+        for disk in self.disks.all():
+            # assign device numbers
+            if disk.dev_num in devnums:
+                devnums.remove(disk.dev_num)
+            else:
+                disk.dev_num = devnums.pop(0)
+                disk.save()
+            # deploy disk
+            disk.deploy()
 
-        :param act: Parent activity.
+    def destroy_disks(self):
+        """Destroy all associated disks.
         """
-        queue_name = self.get_remote_queue_name('vm')
-
-        # Deploy VM on remote machine
-        with act.sub_activity('deploying_vm') as deploy_act:
-            deploy_act.result = vm_tasks.deploy.apply_async(
-                args=[self.get_vm_desc()],
-                queue=queue_name).get(timeout=timeout)
-
-        # Estabilish network connection (vmdriver)
-        with act.sub_activity('deploying_net'):
-            for net in self.interface_set.all():
-                net.deploy()
+        for disk in self.disks.all():
+            disk.destroy()
 
-        # Resume vm
-        with act.sub_activity('booting'):
-            vm_tasks.resume.apply_async(args=[self.vm_name],
-                                        queue=queue_name).get(timeout=timeout)
-
-        self.renew(which='both', base_activity=act)
-
-    def deploy(self, user=None, task_uuid=None):
-        """Deploy new virtual machine with network
-
-        :param self: The virtual machine to deploy.
-        :type self: vm.models.Instance
-
-        :param user: The user who's issuing the command.
-        :type user: django.contrib.auth.models.User
-
-        :param task_uuid: The task's UUID, if the command is being executed
-                          asynchronously.
-        :type task_uuid: str
+    def deploy_net(self):
+        """Deploy all associated network interfaces.
         """
-        if self.destroyed_at:
-            raise self.InstanceDestroyedError(self)
+        for net in self.interface_set.all():
+            net.deploy()
 
-        def __on_commit(activity):
-            activity.resultant_state = 'RUNNING'
-
-        with instance_activity(code_suffix='deploy', instance=self,
-                               on_commit=__on_commit, task_uuid=task_uuid,
-                               user=user) as act:
-
-            self.__schedule_vm(act)
-
-            # Deploy virtual images
-            with act.sub_activity('deploying_disks'):
-                devnums = list(string.ascii_lowercase)  # a-z
-                for disk in self.disks.all():
-                    # assign device numbers
-                    if disk.dev_num in devnums:
-                        devnums.remove(disk.dev_num)
-                    else:
-                        disk.dev_num = devnums.pop(0)
-                        disk.save()
-                    # deploy disk
-                    disk.deploy()
-
-            self.__deploy_vm(act)
-
-    def deploy_async(self, user=None):
-        """Execute deploy asynchronously.
+    def destroy_net(self):
+        """Destroy all associated network interfaces.
         """
-        logger.debug('Calling async local_tasks.deploy(%s, %s)',
-                     unicode(self), unicode(user))
-        return local_tasks.deploy.apply_async(args=[self, user],
-                                              queue="localhost.man")
+        for net in self.interface_set.all():
+            net.destroy()
 
-    def __destroy_vm(self, act, timeout=15):
-        """Destroy the virtual machine and its associated networks.
-
-        :param self: The virtual machine.
-
-        :param act: Parent activity.
+    def shutdown_net(self):
+        """Shutdown all associated network interfaces.
         """
-        # Destroy networks
-        with act.sub_activity('destroying_net'):
-            for net in self.interface_set.all():
-                net.destroy()
-
-        # Destroy virtual machine
-        with act.sub_activity('destroying_vm'):
-            queue_name = self.get_remote_queue_name('vm')
-            try:
-                vm_tasks.destroy.apply_async(args=[self.vm_name],
-                                             queue=queue_name
-                                             ).get(timeout=timeout)
-            except Exception as e:
-                if e.libvirtError and "Domain not found" in str(e):
-                    logger.debug("Domain %s was not found at %s"
-                                 % (self.vm_name, queue_name))
-                else:
-                    raise
+        for net in self.interface_set.all():
+            net.shutdown()
 
-    def __cleanup_after_destroy_vm(self, act, timeout=15):
-        """Clean up the virtual machine's data after destroy.
-
-        :param self: The virtual machine.
-
-        :param act: Parent activity.
-        """
-        # Delete mem. dump if exists
+    def delete_vm(self, timeout=15):
+        queue_name = self.get_remote_queue_name('vm')
         try:
-            queue_name = self.mem_dump['datastore'].get_remote_queue_name(
-                'storage')
-            from storage.tasks.remote_tasks import delete_dump
-            delete_dump.apply_async(args=[self.mem_dump['path']],
-                                    queue=queue_name).get(timeout=timeout)
-        except:
-            pass
-
-        # Clear node and VNC port association
-        self.node = None
-        self.vnc_port = None
-
-    def redeploy(self, user=None, task_uuid=None):
-        """Redeploy virtual machine with network
-
-        :param self: The virtual machine to redeploy.
-
-        :param user: The user who's issuing the command.
-        :type user: django.contrib.auth.models.User
-
-        :param task_uuid: The task's UUID, if the command is being executed
-                          asynchronously.
-        :type task_uuid: str
-        """
-        with instance_activity(code_suffix='redeploy', instance=self,
-                               task_uuid=task_uuid, user=user) as act:
-            # Destroy VM
-            if self.node:
-                self.__destroy_vm(act)
-
-            self.__cleanup_after_destroy_vm(act)
-
-            # Deploy VM
-            self.__schedule_vm(act)
-
-            self.__deploy_vm(act)
-
-    def redeploy_async(self, user=None):
-        """Execute redeploy asynchronously.
-        """
-        return local_tasks.redeploy.apply_async(args=[self, user],
-                                                queue="localhost.man")
-
-    def shut_off(self, user=None, task_uuid=None):
-        """Shut off VM. (plug-out)
-        """
-        def __on_commit(activity):
-            activity.resultant_state = 'STOPPED'
-
-        with instance_activity(code_suffix='shut_off', instance=self,
-                               task_uuid=task_uuid, user=user,
-                               on_commit=__on_commit) as act:
-            # Destroy VM
-            if self.node:
-                self.__destroy_vm(act)
-
-            self.__cleanup_after_destroy_vm(act)
-            self.save()
-
-    def shut_off_async(self, user=None):
-        """Shut off VM. (plug-out)
-        """
-        return local_tasks.shut_off.apply_async(args=[self, user],
-                                                queue="localhost.man")
-
-    def destroy(self, user=None, task_uuid=None):
-        """Remove virtual machine and its networks.
-
-        :param self: The virtual machine to destroy.
-        :type self: vm.models.Instance
-
-        :param user: The user who's issuing the command.
-        :type user: django.contrib.auth.models.User
-
-        :param task_uuid: The task's UUID, if the command is being executed
-                          asynchronously.
-        :type task_uuid: str
-        """
-        if self.destroyed_at:
-            return  # already destroyed, nothing to do here
-
-        def __on_commit(activity):
-            activity.resultant_state = 'DESTROYED'
-
-        with instance_activity(code_suffix='destroy', instance=self,
-                               on_commit=__on_commit, task_uuid=task_uuid,
-                               user=user) as act:
-
-            if self.node:
-                self.__destroy_vm(act)
-
-            # Destroy disks
-            with act.sub_activity('destroying_disks'):
-                for disk in self.disks.all():
-                    disk.destroy()
-
-            self.__cleanup_after_destroy_vm(act)
-
-            self.destroyed_at = timezone.now()
-            self.save()
-
-    def destroy_async(self, user=None):
-        """Execute destroy asynchronously.
-        """
-        return local_tasks.destroy.apply_async(args=[self, user],
-                                               queue="localhost.man")
-
-    def sleep(self, user=None, task_uuid=None, timeout=60):
-        """Suspend virtual machine with memory dump.
-        """
-        if self.status not in ['RUNNING']:
-            raise self.WrongStateError(self)
-
-        def __on_abort(activity, error):
-            if isinstance(error, TimeLimitExceeded):
-                activity.resultant_state = None
+            return vm_tasks.destroy.apply_async(args=[self.vm_name],
+                                                queue=queue_name
+                                                ).get(timeout=timeout)
+        except Exception as e:
+            if e.libvirtError and "Domain not found" in str(e):
+                logger.debug("Domain %s was not found at %s"
+                             % (self.vm_name, queue_name))
             else:
-                activity.resultant_state = 'ERROR'
+                raise
 
-        def __on_commit(activity):
-            activity.resultant_state = 'SUSPENDED'
-
-        with instance_activity(code_suffix='sleep', instance=self,
-                               on_abort=__on_abort, on_commit=__on_commit,
-                               task_uuid=task_uuid, user=user) as act:
-
-            # Destroy networks
-            with act.sub_activity('destroying_net'):
-                for net in self.interface_set.all():
-                    net.shutdown()
-
-            # Suspend vm
-            with act.sub_activity('suspending'):
-                queue_name = self.get_remote_queue_name('vm')
-                vm_tasks.sleep.apply_async(args=[self.vm_name,
-                                                 self.mem_dump['path']],
+    def deploy_vm(self, timeout=15):
+        queue_name = self.get_remote_queue_name('vm')
+        return vm_tasks.deploy.apply_async(args=[self.get_vm_desc()],
                                            queue=queue_name
                                            ).get(timeout=timeout)
-                self.node = None
-                self.save()
-
-    def sleep_async(self, user=None):
-        """Execute sleep asynchronously.
-        """
-        return local_tasks.sleep.apply_async(args=[self, user],
-                                             queue="localhost.man")
-
-    def wake_up(self, user=None, task_uuid=None, timeout=60):
-        """ Wake up Virtual Machine from SUSPENDED state.
-
-        Power on Virtual Machine and load its memory from dump.
-        """
-        if self.status not in ['SUSPENDED']:
-            raise self.WrongStateError(self)
 
-        def __on_abort(activity, error):
-            activity.resultant_state = 'ERROR'
+    def migrate_vm(self, to_node, timeout=120):
+        queue_name = self.get_remote_queue_name('vm')
+        return vm_tasks.migrate.apply_async(args=[self.vm_name,
+                                                  to_node.host.hostname],
+                                            queue=queue_name
+                                            ).get(timeout=timeout)
 
-        def __on_commit(activity):
-            activity.resultant_state = 'RUNNING'
+    def reboot_vm(self, timeout=5):
+        queue_name = self.get_remote_queue_name('vm')
+        return vm_tasks.reboot.apply_async(args=[self.vm_name],
+                                           queue=queue_name
+                                           ).get(timeout=timeout)
 
-        with instance_activity(code_suffix='wake_up', instance=self,
-                               on_abort=__on_abort, on_commit=__on_commit,
-                               task_uuid=task_uuid, user=user) as act:
+    def reset_vm(self, timeout=5):
+        queue_name = self.get_remote_queue_name('vm')
+        return vm_tasks.reset.apply_async(args=[self.vm_name],
+                                          queue=queue_name
+                                          ).get(timeout=timeout)
 
-            # Schedule vm
-            self.__schedule_vm(act)
-            queue_name = self.get_remote_queue_name('vm')
+    def resume_vm(self, timeout=15):
+        queue_name = self.get_remote_queue_name('vm')
+        return vm_tasks.resume.apply_async(args=[self.vm_name],
+                                           queue=queue_name
+                                           ).get(timeout=timeout)
 
-            # Resume vm
-            with act.sub_activity('resuming'):
-                vm_tasks.wake_up.apply_async(args=[self.vm_name,
-                                                   self.mem_dump['path']],
+    def shutdown_vm(self, timeout=120):
+        queue_name = self.get_remote_queue_name('vm')
+        logger.debug("RPC Shutdown at queue: %s, for vm: %s.", queue_name,
+                     self.vm_name)
+        return vm_tasks.shutdown.apply_async(kwargs={'name': self.vm_name},
                                              queue=queue_name
                                              ).get(timeout=timeout)
 
-            # Estabilish network connection (vmdriver)
-            with act.sub_activity('deploying_net'):
-                for net in self.interface_set.all():
-                    net.deploy()
-
-            # Renew vm
-            self.renew(which='both', base_activity=act)
-
-    def wake_up_async(self, user=None):
-        """Execute wake_up asynchronously.
-        """
-        return local_tasks.wake_up.apply_async(args=[self, user],
-                                               queue="localhost.man")
-
-    def shutdown(self, user=None, task_uuid=None, timeout=120):
-        """Shutdown virtual machine with ACPI signal.
-        """
-        def __on_abort(activity, error):
-            if isinstance(error, TimeLimitExceeded):
-                activity.resultant_state = None
-            else:
-                activity.resultant_state = 'ERROR'
-
-        def __on_commit(activity):
-            activity.resultant_state = 'STOPPED'
-
-        with instance_activity(code_suffix='shutdown', instance=self,
-                               on_abort=__on_abort, on_commit=__on_commit,
-                               task_uuid=task_uuid, user=user):
-            queue_name = self.get_remote_queue_name('vm')
-            logger.debug("RPC Shutdown at queue: %s, for vm: %s.",
-                         self.vm_name, queue_name)
-            vm_tasks.shutdown.apply_async(kwargs={'name': self.vm_name},
+    def suspend_vm(self, timeout=60):
+        queue_name = self.get_remote_queue_name('vm')
+        return vm_tasks.sleep.apply_async(args=[self.vm_name,
+                                                self.mem_dump['path']],
                                           queue=queue_name
                                           ).get(timeout=timeout)
-            self.node = None
-            self.vnc_port = None
-            self.save()
-
-    def shutdown_async(self, user=None):
-        """Execute shutdown asynchronously.
-        """
-        return local_tasks.shutdown.apply_async(args=[self, user],
-                                                queue="localhost.man")
 
-    def reset(self, user=None, task_uuid=None, timeout=5):
-        """Reset virtual machine (reset button)
-        """
-        with instance_activity(code_suffix='reset', instance=self,
-                               task_uuid=task_uuid, user=user):
-
-            queue_name = self.get_remote_queue_name('vm')
-            vm_tasks.reset.apply_async(args=[self.vm_name],
-                                       queue=queue_name
-                                       ).get(timeout=timeout)
-
-    def reset_async(self, user=None):
-        """Execute reset asynchronously.
-        """
-        return local_tasks.reset.apply_async(args=[self, user],
-                                             queue="localhost.man")
-
-    def reboot(self, user=None, task_uuid=None, timeout=5):
-        """Reboot virtual machine with Ctrl+Alt+Del signal.
-        """
-        with instance_activity(code_suffix='reboot', instance=self,
-                               task_uuid=task_uuid, user=user):
-
-            queue_name = self.get_remote_queue_name('vm')
-            vm_tasks.reboot.apply_async(args=[self.vm_name],
-                                        queue=queue_name
-                                        ).get(timeout=timeout)
-
-    def reboot_async(self, user=None):
-        """Execute reboot asynchronously. """
-        return local_tasks.reboot.apply_async(args=[self, user],
-                                              queue="localhost.man")
-
-    def migrate_async(self, to_node, user=None):
-        """Execute migrate asynchronously. """
-        return local_tasks.migrate.apply_async(args=[self, to_node, user],
-                                               queue="localhost.man")
-
-    def migrate(self, to_node=None, user=None, task_uuid=None, timeout=120):
-        """Live migrate running vm to another node. """
-        with instance_activity(code_suffix='migrate', instance=self,
-                               task_uuid=task_uuid, user=user) as act:
-            if not to_node:
-                with act.sub_activity('scheduling') as sa:
-                    to_node = self.select_node()
-                    sa.result = to_node
-
-            # Destroy networks
-            with act.sub_activity('destroying_net'):
-                for net in self.interface_set.all():
-                    net.shutdown()
-
-            with act.sub_activity('migrate_vm'):
-                queue_name = self.get_remote_queue_name('vm')
-                vm_tasks.migrate.apply_async(args=[self.vm_name,
-                                             to_node.host.hostname],
-                                             queue=queue_name
-                                             ).get(timeout=timeout)
-            # Refresh node information
-            self.node = to_node
+    def wake_up_vm(self, timeout=60):
+        queue_name = self.get_remote_queue_name('vm')
+        return vm_tasks.wake_up.apply_async(args=[self.vm_name,
+                                                  self.mem_dump['path']],
+                                            queue=queue_name
+                                            ).get(timeout=timeout)
+
+    def delete_mem_dump(self, timeout=15):
+        queue_name = self.mem_dump['datastore'].get_remote_queue_name(
+            'storage')
+        from storage.tasks.remote_tasks import delete_dump
+        delete_dump.apply_async(args=[self.mem_dump['path']],
+                                queue=queue_name).get(timeout=timeout)
+
+    def allocate_node(self):
+        if self.node is None:
+            self.node = self.select_node()
             self.save()
-            # Estabilish network connection (vmdriver)
-            with act.sub_activity('deploying_net'):
-                for net in self.interface_set.all():
-                    net.deploy()
-
-    def save_as_template_async(self, name, user=None, **kwargs):
-        """ Save as template asynchronusly.
-        """
-        return local_tasks.save_as_template.apply_async(
-            args=[self, name, user, kwargs], queue="localhost.man")
 
-    def save_as_template(self, name, task_uuid=None, user=None,
-                         timeout=300, **kwargs):
-        """ Save Virtual Machine as a Template.
+    def yield_node(self):
+        if self.node is not None:
+            self.node = None
+            self.save()
 
-        Template can be shared with groups and users.
-        Users can instantiate Virtual Machines from Templates.
-        """
-        with instance_activity(code_suffix="save_as_template", instance=self,
-                               task_uuid=task_uuid, user=user) as act:
-            # prepare parameters
-            params = {
-                'access_method': self.access_method,
-                'arch': self.arch,
-                'boot_menu': self.boot_menu,
-                'description': self.description,
-                'lease': self.lease,  # Can be problem in new VM
-                'max_ram_size': self.max_ram_size,
-                'name': name,
-                'num_cores': self.num_cores,
-                'owner': user,
-                'parent': self.template,  # Can be problem
-                'priority': self.priority,
-                'ram_size': self.ram_size,
-                'raw_data': self.raw_data,
-                'system': self.system,
-            }
-            params.update(kwargs)
-
-            def __try_save_disk(disk):
-                try:
-                    return disk.save_as()  # can do in parallel
-                except Disk.WrongDiskTypeError:
-                    return disk
-
-            # create template and do additional setup
-            tmpl = InstanceTemplate(**params)
-            tmpl.full_clean()  # Avoiding database errors.
-            tmpl.save()
-            try:
-                with act.sub_activity('saving_disks'):
-                    tmpl.disks.add(*[__try_save_disk(disk)
-                                   for disk in self.disks.all()])
-                # create interface templates
-                for i in self.interface_set.all():
-                    i.save_as_template(tmpl)
-            except:
-                tmpl.delete()
-                raise
-            else:
-                return tmpl
+    def allocate_vnc_port(self):
+        if self.vnc_port is None:
+            self.vnc_port = find_unused_vnc_port()
+            self.save()
 
-    def shutdown_and_save_as_template(self, name, user=None, task_uuid=None,
-                                      **kwargs):
-        self.shutdown(user, task_uuid)
-        self.save_as_template(name, **kwargs)
+    def yield_vnc_port(self):
+        if self.vnc_port is not None:
+            self.vnc_port = None
+            self.save()
 
     def get_status_icon(self):
         return {

circle/vm/models/network.py

@@ -137,7 +137,7 @@ class Interface(Model):
         iface.save()
         return iface
 
-    def deploy(self, user=None, task_uuid=None):
+    def deploy(self):
         if self.destroyed:
             from .instance import Instance
             raise Instance.InstanceDestroyedError(self.instance,
@@ -149,16 +149,23 @@ class Interface(Model):
             args=[self.get_vmnetwork_desc()],
             queue=self.instance.get_remote_queue_name('net'))
 
-    def shutdown(self, user=None, task_uuid=None):
-        net_tasks.destroy.apply_async(
-            args=[self.get_vmnetwork_desc()],
-            queue=self.instance.get_remote_queue_name('net'))
+    def shutdown(self):
+        if self.destroyed:
+            from .instance import Instance
+            raise Instance.InstanceDestroyedError(self.instance,
+                                                  "The associated instance "
+                                                  "(%s) has already been "
+                                                  "destroyed" % self.instance)
+
+        queue_name = self.instance.get_remote_queue_name('net')
+        net_tasks.destroy.apply_async(args=[self.get_vmnetwork_desc()],
+                                      queue=queue_name)
 
-    def destroy(self, user=None, task_uuid=None):
+    def destroy(self):
         if self.destroyed:
             return
 
-        self.shutdown(user, task_uuid)
+        self.shutdown()
         if self.host is not None:
             self.host.delete()
 

circle/vm/models/node.py

 from __future__ import absolute_import, unicode_literals
 from logging import getLogger
+from warnings import warn
 
 from django.db.models import (
     CharField, IntegerField, ForeignKey, BooleanField, ManyToManyField,
     FloatField, permalink,
 )
+from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
 
-from warnings import warn
-
 from celery.exceptions import TimeoutError
 from model_utils.models import TimeStampedModel
 from taggit.managers import TaggableManager
 
 from common.models import method_cache, WorkerNotFound, HumanSortField
+from common.operations import OperatedMixin
 from firewall.models import Host
-from ..tasks import vm_tasks, local_tasks
-from .common import Trait
-
-from .activity import node_activity, NodeActivity
-
 from monitor.calvin.calvin import Query
 from monitor.calvin.calvin import GraphiteHandler
-from django.utils import timezone
+from ..tasks import vm_tasks
+from .activity import node_activity, NodeActivity
+from .common import Trait
+
 
 logger = getLogger(__name__)
 
@@ -38,7 +37,7 @@ def node_available(function):
     return decorate
 
 
-class Node(TimeStampedModel):
+class Node(OperatedMixin, TimeStampedModel):
 
     """A VM host machine, a hypervisor.
     """
@@ -131,22 +130,6 @@ class Node(TimeStampedModel):
                 self.enabled = False
                 self.save()
 
-    def flush(self, user=None, task_uuid=None):
-        """Disable node and move all instances to other ones.
-        """
-        with node_activity('flush', node=self, user=user,
-                           task_uuid=task_uuid) as act:
-            self.disable(user, act)
-            for i in self.instance_set.all():
-                with act.sub_activity('migrate_instance_%d' % i.pk):
-                    i.migrate()
-
-    def flush_async(self, user=None):
-        """Execute flush asynchronously.
-        """
-        return local_tasks.flush.apply_async(args=[self, user],
-                                             queue="localhost.man")
-
     def enable(self, user=None):
         ''' Enable the node. '''
         if self.enabled is not True:
@@ -164,10 +147,10 @@ class Node(TimeStampedModel):
 
     @method_cache(30)
     def get_remote_queue_name(self, queue_id):
-        """Return the name of the remote celery queue for this node.
+        """Returns the name of the remote celery queue for this node.
 
-        throws Exception if there is no worker on the queue.
-        Until the cache provide reult there can be dead queues.
+        Throws Exception if there is no worker on the queue.
+        The result may include dead queues because of caching.
         """
 
         if vm_tasks.check_queue(self.host.hostname, queue_id):
@@ -189,7 +172,7 @@ class Node(TimeStampedModel):
         else:
             logger.debug("The last activity was %s" % act)
             if act.activity_code.endswith("offline"):
-                act = NodeActivity.create(code_suffix='monitor_succes_online',
+                act = NodeActivity.create(code_suffix='monitor_success_online',
                                           node=self, user=None)
                 act.started = timezone.now()
                 act.finished = timezone.now()

circle/vm/operations.py

+from __future__ import absolute_import, unicode_literals
+from logging import getLogger
+
+from django.core.exceptions import PermissionDenied
+from django.utils import timezone
+from django.utils.translation import ugettext_lazy as _
+
+from celery.exceptions import TimeLimitExceeded
+
+from common.operations import Operation, register_operation
+from storage.models import Disk
+from .tasks.local_tasks import async_instance_operation, async_node_operation
+from .models import (
+    Instance, InstanceActivity, InstanceTemplate, Node, NodeActivity,
+)
+
+
+logger = getLogger(__name__)
+
+
+class InstanceOperation(Operation):
+    acl_level = 'owner'
+    async_operation = async_instance_operation
+
+    def __init__(self, instance):
+        super(InstanceOperation, self).__init__(subject=instance)
+        self.instance = instance
+
+    def check_precond(self):
+        if self.instance.destroyed_at:
+            raise self.instance.InstanceDestroyedError(self.instance)
+
+    def check_auth(self, user):
+        if not self.instance.has_level(user, self.acl_level):
+            raise PermissionDenied("%s doesn't have the required ACL level." %
+                                   user)
+
+        super(InstanceOperation, self).check_auth(user=user)
+
+    def create_activity(self, parent, user):
+        if parent:
+            if parent.instance != self.instance:
+                raise ValueError("The instance associated with the specified "
+                                 "parent activity does not match the instance "
+                                 "bound to the operation.")
+            if parent.user != user:
+                raise ValueError("The user associated with the specified "
+                                 "parent activity does not match the user "
+                                 "provided as parameter.")
+
+            return parent.create_sub(code_suffix=self.activity_code_suffix)
+        else:
+            return InstanceActivity.create(
+                code_suffix=self.activity_code_suffix, instance=self.instance,
+                user=user)
+
+
+def register_instance_operation(op_cls, op_id=None):
+    return register_operation(Instance, op_cls, op_id)
+
+
+class DeployOperation(InstanceOperation):
+    activity_code_suffix = 'deploy'
+    id = 'deploy'
+    name = _("deploy")
+    description = _("Deploy new virtual machine with network.")
+
+    def on_commit(self, activity):
+        activity.resultant_state = 'RUNNING'
+
+    def _operation(self, activity, user, system, timeout=15):
+        # Allocate VNC port and host node
+        self.instance.allocate_vnc_port()
+        self.instance.allocate_node()
+
+        # Deploy virtual images
+        with activity.sub_activity('deploying_disks'):
+            self.instance.deploy_disks()
+
+        # Deploy VM on remote machine
+        with activity.sub_activity('deploying_vm') as deploy_act:
+            deploy_act.result = self.instance.deploy_vm(timeout=timeout)
+
+        # Establish network connection (vmdriver)
+        with activity.sub_activity('deploying_net'):
+            self.instance.deploy_net()
+
+        # Resume vm
+        with activity.sub_activity('booting'):
+            self.instance.resume_vm(timeout=timeout)
+
+        self.instance.renew(which='both', base_activity=activity)
+
+
+register_instance_operation(DeployOperation)
+
+
+class DestroyOperation(InstanceOperation):
+    activity_code_suffix = 'destroy'
+    id = 'destroy'
+    name = _("destroy")
+    description = _("Destroy virtual machine and its networks.")
+
+    def on_commit(self, activity):
+        activity.resultant_state = 'DESTROYED'
+
+    def _operation(self, activity, user, system):
+        if self.instance.node:
+            # Destroy networks
+            with activity.sub_activity('destroying_net'):
+                self.instance.destroy_net()
+
+            # Delete virtual machine
+            with activity.sub_activity('destroying_vm'):
+                self.instance.delete_vm()
+
+        # Destroy disks
+        with activity.sub_activity('destroying_disks'):
+            self.instance.destroy_disks()
+
+        # Delete mem. dump if exists
+        try:
+            self.instance.delete_mem_dump()
+        except:
+            pass
+
+        # Clear node and VNC port association
+        self.instance.yield_node()
+        self.instance.yield_vnc_port()
+
+        self.instance.destroyed_at = timezone.now()
+        self.instance.save()
+
+
+register_instance_operation(DestroyOperation)
+
+
+class MigrateOperation(InstanceOperation):
+    activity_code_suffix = 'migrate'
+    id = 'migrate'
+    name = _("migrate")
+    description = _("Live migrate running VM to another node.")
+
+    def _operation(self, activity, user, system, to_node=None, timeout=120):
+        if not to_node:
+            with activity.sub_activity('scheduling') as sa:
+                to_node = self.instance.select_node()
+                sa.result = to_node
+
+        # Shutdown networks
+        with activity.sub_activity('shutdown_net'):
+            self.instance.shutdown_net()
+
+        with activity.sub_activity('migrate_vm'):
+            self.instance.migrate_vm(to_node=to_node, timeout=timeout)
+
+        # Refresh node information
+        self.instance.node = to_node
+        self.instance.save()
+        # Estabilish network connection (vmdriver)
+        with activity.sub_activity('deploying_net'):
+            self.instance.deploy_net()
+
+
+register_instance_operation(MigrateOperation)
+
+
+class RebootOperation(InstanceOperation):
+    activity_code_suffix = 'reboot'
+    id = 'reboot'
+    name = _("reboot")
+    description = _("Reboot virtual machine with Ctrl+Alt+Del signal.")
+
+    def _operation(self, activity, user, system, timeout=5):
+        self.instance.reboot_vm(timeout=timeout)
+
+
+register_instance_operation(RebootOperation)
+
+
+class ResetOperation(InstanceOperation):
+    activity_code_suffix = 'reset'
+    id = 'reset'
+    name = _("reset")
+    description = _("Reset virtual machine (reset button).")
+
+    def _operation(self, activity, user, system, timeout=5):
+        self.instance.reset_vm(timeout=timeout)
+
+register_instance_operation(ResetOperation)
+
+
+class SaveAsTemplateOperation(InstanceOperation):
+    activity_code_suffix = 'save_as_template'
+    id = 'save_as_template'
+    name = _("save as template")
+    description = _("""Save Virtual Machine as a Template.
+
+        Template can be shared with groups and users.
+        Users can instantiate Virtual Machines from Templates.
+        """)
+
+    def _operation(self, activity, name, user, system, timeout=300,
+                   with_shutdown=True, **kwargs):
+        if with_shutdown:
+            ShutdownOperation(self.instance).call(parent_activity=activity,
+                                                  user=user)
+        # prepare parameters
+        params = {
+            'access_method': self.instance.access_method,
+            'arch': self.instance.arch,
+            'boot_menu': self.instance.boot_menu,
+            'description': self.instance.description,
+            'lease': self.instance.lease,  # Can be problem in new VM
+            'max_ram_size': self.instance.max_ram_size,
+            'name': name,
+            'num_cores': self.instance.num_cores,
+            'owner': user,
+            'parent': self.instance.template,  # Can be problem
+            'priority': self.instance.priority,
+            'ram_size': self.instance.ram_size,
+            'raw_data': self.instance.raw_data,
+            'system': self.instance.system,
+        }
+        params.update(kwargs)
+
+        def __try_save_disk(disk):
+            try:
+                return disk.save_as()
+            except Disk.WrongDiskTypeError:
+                return disk
+
+        # create template and do additional setup
+        tmpl = InstanceTemplate(**params)
+        tmpl.full_clean()  # Avoiding database errors.
+        tmpl.save()
+        try:
+            with activity.sub_activity('saving_disks'):
+                tmpl.disks.add(*[__try_save_disk(disk)
+                                 for disk in self.instance.disks.all()])
+            # create interface templates
+            for i in self.instance.interface_set.all():
+                i.save_as_template(tmpl)
+        except:
+            tmpl.delete()
+            raise
+        else:
+            return tmpl
+
+
+register_instance_operation(SaveAsTemplateOperation)
+
+
+class ShutdownOperation(InstanceOperation):
+    activity_code_suffix = 'shutdown'
+    id = 'shutdown'
+    name = _("shutdown")
+    description = _("Shutdown virtual machine with ACPI signal.")
+
+    def on_abort(self, activity, error):
+        if isinstance(error, TimeLimitExceeded):
+            activity.resultant_state = None
+        else:
+            activity.resultant_state = 'ERROR'
+
+    def on_commit(self, activity):
+        activity.resultant_state = 'STOPPED'
+
+    def _operation(self, activity, user, system, timeout=120):
+        self.instance.shutdown_vm(timeout=timeout)
+        self.instance.yield_node()
+        self.instance.yield_vnc_port()
+
+
+register_instance_operation(ShutdownOperation)
+
+
+class ShutOffOperation(InstanceOperation):
+    activity_code_suffix = 'shut_off'
+    id = 'shut_off'
+    name = _("shut off")
+    description = _("Shut off VM (plug-out).")
+
+    def on_commit(activity):
+        activity.resultant_state = 'STOPPED'
+
+    def _operation(self, activity, user, system):
+        # Shutdown networks
+        with activity.sub_activity('shutdown_net'):
+            self.instance.shutdown_net()
+
+        # Delete virtual machine
+        with activity.sub_activity('delete_vm'):
+            self.instance.delete_vm()
+
+        # Clear node and VNC port association
+        self.instance.yield_node()
+        self.instance.yield_vnc_port()
+
+
+register_instance_operation(ShutOffOperation)
+
+
+class SleepOperation(InstanceOperation):
+    activity_code_suffix = 'sleep'
+    id = 'sleep'
+    name = _("sleep")
+    description = _("Suspend virtual machine with memory dump.")
+
+    def check_precond(self):
+        super(SleepOperation, self).check_precond()
+        if self.instance.status not in ['RUNNING']:
+            raise self.instance.WrongStateError(self.instance)
+
+    def on_abort(self, activity, error):
+        if isinstance(error, TimeLimitExceeded):
+            activity.resultant_state = None
+        else:
+            activity.resultant_state = 'ERROR'
+
+    def on_commit(self, activity):
+        activity.resultant_state = 'SUSPENDED'
+
+    def _operation(self, activity, user, system, timeout=60):
+        # Destroy networks
+        with activity.sub_activity('shutdown_net'):
+            self.instance.shutdown_net()
+
+        # Suspend vm
+        with activity.sub_activity('suspending'):
+            self.instance.suspend_vm(timeout=timeout)
+
+        self.instance.yield_node()
+        # VNC port needs to be kept
+
+
+register_instance_operation(SleepOperation)
+
+
+class WakeUpOperation(InstanceOperation):
+    activity_code_suffix = 'wake_up'
+    id = 'wake_up'
+    name = _("wake up")
+    description = _("""Wake up Virtual Machine from SUSPENDED state.
+
+        Power on Virtual Machine and load its memory from dump.
+        """)
+
+    def check_precond(self):
+        super(WakeUpOperation, self).check_precond()
+        if self.instance.status not in ['SUSPENDED']:
+            raise self.instance.WrongStateError(self.instance)
+
+    def on_abort(self, activity, error):
+        activity.resultant_state = 'ERROR'
+
+    def on_commit(self, activity):
+        activity.resultant_state = 'RUNNING'
+
+    def _operation(self, activity, user, system, timeout=60):
+        # Schedule vm
+        self.instance.allocate_vnc_port()
+        self.instance.allocate_node()
+
+        # Resume vm
+        with activity.sub_activity('resuming'):
+            self.instance.wake_up_vm(timeout=timeout)
+
+        # Estabilish network connection (vmdriver)
+        with activity.sub_activity('deploying_net'):
+            self.instance.deploy_net()
+
+        # Renew vm
+        self.instance.renew(which='both', base_activity=activity)
+
+
+register_instance_operation(WakeUpOperation)
+
+
+class NodeOperation(Operation):
+    async_operation = async_node_operation
+
+    def __init__(self, node):
+        super(NodeOperation, self).__init__(subject=node)
+        self.node = node
+
+    def create_activity(self, parent, user):
+        if parent:
+            if parent.node != self.node:
+                raise ValueError("The node associated with the specified "
+                                 "parent activity does not match the node "
+                                 "bound to the operation.")
+            if parent.user != user:
+                raise ValueError("The user associated with the specified "
+                                 "parent activity does not match the user "
+                                 "provided as parameter.")
+
+            return parent.create_sub(code_suffix=self.activity_code_suffix)
+        else:
+            return NodeActivity.create(code_suffix=self.activity_code_suffix,
+                                       node=self.node, user=user)
+
+
+def register_node_operation(op_cls, op_id=None):
+    return register_operation(Node, op_cls, op_id)
+
+
+class FlushOperation(NodeOperation):
+    activity_code_suffix = 'flush'
+    id = 'flush'
+    name = _("flush")
+    description = _("Disable node and move all instances to other ones.")
+
+    def _operation(self, activity, user, system):
+        self.node.disable(user, activity)
+        for i in self.node.instance_set.all():
+            with activity.sub_activity('migrate_instance_%d' % i.pk):
+                i.migrate()
+
+
+register_node_operation(FlushOperation)

circle/vm/tasks/local_periodic_tasks.py

@@ -27,7 +27,7 @@ def garbage_collector(timeout=15):
     now = timezone.now()
     for i in Instance.objects.filter(destroyed_at=None).all():
         if i.time_of_delete and now > i.time_of_delete:
-            i.destroy_async()
+            i.destroy.async()
             logger.info("Expired instance %d destroyed.", i.pk)
             try:
                 i.owner.profile.notify(
@@ -39,7 +39,7 @@ def garbage_collector(timeout=15):
                              i.pk, unicode(e))
         elif (i.time_of_suspend and now > i.time_of_suspend and
               i.state == 'RUNNING'):
-            i.sleep_async()
+            i.sleep.async()
             logger.info("Expired instance %d suspended." % i.pk)
             try:
                 i.owner.profile.notify(

circle/vm/tasks/local_tasks.py

 from manager.mancelery import celery
 
-# TODO: Keep synchronised with Instance funcs
-
-
-@celery.task
-def deploy(instance, user):
-    instance.deploy(task_uuid=deploy.request.id, user=user)
-
-
-@celery.task
-def redeploy(instance, user):
-    instance.redeploy(task_uuid=redeploy.request.id, user=user)
-
 
 @celery.task
-def shut_off(instance, user):
-    instance.shut_off(task_uuid=shut_off.request.id, user=user)
+def async_instance_operation(operation_id, instance_pk, activity_pk, **kwargs):
+    from vm.models import Instance, InstanceActivity
+    instance = Instance.objects.get(pk=instance_pk)
+    operation = getattr(instance, operation_id)
+    activity = InstanceActivity.objects.get(pk=activity_pk)
 
+    # save async task UUID to activity
+    activity.task_uuid = async_instance_operation.request.id
+    activity.save()
 
-@celery.task
-def destroy(instance, user):
-    instance.destroy(task_uuid=destroy.request.id, user=user)
-
-
-@celery.task
-def save_as_template(instance, name, user, params):
-    instance.save_as_template(name, task_uuid=save_as_template.request.id,
-                              user=user, **params)
-
-
-@celery.task
-def sleep(instance, user):
-    instance.sleep(task_uuid=sleep.request.id, user=user)
+    return operation._exec_op(activity=activity, **kwargs)
 
 
 @celery.task
-def wake_up(instance, user):
-    instance.wake_up(task_uuid=wake_up.request.id, user=user)
+def async_node_operation(operation_id, node_pk, activity_pk, **kwargs):
+    from vm.models import Node, NodeActivity
+    node = Node.objects.get(pk=node_pk)
+    operation = getattr(node, operation_id)
+    activity = NodeActivity.objects.get(pk=activity_pk)
 
+    # save async task UUID to activity
+    activity.task_uuid = async_node_operation.request.id
+    activity.save()
 
-@celery.task
-def shutdown(instance, user):
-    instance.shutdown(task_uuid=shutdown.request.id, user=user)
-
-
-@celery.task
-def reset(instance, user):
-    instance.reset(task_uuid=reset.request.id, user=user)
-
-
-@celery.task
-def reboot(instance, user):
-    instance.reboot(task_uuid=reboot.request.id, user=user)
-
-
-@celery.task
-def migrate(instance, to_node,  user):
-    instance.migrate(to_node, task_uuid=migrate.request.id, user=user)
-
-
-@celery.task
-def flush(node, user):
-    node.flush(task_uuid=flush.request.id, user=user)
+    return operation._exec_op(activity=activity, **kwargs)

circle/vm/tasks/vm_tasks.py

@@ -19,18 +19,16 @@ def check_queue(node_hostname, queue_id):
     active_queues = get_queues()
     if active_queues is None:
         return False
-    # v is List of List of queues dict
-    node_workers = [v for k, v in active_queues.iteritems()]
-    for worker in node_workers:
-        for queue in worker:
-            if queue['name'] == queue_name:
-                return True
-    return False
+    queue_names = (queue['name'] for worker in active_queues.itervalues()
+                   for queue in worker)
+    return queue_name in queue_names
 
 
 def get_queues():
     """Get active celery queues.
 
+    Returns a dictionary whose entries are (worker name;list of queues) pairs,
+    where queues are represented by dictionaries.
     Result is cached for 10 seconds!
     """
     key = __name__ + u'queues'

circle/vm/tests/test_models.py

 from datetime import datetime
+from mock import Mock, MagicMock, patch, call
 
 from django.contrib.auth.models import User
 from django.test import TestCase
 from django.utils.translation import ugettext_lazy as _
-from mock import Mock, MagicMock, patch, call
 
 from ..models import (
     Lease, Node, Interface, Instance, InstanceTemplate, InstanceActivity,
 )
 from ..models.instance import find_unused_port, ActivityInProgressError
+from ..operations import (
+    DeployOperation, DestroyOperation, FlushOperation, MigrateOperation,
+)
 
 
 class PortFinderTestCase(TestCase):
@@ -52,50 +55,60 @@ class InstanceTestCase(TestCase):
     def test_deploy_destroyed(self):
         inst = Mock(destroyed_at=datetime.now(), spec=Instance,
                     InstanceDestroyedError=Instance.InstanceDestroyedError)
-        with self.assertRaises(Instance.InstanceDestroyedError):
-            Instance.deploy(inst)
+        deploy_op = DeployOperation(inst)
+        with patch.object(DeployOperation, 'create_activity'):
+            with self.assertRaises(Instance.InstanceDestroyedError):
+                deploy_op(system=True)
 
     def test_destroy_destroyed(self):
-        inst = Mock(destroyed_at=datetime.now(), spec=Instance)
-        Instance.destroy(inst)
+        inst = Mock(destroyed_at=datetime.now(), spec=Instance,
+                    InstanceDestroyedError=Instance.InstanceDestroyedError)
+        destroy_op = DestroyOperation(inst)
+        with patch.object(DestroyOperation, 'create_activity'):
+            with self.assertRaises(Instance.InstanceDestroyedError):
+                destroy_op(system=True)
         self.assertFalse(inst.save.called)
 
     def test_destroy_sets_destroyed(self):
-        inst = MagicMock(destroyed_at=None, spec=Instance)
+        inst = Mock(destroyed_at=None, spec=Instance,
+                    InstanceDestroyedError=Instance.InstanceDestroyedError)
         inst.node = MagicMock(spec=Node)
         inst.disks.all.return_value = []
-        with patch('vm.models.instance.instance_activity') as ia:
-            ia.return_value = MagicMock()
-            Instance.destroy(inst)
+        destroy_op = DestroyOperation(inst)
+        with patch.object(DestroyOperation, 'create_activity'):
+            destroy_op(system=True)
         self.assertTrue(inst.destroyed_at)
         inst.save.assert_called()
 
     def test_migrate_with_scheduling(self):
-        inst = MagicMock(spec=Instance)
+        inst = Mock(destroyed_at=None, spec=Instance)
         inst.interface_set.all.return_value = []
         inst.node = MagicMock(spec=Node)
-        with patch('vm.models.instance.instance_activity') as ia, \
-                patch('vm.models.instance.vm_tasks.migrate') as migr:
-            Instance.migrate(inst)
+        migrate_op = MigrateOperation(inst)
+        with patch('vm.models.instance.vm_tasks.migrate') as migr:
+            act = MagicMock()
+            with patch.object(MigrateOperation, 'create_activity',
+                              return_value=act):
+                migrate_op(system=True)
 
             migr.apply_async.assert_called()
-            self.assertIn(call().__enter__().sub_activity(u'scheduling'),
-                          ia.mock_calls)
+            self.assertIn(call.sub_activity(u'scheduling'), act.mock_calls)
             inst.select_node.assert_called()
 
     def test_migrate_wo_scheduling(self):
-        inst = MagicMock(spec=Instance)
+        inst = MagicMock(destroyed_at=None, spec=Instance)
         inst.interface_set.all.return_value = []
         inst.node = MagicMock(spec=Node)
-        with patch('vm.models.instance.instance_activity') as ia, \
-                patch('vm.models.instance.vm_tasks.migrate') as migr:
+        migrate_op = MigrateOperation(inst)
+        with patch('vm.models.instance.vm_tasks.migrate') as migr:
             inst.select_node.side_effect = AssertionError
-
-            Instance.migrate(inst, inst.node)
+            act = MagicMock()
+            with patch.object(MigrateOperation, 'create_activity',
+                              return_value=act):
+                migrate_op(to_node=inst.node, system=True)
 
             migr.apply_async.assert_called()
-            self.assertNotIn(call().__enter__().sub_activity(u'scheduling'),
-                             ia.mock_calls)
+            self.assertNotIn(call.sub_activity(u'scheduling'), act.mock_calls)
 
     def test_status_icon(self):
         inst = MagicMock(spec=Instance)
@@ -162,25 +175,19 @@ class InstanceActivityTestCase(TestCase):
         instance.activity_log.filter.return_value.exists.return_value = True
 
         with self.assertRaises(ActivityInProgressError):
-            InstanceActivity.create("test", instance, concurrency_check=True)
+            InstanceActivity.create('test', instance, concurrency_check=True)
 
     def test_create_no_concurrency_check(self):
         instance = MagicMock(spec=Instance)
         instance.activity_log.filter.return_value.exists.return_value = True
 
-        original_method = InstanceActivity.create.__func__
-
-        with patch('vm.models.activity.InstanceActivity') as ia, \
-                patch('vm.models.activity.timezone.now'):
-            # ia.__init__ = MagicMock()  raises AttributeError
-
-            original_method(ia, "test", instance, concurrency_check=False)
-            ia.save.assert_called()
-
-            # ia.__init__.assert_called_with(activity_code='vm.Instance.test',
-            #                                instance=instance, parent=None,
-            #                                resultant_state=None, started=now,
-            #                                task_uuid=None, user=None)
+        with patch.object(InstanceActivity, '__new__'):
+            try:
+                InstanceActivity.create('test', instance,
+                                        concurrency_check=False)
+            except ActivityInProgressError:
+                raise AssertionError("'create' method checked for concurrent "
+                                     "activities.")
 
     def test_create_sub_concurrency_check(self):
         iaobj = MagicMock(spec=InstanceActivity)
@@ -194,12 +201,13 @@ class InstanceActivityTestCase(TestCase):
         iaobj.activity_code = 'test'
         iaobj.children.filter.return_value.exists.return_value = True
 
-        original_method = InstanceActivity.create_sub
-
-        with patch('vm.models.activity.InstanceActivity') as ia, \
-                patch('vm.models.activity.timezone.now'):
-            original_method(iaobj, "test", concurrency_check=False)
-            ia.save.assert_called()
+        with patch.object(InstanceActivity, '__new__'):
+            try:
+                InstanceActivity.create_sub(iaobj, 'test',
+                                            concurrency_check=False)
+            except ActivityInProgressError:
+                raise AssertionError("'create_sub' method checked for "
+                                     "concurrent activities.")
 
     def test_disable_enabled(self):
         node = MagicMock(spec=Node, enabled=True)
@@ -231,33 +239,37 @@ class InstanceActivityTestCase(TestCase):
         subact.__enter__.assert_called()
 
     def test_flush(self):
+        insts = [MagicMock(spec=Instance, migrate=MagicMock()),
+                 MagicMock(spec=Instance, migrate=MagicMock())]
         node = MagicMock(spec=Node, enabled=True)
+        node.instance_set.all.return_value = insts
         user = MagicMock(spec=User)
-        insts = [MagicMock(spec=Instance), MagicMock(spec=Instance)]
+        flush_op = FlushOperation(node)
 
-        with patch('vm.models.node.node_activity') as na:
-            act = na.return_value.__enter__.return_value = MagicMock()
-            node.instance_set.all.return_value = insts
+        with patch.object(FlushOperation, 'create_activity') as create_act:
+            act = create_act.return_value = MagicMock()
 
-            Node.flush(node, user)
+            flush_op(user=user)
 
-            na.__enter__.assert_called()
+            create_act.assert_called()
             node.disable.assert_called_with(user, act)
             for i in insts:
                 i.migrate.assert_called()
 
     def test_flush_disabled_wo_user(self):
+        insts = [MagicMock(spec=Instance, migrate=MagicMock()),
+                 MagicMock(spec=Instance, migrate=MagicMock())]
         node = MagicMock(spec=Node, enabled=False)
-        insts = [MagicMock(spec=Instance), MagicMock(spec=Instance)]
+        node.instance_set.all.return_value = insts
+        flush_op = FlushOperation(node)
 
-        with patch('vm.models.node.node_activity') as na:
-            act = na.return_value.__enter__.return_value = MagicMock()
-            node.instance_set.all.return_value = insts
+        with patch.object(FlushOperation, 'create_activity') as create_act:
+            act = create_act.return_value = MagicMock()
 
-            Node.flush(node)
+            flush_op(system=True)
 
+            create_act.assert_called()
             node.disable.assert_called_with(None, act)
             # ^ should be called, but real method no-ops if disabled
-            na.__enter__.assert_called()
             for i in insts:
                 i.migrate.assert_called()

circle/vm/tests/test_operations.py

+from django.test import TestCase
+
+from common.operations import operation_registry_name as op_reg_name
+from vm.models import Instance, Node
+from vm.operations import (
+    DeployOperation, DestroyOperation, FlushOperation, MigrateOperation,
+    RebootOperation, ResetOperation, SaveAsTemplateOperation,
+    ShutdownOperation, ShutOffOperation, SleepOperation, WakeUpOperation,
+)
+
+
+class DeployOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert DeployOperation.id in getattr(Instance, op_reg_name)
+
+
+class DestroyOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert DestroyOperation.id in getattr(Instance, op_reg_name)
+
+
+class FlushOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert FlushOperation.id in getattr(Node, op_reg_name)
+
+
+class MigrateOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert MigrateOperation.id in getattr(Instance, op_reg_name)
+
+
+class RebootOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert RebootOperation.id in getattr(Instance, op_reg_name)
+
+
+class ResetOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert ResetOperation.id in getattr(Instance, op_reg_name)
+
+
+class SaveAsTemplateOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert SaveAsTemplateOperation.id in getattr(Instance, op_reg_name)
+
+
+class ShutdownOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert ShutdownOperation.id in getattr(Instance, op_reg_name)
+
+
+class ShutOffOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert ShutOffOperation.id in getattr(Instance, op_reg_name)
+
+
+class SleepOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert SleepOperation.id in getattr(Instance, op_reg_name)
+
+
+class WakeUpOperationTestCase(TestCase):
+    def test_operation_registered(self):
+        assert WakeUpOperation.id in getattr(Instance, op_reg_name)