operations.py 59.4 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
# Copyright 2014 Budapest University of Technology and Economics (BME IK)
#
# This file is part of CIRCLE Cloud.
#
# CIRCLE is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option)
# any later version.
#
# CIRCLE is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along
# with CIRCLE.  If not, see <http://www.gnu.org/licenses/>.

18
from __future__ import absolute_import, unicode_literals
Bach Dániel committed
19 20
from base64 import encodestring
from hashlib import md5
Dudás Ádám committed
21
from logging import getLogger
Bach Dániel committed
22
import os
23
from re import search
Őry Máté committed
24
from string import ascii_lowercase
Bach Dániel committed
25 26 27
from StringIO import StringIO
from tarfile import TarFile, TarInfo
import time
Kálmán Viktor committed
28
from urlparse import urlsplit
29
from salt.client import LocalClient
Dudás Ádám committed
30

31
from django.core.exceptions import PermissionDenied, SuspiciousOperation
Dudás Ádám committed
32
from django.utils import timezone
33
from django.utils.translation import ugettext_lazy as _, ugettext_noop
Kálmán Viktor committed
34
from django.conf import settings
Bach Dániel committed
35
from django.db.models import Q
Dudás Ádám committed
36

37 38
from sizefield.utils import filesizeformat

39 40
from celery.contrib.abortable import AbortableAsyncResult
from celery.exceptions import TimeLimitExceeded, TimeoutError
41

42 43 44
from common.models import (
    create_readable, humanize_exception, HumanReadableException
)
45
from common.operations import Operation, register_operation, SubOperationMixin
Bach Dániel committed
46
from manager.scheduler import SchedulerError
47 48 49
from .tasks.local_tasks import (
    abortable_async_instance_operation, abortable_async_node_operation,
)
50
from .models import (
51
    Instance, InstanceActivity, InstanceTemplate, Interface, Node,
52
    NodeActivity, pwgen
53
)
Bach Dániel committed
54
from .tasks import agent_tasks, vm_tasks
Dudás Ádám committed
55

Kálmán Viktor committed
56
from dashboard.store_api import Store, NoStoreException
Bach Dániel committed
57 58
from firewall.models import Host
from monitor.client import Client
59
from storage.tasks import storage_tasks
Kálmán Viktor committed
60

Dudás Ádám committed
61
logger = getLogger(__name__)
62 63


64 65 66 67 68
class RemoteOperationMixin(object):

    remote_timeout = 30

    def _operation(self, **kwargs):
Őry Máté committed
69
        args = self._get_remote_args(**kwargs)
70 71 72 73 74 75 76 77 78
        return self.task.apply_async(
            args=args, queue=self._get_remote_queue()
        ).get(timeout=self.remote_timeout)

    def check_precond(self):
        super(RemoteOperationMixin, self).check_precond()
        self._get_remote_queue()


79 80 81 82 83 84 85
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())
86
        for i in xrange(0, self.remote_timeout, self.remote_step):
87 88 89 90 91 92 93
            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)
94
        raise TimeLimitExceeded()
95 96


97
class InstanceOperation(Operation):
98
    acl_level = 'owner'
99
    async_operation = abortable_async_instance_operation
100
    host_cls = Instance
101
    concurrency_check = True
102 103
    accept_states = None
    deny_states = None
104
    resultant_state = None
Dudás Ádám committed
105

106
    def __init__(self, instance):
107
        super(InstanceOperation, self).__init__(subject=instance)
108 109 110
        self.instance = instance

    def check_precond(self):
111 112
        if self.instance.destroyed_at:
            raise self.instance.InstanceDestroyedError(self.instance)
113 114 115 116 117 118 119 120 121 122 123 124 125 126
        if self.accept_states:
            if self.instance.status not in self.accept_states:
                logger.debug("precond failed for %s: %s not in %s",
                             unicode(self.__class__),
                             unicode(self.instance.status),
                             unicode(self.accept_states))
                raise self.instance.WrongStateError(self.instance)
        if self.deny_states:
            if self.instance.status in self.deny_states:
                logger.debug("precond failed for %s: %s in %s",
                             unicode(self.__class__),
                             unicode(self.instance.status),
                             unicode(self.accept_states))
                raise self.instance.WrongStateError(self.instance)
127 128

    def check_auth(self, user):
129
        if not self.instance.has_level(user, self.acl_level):
130 131 132
            raise humanize_exception(ugettext_noop(
                "%(acl_level)s level is required for this operation."),
                PermissionDenied(), acl_level=self.acl_level)
133

134
        super(InstanceOperation, self).check_auth(user=user)
135

136 137
    def create_activity(self, parent, user, kwargs):
        name = self.get_activity_name(kwargs)
138 139 140 141 142 143 144 145 146 147
        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.")

148 149 150
            return parent.create_sub(
                code_suffix=self.get_activity_code_suffix(),
                readable_name=name, resultant_state=self.resultant_state)
151 152
        else:
            return InstanceActivity.create(
153 154
                code_suffix=self.get_activity_code_suffix(),
                instance=self.instance,
155
                readable_name=name, user=user,
156 157
                concurrency_check=self.concurrency_check,
                resultant_state=self.resultant_state)
158

159 160 161 162 163
    def is_preferred(self):
        """If this is the recommended op in the current state of the instance.
        """
        return False

164

165 166 167 168 169 170 171 172 173 174 175
class RemoteInstanceOperation(RemoteOperationMixin, InstanceOperation):

    remote_queue = ('vm', 'fast')

    def _get_remote_queue(self):
        return self.instance.get_remote_queue_name(*self.remote_queue)

    def _get_remote_args(self, **kwargs):
        return [self.instance.vm_name]


Bach Dániel committed
176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199
class EnsureAgentMixin(object):
    accept_states = ('RUNNING', )

    def check_precond(self):
        super(EnsureAgentMixin, self).check_precond()

        last_boot_time = self.instance.activity_log.filter(
            succeeded=True, activity_code__in=(
                "vm.Instance.deploy", "vm.Instance.reset",
                "vm.Instance.reboot")).latest("finished").finished

        try:
            InstanceActivity.objects.filter(
                activity_code="vm.Instance.agent.starting",
                started__gt=last_boot_time).latest("started")
        except InstanceActivity.DoesNotExist:  # no agent since last boot
            raise self.instance.NoAgentError(self.instance)


class RemoteAgentOperation(EnsureAgentMixin, RemoteInstanceOperation):
    remote_queue = ('agent', )
    concurrency_check = False


200
@register_operation
201 202 203 204 205
class AddInterfaceOperation(InstanceOperation):
    id = 'add_interface'
    name = _("add interface")
    description = _("Add a new network interface for the specified VLAN to "
                    "the VM.")
206
    required_perms = ()
207
    accept_states = ('STOPPED', 'PENDING', 'RUNNING')
208

209 210 211 212 213 214 215
    def rollback(self, net, activity):
        with activity.sub_activity(
            'destroying_net',
                readable_name=ugettext_noop("destroy network (rollback)")):
            net.destroy()
            net.delete()

216
    def _operation(self, activity, user, system, vlan, managed=None):
217
        if not vlan.has_level(user, 'user'):
218 219 220
            raise humanize_exception(ugettext_noop(
                "User acces to vlan %(vlan)s is required."),
                PermissionDenied(), vlan=vlan)
221 222 223 224 225 226 227
        if managed is None:
            managed = vlan.managed

        net = Interface.create(base_activity=activity, instance=self.instance,
                               managed=managed, owner=user, vlan=vlan)

        if self.instance.is_running:
228
            try:
229 230
                self.instance._attach_network(
                    interface=net, parent_activity=activity)
231 232 233 234
            except Exception as e:
                if hasattr(e, 'libvirtError'):
                    self.rollback(net, activity)
                raise
235
            net.deploy()
Bach Dániel committed
236 237
            self.instance._change_ip(parent_activity=activity)
            self.instance._restart_networking(parent_activity=activity)
238

239 240 241 242
    def get_activity_name(self, kwargs):
        return create_readable(ugettext_noop("add %(vlan)s interface"),
                               vlan=kwargs['vlan'])

243

244
@register_operation
245
class CreateDiskOperation(InstanceOperation):
246

247 248
    id = 'create_disk'
    name = _("create disk")
249
    description = _("Create and attach empty disk to the virtual machine.")
250
    required_perms = ('storage.create_empty_disk', )
251
    accept_states = ('STOPPED', 'PENDING', 'RUNNING')
252

253
    def _operation(self, user, size, activity, name=None):
Bach Dániel committed
254 255
        from storage.models import Disk

256 257 258
        if not name:
            name = "new disk"
        disk = Disk.create(size=size, name=name, type="qcow2-norm")
259
        disk.full_clean()
260 261 262 263
        devnums = list(ascii_lowercase)
        for d in self.instance.disks.all():
            devnums.remove(d.dev_num)
        disk.dev_num = devnums.pop(0)
264
        disk.save()
265 266
        self.instance.disks.add(disk)

267
        if self.instance.is_running:
268 269 270 271
            with activity.sub_activity(
                'deploying_disk',
                readable_name=ugettext_noop("deploying disk")
            ):
272
                disk.deploy()
273
            self.instance._attach_disk(parent_activity=activity, disk=disk)
274

275
    def get_activity_name(self, kwargs):
276 277 278
        return create_readable(
            ugettext_noop("create disk %(name)s (%(size)s)"),
            size=filesizeformat(kwargs['size']), name=kwargs['name'])
279 280


281
@register_operation
282
class ResizeDiskOperation(RemoteInstanceOperation):
283 284 285 286 287

    id = 'resize_disk'
    name = _("resize disk")
    description = _("Resize the virtual disk image. "
                    "Size must be greater value than the actual size.")
288
    required_perms = ('storage.resize_disk', )
289 290
    accept_states = ('RUNNING', )
    async_queue = "localhost.man.slow"
291 292
    remote_queue = ('vm', 'slow')
    task = vm_tasks.resize_disk
293

294 295 296
    def _get_remote_args(self, disk, size, **kwargs):
        return (super(ResizeDiskOperation, self)
                ._get_remote_args(**kwargs) + [disk.path, size])
297 298 299 300 301 302

    def get_activity_name(self, kwargs):
        return create_readable(
            ugettext_noop("resize disk %(name)s to %(size)s"),
            size=filesizeformat(kwargs['size']), name=kwargs['disk'].name)

303 304 305 306 307
    def _operation(self, disk, size):
        super(ResizeDiskOperation, self)._operation(disk=disk, size=size)
        disk.size = size
        disk.save()

308

309
@register_operation
310 311 312
class DownloadDiskOperation(InstanceOperation):
    id = 'download_disk'
    name = _("download disk")
313 314 315 316
    description = _("Download and attach disk image (ISO file) for the "
                    "virtual machine. Most operating systems do not detect a "
                    "new optical drive, so you may have to reboot the "
                    "machine.")
317
    abortable = True
318
    has_percentage = True
319
    required_perms = ('storage.download_disk', )
320
    accept_states = ('STOPPED', 'PENDING', 'RUNNING')
321
    async_queue = "localhost.man.slow"
322

323
    def _operation(self, user, url, task, activity, name=None):
Bach Dániel committed
324 325
        from storage.models import Disk

326
        disk = Disk.download(url=url, name=name, task=task)
327 328 329 330
        devnums = list(ascii_lowercase)
        for d in self.instance.disks.all():
            devnums.remove(d.dev_num)
        disk.dev_num = devnums.pop(0)
331
        disk.full_clean()
332
        disk.save()
333
        self.instance.disks.add(disk)
334 335
        activity.readable_name = create_readable(
            ugettext_noop("download %(name)s"), name=disk.name)
336

337 338 339 340
        activity.result = create_readable(ugettext_noop(
            "Downloading %(url)s is finished. The file md5sum "
            "is: '%(checksum)s'."),
            url=url, checksum=disk.checksum)
Őry Máté committed
341
        # TODO iso (cd) hot-plug is not supported by kvm/guests
342
        if self.instance.is_running and disk.type not in ["iso"]:
343
            self.instance._attach_disk(parent_activity=activity, disk=disk)
344

345

346
@register_operation
347
class DeployOperation(InstanceOperation):
Dudás Ádám committed
348 349
    id = 'deploy'
    name = _("deploy")
350 351
    description = _("Deploy and start the virtual machine (including storage "
                    "and network configuration).")
352
    required_perms = ()
353
    deny_states = ('SUSPENDED', 'RUNNING')
354
    resultant_state = 'RUNNING'
Dudás Ádám committed
355

356 357
    def is_preferred(self):
        return self.instance.status in (self.instance.STATUS.STOPPED,
358
                                        self.instance.STATUS.PENDING,
359 360
                                        self.instance.STATUS.ERROR)

361 362 363
    def on_abort(self, activity, error):
        activity.resultant_state = 'STOPPED'

Dudás Ádám committed
364 365
    def on_commit(self, activity):
        activity.resultant_state = 'RUNNING'
366
        activity.result = create_readable(
Guba Sándor committed
367
            ugettext_noop("virtual machine successfully "
368 369
                          "deployed to node: %(node)s"),
            node=self.instance.node)
Dudás Ádám committed
370

371
    def _operation(self, activity, node=None):
Dudás Ádám committed
372 373
        # Allocate VNC port and host node
        self.instance.allocate_vnc_port()
374 375 376 377 378
        if node is not None:
            self.instance.node = node
            self.instance.save()
        else:
            self.instance.allocate_node()
Dudás Ádám committed
379 380

        # Deploy virtual images
381
        self.instance._deploy_disks(parent_activity=activity)
Dudás Ádám committed
382 383

        # Deploy VM on remote machine
384
        if self.instance.state not in ['PAUSED']:
385
            self.instance._deploy_vm(parent_activity=activity)
Dudás Ádám committed
386 387

        # Establish network connection (vmdriver)
388 389 390
        with activity.sub_activity(
            'deploying_net', readable_name=ugettext_noop(
                "deploy network")):
Dudás Ádám committed
391 392
            self.instance.deploy_net()

393 394 395 396 397
        try:
            self.instance.renew(parent_activity=activity)
        except:
            pass

398
        self.instance._resume_vm(parent_activity=activity)
Dudás Ádám committed
399

400 401 402
        if self.instance.has_agent:
            activity.sub_activity('os_boot', readable_name=ugettext_noop(
                "wait operating system loading"), interruptible=True)
Dudás Ádám committed
403

404
    @register_operation
Őry Máté committed
405
    class DeployVmOperation(SubOperationMixin, RemoteInstanceOperation):
406 407
        id = "_deploy_vm"
        name = _("deploy vm")
Őry Máté committed
408
        description = _("Deploy virtual machine.")
409 410 411
        remote_queue = ("vm", "slow")
        task = vm_tasks.deploy

Őry Máté committed
412
        def _get_remote_args(self, **kwargs):
413 414 415 416 417 418 419 420 421
            return [self.instance.get_vm_desc()]
            # intentionally not calling super

        def get_activity_name(self, kwargs):
            return create_readable(ugettext_noop("deploy virtual machine"),
                                   ugettext_noop("deploy vm to %(node)s"),
                                   node=self.instance.node)

    @register_operation
422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438
    class DeployDisksOperation(SubOperationMixin, InstanceOperation):
        id = "_deploy_disks"
        name = _("deploy disks")
        description = _("Deploy all associated disks.")

        def _operation(self):
            devnums = list(ascii_lowercase)  # a-z
            for disk in self.instance.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()

439
    @register_operation
Őry Máté committed
440
    class ResumeVmOperation(SubOperationMixin, RemoteInstanceOperation):
441 442 443 444 445
        id = "_resume_vm"
        name = _("boot virtual machine")
        remote_queue = ("vm", "slow")
        task = vm_tasks.resume

Dudás Ádám committed
446

447
@register_operation
448
class DestroyOperation(InstanceOperation):
Dudás Ádám committed
449 450
    id = 'destroy'
    name = _("destroy")
451 452
    description = _("Permanently destroy virtual machine, its network "
                    "settings and disks.")
453
    required_perms = ()
454
    resultant_state = 'DESTROYED'
Dudás Ádám committed
455

456
    def _operation(self, activity, system):
457
        # Destroy networks
458 459 460
        with activity.sub_activity(
                'destroying_net',
                readable_name=ugettext_noop("destroy network")):
461
            if self.instance.node:
462
                self.instance.shutdown_net()
463
            self.instance.destroy_net()
Dudás Ádám committed
464

465
        if self.instance.node:
466
            self.instance._delete_vm(parent_activity=activity)
Dudás Ádám committed
467 468

        # Destroy disks
469 470 471
        with activity.sub_activity(
                'destroying_disks',
                readable_name=ugettext_noop("destroy disks")):
Dudás Ádám committed
472
            self.instance.destroy_disks()
Dudás Ádám committed
473

Dudás Ádám committed
474 475
        # Delete mem. dump if exists
        try:
476
            self.instance._delete_mem_dump(parent_activity=activity)
Dudás Ádám committed
477 478 479 480 481 482
        except:
            pass

        # Clear node and VNC port association
        self.instance.yield_node()
        self.instance.yield_vnc_port()
Dudás Ádám committed
483 484 485 486

        self.instance.destroyed_at = timezone.now()
        self.instance.save()

487
    @register_operation
488
    class DeleteVmOperation(SubOperationMixin, RemoteInstanceOperation):
489 490 491 492 493
        id = "_delete_vm"
        name = _("destroy virtual machine")
        task = vm_tasks.destroy
        # if e.libvirtError and "Domain not found" in str(e):

494 495 496 497 498 499 500 501 502 503 504 505 506 507
    @register_operation
    class DeleteMemDumpOperation(RemoteOperationMixin, SubOperationMixin,
                                 InstanceOperation):
        id = "_delete_mem_dump"
        name = _("removing memory dump")
        task = storage_tasks.delete_dump

        def _get_remote_queue(self):
            return self.instance.mem_dump['datastore'].get_remote_queue_name(
                "storage", "fast")

        def _get_remote_args(self, **kwargs):
            return [self.instance.mem_dump['path']]

Dudás Ádám committed
508

509
@register_operation
510
class MigrateOperation(RemoteInstanceOperation):
Dudás Ádám committed
511 512
    id = 'migrate'
    name = _("migrate")
513 514
    description = _("Move a running virtual machine to an other worker node "
                    "keeping its full state.")
515
    required_perms = ()
516
    superuser_required = True
517
    accept_states = ('RUNNING', )
518
    async_queue = "localhost.man.slow"
519 520
    task = vm_tasks.migrate
    remote_queue = ("vm", "slow")
521
    remote_timeout = 1000
522

523
    def _get_remote_args(self, to_node, live_migration, **kwargs):
524
        return (super(MigrateOperation, self)._get_remote_args(**kwargs)
525
                + [to_node.host.hostname, live_migration])
Dudás Ádám committed
526

527
    def rollback(self, activity):
528 529 530
        with activity.sub_activity(
            'rollback_net', readable_name=ugettext_noop(
                "redeploy network (rollback)")):
531 532
            self.instance.deploy_net()

533
    def _operation(self, activity, to_node=None, live_migration=True):
Dudás Ádám committed
534
        if not to_node:
Bach Dániel committed
535 536 537 538 539 540
            with activity.sub_activity('scheduling',
                                       readable_name=ugettext_noop(
                                           "schedule")) as sa:
                to_node = self.instance.select_node()
                sa.result = to_node

541
        try:
542 543 544
            with activity.sub_activity(
                'migrate_vm', readable_name=create_readable(
                    ugettext_noop("migrate to %(node)s"), node=to_node)):
545 546
                super(MigrateOperation, self)._operation(
                    to_node=to_node, live_migration=live_migration)
547 548 549
        except Exception as e:
            if hasattr(e, 'libvirtError'):
                self.rollback(activity)
Bach Dániel committed
550
            raise
Dudás Ádám committed
551

552
        # Shutdown networks
553 554 555
        with activity.sub_activity(
            'shutdown_net', readable_name=ugettext_noop(
                "shutdown network")):
556 557
            self.instance.shutdown_net()

Dudás Ádám committed
558 559 560
        # Refresh node information
        self.instance.node = to_node
        self.instance.save()
561

Dudás Ádám committed
562
        # Estabilish network connection (vmdriver)
563 564 565
        with activity.sub_activity(
            'deploying_net', readable_name=ugettext_noop(
                "deploy network")):
Dudás Ádám committed
566
            self.instance.deploy_net()
Dudás Ádám committed
567 568


569
@register_operation
570
class RebootOperation(RemoteInstanceOperation):
Dudás Ádám committed
571 572
    id = 'reboot'
    name = _("reboot")
573 574
    description = _("Warm reboot virtual machine by sending Ctrl+Alt+Del "
                    "signal to its console.")
575
    required_perms = ()
576
    accept_states = ('RUNNING', )
577
    task = vm_tasks.reboot
Dudás Ádám committed
578

579 580
    def _operation(self, activity):
        super(RebootOperation, self)._operation()
581 582 583
        if self.instance.has_agent:
            activity.sub_activity('os_boot', readable_name=ugettext_noop(
                "wait operating system loading"), interruptible=True)
Dudás Ádám committed
584 585


586
@register_operation
587 588 589
class RemoveInterfaceOperation(InstanceOperation):
    id = 'remove_interface'
    name = _("remove interface")
590 591 592
    description = _("Remove the specified network interface and erase IP "
                    "address allocations, related firewall rules and "
                    "hostnames.")
593
    required_perms = ()
594
    accept_states = ('STOPPED', 'PENDING', 'RUNNING')
595

596 597
    def _operation(self, activity, user, system, interface):
        if self.instance.is_running:
598 599
            self.instance._detach_network(interface=interface,
                                          parent_activity=activity)
600 601 602 603 604
            interface.shutdown()

        interface.destroy()
        interface.delete()

605 606
    def get_activity_name(self, kwargs):
        return create_readable(ugettext_noop("remove %(vlan)s interface"),
607
                               vlan=kwargs['interface'].vlan)
608

609

610
@register_operation
611 612 613 614 615
class RemovePortOperation(InstanceOperation):
    id = 'remove_port'
    name = _("close port")
    description = _("Close the specified port.")
    concurrency_check = False
616
    required_perms = ('vm.config_ports', )
617 618 619 620

    def _operation(self, activity, rule):
        interface = rule.host.interface_set.get()
        if interface.instance != self.instance:
621
            raise SuspiciousOperation()
622 623 624 625 626 627 628
        activity.readable_name = create_readable(
            ugettext_noop("close %(proto)s/%(port)d on %(host)s"),
            proto=rule.proto, port=rule.dport, host=rule.host)
        rule.delete()


@register_operation
629 630 631 632 633
class AddPortOperation(InstanceOperation):
    id = 'add_port'
    name = _("open port")
    description = _("Open the specified port.")
    concurrency_check = False
634
    required_perms = ('vm.config_ports', )
635 636 637

    def _operation(self, activity, host, proto, port):
        if host.interface_set.get().instance != self.instance:
638
            raise SuspiciousOperation()
639 640 641 642 643 644 645
        host.add_port(proto, private=port)
        activity.readable_name = create_readable(
            ugettext_noop("open %(proto)s/%(port)d on %(host)s"),
            proto=proto, port=port, host=host)


@register_operation
646 647 648
class RemoveDiskOperation(InstanceOperation):
    id = 'remove_disk'
    name = _("remove disk")
649 650
    description = _("Remove the specified disk from the virtual machine, and "
                    "destroy the data.")
651
    required_perms = ()
652
    accept_states = ('STOPPED', 'PENDING', 'RUNNING')
653 654

    def _operation(self, activity, user, system, disk):
655
        if self.instance.is_running and disk.type not in ["iso"]:
656
            self.instance._detach_disk(disk=disk, parent_activity=activity)
657 658 659 660
        with activity.sub_activity(
            'destroy_disk',
            readable_name=ugettext_noop('destroy disk')
        ):
661
            disk.destroy()
662
            return self.instance.disks.remove(disk)
663

664 665 666
    def get_activity_name(self, kwargs):
        return create_readable(ugettext_noop('remove disk %(name)s'),
                               name=kwargs["disk"].name)
667 668


669
@register_operation
670
class ResetOperation(RemoteInstanceOperation):
Dudás Ádám committed
671 672
    id = 'reset'
    name = _("reset")
673
    description = _("Cold reboot virtual machine (power cycle).")
674
    required_perms = ()
675
    accept_states = ('RUNNING', )
676
    task = vm_tasks.reset
Dudás Ádám committed
677

678 679
    def _operation(self, activity):
        super(ResetOperation, self)._operation()
680 681 682
        if self.instance.has_agent:
            activity.sub_activity('os_boot', readable_name=ugettext_noop(
                "wait operating system loading"), interruptible=True)
Dudás Ádám committed
683 684


685
@register_operation
686
class SaveAsTemplateOperation(InstanceOperation):
Dudás Ádám committed
687 688
    id = 'save_as_template'
    name = _("save as template")
689 690 691 692
    description = _("Save virtual machine as a template so they can be shared "
                    "with users and groups.  Anyone who has access to a "
                    "template (and to the networks it uses) will be able to "
                    "start an instance of it.")
693
    has_percentage = True
694
    abortable = True
695
    required_perms = ('vm.create_template', )
696
    accept_states = ('RUNNING', 'STOPPED')
697
    async_queue = "localhost.man.slow"
Dudás Ádám committed
698

699 700 701 702
    def is_preferred(self):
        return (self.instance.is_base and
                self.instance.status == self.instance.STATUS.RUNNING)

703 704 705 706 707 708
    @staticmethod
    def _rename(name):
        m = search(r" v(\d+)$", name)
        if m:
            v = int(m.group(1)) + 1
            name = search(r"^(.*) v(\d+)$", name).group(1)
709
        else:
710 711
            v = 1
        return "%s v%d" % (name, v)
712

713
    def on_abort(self, activity, error):
714
        if hasattr(self, 'disks'):
715 716 717
            for disk in self.disks:
                disk.destroy()

718
    def _operation(self, activity, user, system, name=None,
719
                   with_shutdown=True, clone=False, task=None, **kwargs):
720 721
        try:
            self.instance._cleanup(parent_activity=activity, user=user)
722
        except:
723 724
            pass

725
        if with_shutdown:
726
            try:
727 728
                self.instance.shutdown(parent_activity=activity,
                                       user=user, task=task)
729 730 731
            except Instance.WrongStateError:
                pass

Dudás Ádám committed
732 733 734 735 736 737 738 739
        # 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,
740
            'name': name or self._rename(self.instance.name),
Dudás Ádám committed
741 742
            'num_cores': self.instance.num_cores,
            'owner': user,
743
            'parent': self.instance.template or None,  # Can be problem
Dudás Ádám committed
744 745 746 747 748 749
            'priority': self.instance.priority,
            'ram_size': self.instance.ram_size,
            'raw_data': self.instance.raw_data,
            'system': self.instance.system,
        }
        params.update(kwargs)
Bach Dániel committed
750
        params.pop("parent_activity", None)
Dudás Ádám committed
751

752 753
        from storage.models import Disk

Dudás Ádám committed
754 755
        def __try_save_disk(disk):
            try:
756
                return disk.save_as(task)
Dudás Ádám committed
757 758 759
            except Disk.WrongDiskTypeError:
                return disk

760
        self.disks = []
761 762 763 764 765 766 767
        for disk in self.instance.disks.all():
            with activity.sub_activity(
                'saving_disk',
                readable_name=create_readable(
                    ugettext_noop("saving disk %(name)s"),
                    name=disk.name)
            ):
768 769
                self.disks.append(__try_save_disk(disk))

Dudás Ádám committed
770 771 772 773
        # create template and do additional setup
        tmpl = InstanceTemplate(**params)
        tmpl.full_clean()  # Avoiding database errors.
        tmpl.save()
774
        # Copy traits from the VM instance
775
        tmpl.req_traits.add(*self.instance.req_traits.all())
776 777
        if clone:
            tmpl.clone_acl(self.instance.template)
Guba Sándor committed
778
            # Add permission for the original owner of the template
779 780
            tmpl.set_level(self.instance.template.owner, 'owner')
            tmpl.set_level(user, 'owner')
Dudás Ádám committed
781
        try:
782
            tmpl.disks.add(*self.disks)
Dudás Ádám committed
783 784 785 786 787 788 789 790 791 792
            # create interface templates
            for i in self.instance.interface_set.all():
                i.save_as_template(tmpl)
        except:
            tmpl.delete()
            raise
        else:
            return tmpl


793
@register_operation
794 795
class ShutdownOperation(AbortableRemoteOperationMixin,
                        RemoteInstanceOperation):
Dudás Ádám committed
796 797
    id = 'shutdown'
    name = _("shutdown")
798 799 800 801
    description = _("Try to halt virtual machine by a standard ACPI signal, "
                    "allowing the operating system to keep a consistent "
                    "state. The operation will fail if the machine does not "
                    "turn itself off in a period.")
Kálmán Viktor committed
802
    abortable = True
803
    required_perms = ()
804
    accept_states = ('RUNNING', )
805
    resultant_state = 'STOPPED'
806 807
    task = vm_tasks.shutdown
    remote_queue = ("vm", "slow")
808
    remote_timeout = 180
Dudás Ádám committed
809

810 811
    def _operation(self, task):
        super(ShutdownOperation, self)._operation(task=task)
Dudás Ádám committed
812
        self.instance.yield_node()
Dudás Ádám committed
813

814 815 816 817 818 819 820 821 822 823 824
    def on_abort(self, activity, error):
        if isinstance(error, TimeLimitExceeded):
            activity.result = humanize_exception(ugettext_noop(
                "The virtual machine did not switch off in the provided time "
                "limit. Most of the time this is caused by incorrect ACPI "
                "settings. You can also try to power off the machine from the "
                "operating system manually."), error)
            activity.resultant_state = None
        else:
            super(ShutdownOperation, self).on_abort(activity, error)

Dudás Ádám committed
825

826
@register_operation
827
class ShutOffOperation(InstanceOperation):
Dudás Ádám committed
828 829
    id = 'shut_off'
    name = _("shut off")
830 831 832 833 834 835 836
    description = _("Forcibly halt a virtual machine without notifying the "
                    "operating system. This operation will even work in cases "
                    "when shutdown does not, but the operating system and the "
                    "file systems are likely to be in an inconsistent state,  "
                    "so data loss is also possible. The effect of this "
                    "operation is the same as interrupting the power supply "
                    "of a physical machine.")
837
    required_perms = ()
838
    accept_states = ('RUNNING', 'PAUSED')
839
    resultant_state = 'STOPPED'
Dudás Ádám committed
840

841
    def _operation(self, activity):
Dudás Ádám committed
842 843 844
        # Shutdown networks
        with activity.sub_activity('shutdown_net'):
            self.instance.shutdown_net()
Dudás Ádám committed
845

846
        self.instance._delete_vm(parent_activity=activity)
Dudás Ádám committed
847
        self.instance.yield_node()
Dudás Ádám committed
848 849


850
@register_operation
851
class SleepOperation(InstanceOperation):
Dudás Ádám committed
852 853
    id = 'sleep'
    name = _("sleep")
854 855 856 857 858 859 860 861
    description = _("Suspend virtual machine. This means the machine is "
                    "stopped and its memory is saved to disk, so if the "
                    "machine is waked up, all the applications will keep "
                    "running. Most of the applications will be able to "
                    "continue even after a long suspension, but those which "
                    "need a continous network connection may fail when "
                    "resumed. In the meantime, the machine will only use "
                    "storage resources, and keep network resources allocated.")
862
    required_perms = ()
863
    accept_states = ('RUNNING', )
864
    resultant_state = 'SUSPENDED'
865
    async_queue = "localhost.man.slow"
Dudás Ádám committed
866

867 868 869 870
    def is_preferred(self):
        return (not self.instance.is_base and
                self.instance.status == self.instance.STATUS.RUNNING)

Dudás Ádám committed
871 872 873 874 875 876
    def on_abort(self, activity, error):
        if isinstance(error, TimeLimitExceeded):
            activity.resultant_state = None
        else:
            activity.resultant_state = 'ERROR'

877
    def _operation(self, activity, system):
878 879 880
        with activity.sub_activity('shutdown_net',
                                   readable_name=ugettext_noop(
                                       "shutdown network")):
Dudás Ádám committed
881
            self.instance.shutdown_net()
882
        self.instance._suspend_vm(parent_activity=activity)
883
        self.instance.yield_node()
Dudás Ádám committed
884

885 886 887 888
    @register_operation
    class SuspendVmOperation(SubOperationMixin, RemoteInstanceOperation):
        id = "_suspend_vm"
        name = _("suspend virtual machine")
889
        task = vm_tasks.sleep
890
        remote_queue = ("vm", "slow")
891
        remote_timeout = 1000
Dudás Ádám committed
892

893 894 895 896
        def _get_remote_args(self, **kwargs):
            return (super(SleepOperation.SuspendVmOperation, self)
                    ._get_remote_args(**kwargs)
                    + [self.instance.mem_dump['path']])
Dudás Ádám committed
897 898


899
@register_operation
900
class WakeUpOperation(InstanceOperation):
Dudás Ádám committed
901 902
    id = 'wake_up'
    name = _("wake up")
903 904 905
    description = _("Wake up sleeping (suspended) virtual machine. This will "
                    "load the saved memory of the system and start the "
                    "virtual machine from this state.")
906
    required_perms = ()
907
    accept_states = ('SUSPENDED', )
908
    resultant_state = 'RUNNING'
909
    async_queue = "localhost.man.slow"
Dudás Ádám committed
910

911
    def is_preferred(self):
912
        return self.instance.status == self.instance.STATUS.SUSPENDED
913

Dudás Ádám committed
914
    def on_abort(self, activity, error):
Bach Dániel committed
915 916 917 918
        if isinstance(error, SchedulerError):
            activity.resultant_state = None
        else:
            activity.resultant_state = 'ERROR'
Dudás Ádám committed
919

920
    def _operation(self, activity):
Dudás Ádám committed
921
        # Schedule vm
Dudás Ádám committed
922
        self.instance.allocate_vnc_port()
923
        self.instance.allocate_node()
Dudás Ádám committed
924 925

        # Resume vm
926
        self.instance._wake_up_vm(parent_activity=activity)
Dudás Ádám committed
927 928

        # Estabilish network connection (vmdriver)
929 930 931
        with activity.sub_activity(
            'deploying_net', readable_name=ugettext_noop(
                "deploy network")):
Dudás Ádám committed
932
            self.instance.deploy_net()
Dudás Ádám committed
933

934 935 936 937
        try:
            self.instance.renew(parent_activity=activity)
        except:
            pass
Dudás Ádám committed
938

939 940 941 942 943 944
    @register_operation
    class WakeUpVmOperation(SubOperationMixin, RemoteInstanceOperation):
        id = "_wake_up_vm"
        name = _("resume virtual machine")
        task = vm_tasks.wake_up
        remote_queue = ("vm", "slow")
945
        remote_timeout = 1000
946 947 948 949 950 951

        def _get_remote_args(self, **kwargs):
            return (super(WakeUpOperation.WakeUpVmOperation, self)
                    ._get_remote_args(**kwargs)
                    + [self.instance.mem_dump['path']])

Dudás Ádám committed
952

953
@register_operation
954 955 956
class RenewOperation(InstanceOperation):
    id = 'renew'
    name = _("renew")
957 958 959 960
    description = _("Virtual machines are suspended and destroyed after they "
                    "expire. This operation renews expiration times according "
                    "to the lease type. If the machine is close to the "
                    "expiration, its owner will be notified.")
961
    acl_level = "operator"
962
    required_perms = ()
963
    concurrency_check = False
964

Őry Máté committed
965
    def _operation(self, activity, lease=None, force=False, save=False):
966 967 968 969 970 971 972 973 974 975 976 977 978
        suspend, delete = self.instance.get_renew_times(lease)
        if (not force and suspend and self.instance.time_of_suspend and
                suspend < self.instance.time_of_suspend):
            raise HumanReadableException.create(ugettext_noop(
                "Renewing the machine with the selected lease would result "
                "in its suspension time get earlier than before."))
        if (not force and delete and self.instance.time_of_delete and
                delete < self.instance.time_of_delete):
            raise HumanReadableException.create(ugettext_noop(
                "Renewing the machine with the selected lease would result "
                "in its delete time get earlier than before."))
        self.instance.time_of_suspend = suspend
        self.instance.time_of_delete = delete
Őry Máté committed
979 980
        if save:
            self.instance.lease = lease
981
        self.instance.save()
982 983 984
        activity.result = create_readable(ugettext_noop(
            "Renewed to suspend at %(suspend)s and destroy at %(delete)s."),
            suspend=suspend, delete=delete)
985 986


987
@register_operation
988
class ChangeStateOperation(InstanceOperation):
Guba Sándor committed
989
    id = 'emergency_change_state'
990 991 992 993 994 995
    name = _("emergency state change")
    description = _("Change the virtual machine state to NOSTATE. This "
                    "should only be used if manual intervention was needed in "
                    "the virtualization layer, and the machine has to be "
                    "redeployed without losing its storage and network "
                    "resources.")
996
    acl_level = "owner"
Guba Sándor committed
997
    required_perms = ('vm.emergency_change_state', )
998
    concurrency_check = False
999

1000 1001
    def _operation(self, user, activity, new_state="NOSTATE", interrupt=False,
                   reset_node=False):
1002
        activity.resultant_state = new_state
1003 1004 1005 1006 1007 1008 1009
        if interrupt:
            msg_txt = ugettext_noop("Activity is forcibly interrupted.")
            message = create_readable(msg_txt, msg_txt)
            for i in InstanceActivity.objects.filter(
                    finished__isnull=True, instance=self.instance):
                i.finish(False, result=message)
                logger.error('Forced finishing activity %s', i)
1010

1011 1012 1013 1014
        if reset_node:
            self.instance.node = None
            self.instance.save()

1015

1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040
@register_operation
class RedeployOperation(InstanceOperation):
    id = 'redeploy'
    name = _("redeploy")
    description = _("Change the virtual machine state to NOSTATE "
                    "and redeploy the VM. This operation allows starting "
                    "machines formerly running on a failed node.")
    acl_level = "owner"
    required_perms = ('vm.redeploy', )
    concurrency_check = False

    def _operation(self, user, activity, with_emergency_change_state=True):
        if with_emergency_change_state:
            ChangeStateOperation(self.instance).call(
                parent_activity=activity, user=user,
                new_state='NOSTATE', interrupt=False, reset_node=True)
        else:
            ShutOffOperation(self.instance).call(
                parent_activity=activity, user=user)

        self.instance._update_status()

        DeployOperation(self.instance).call(
            parent_activity=activity, user=user)

1041

1042
class NodeOperation(Operation):
1043
    async_operation = abortable_async_node_operation
1044
    host_cls = Node
1045 1046
    online_required = True
    superuser_required = True
1047 1048 1049 1050 1051

    def __init__(self, node):
        super(NodeOperation, self).__init__(subject=node)
        self.node = node

1052 1053 1054 1055 1056 1057 1058
    def check_precond(self):
        super(NodeOperation, self).check_precond()
        if self.online_required and not self.node.online:
            raise humanize_exception(ugettext_noop(
                "You cannot call this operation on an offline node."),
                Exception())

1059 1060
    def create_activity(self, parent, user, kwargs):
        name = self.get_activity_name(kwargs)
1061 1062 1063 1064 1065 1066 1067 1068 1069 1070
        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.")

1071 1072 1073
            return parent.create_sub(
                code_suffix=self.get_activity_code_suffix(),
                readable_name=name)
1074
        else:
1075 1076 1077
            return NodeActivity.create(
                code_suffix=self.get_activity_code_suffix(), node=self.node,
                user=user, readable_name=name)
1078 1079


1080
@register_operation
1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108
class ResetNodeOperation(NodeOperation):
    id = 'reset'
    name = _("reset")
    description = _("Disable missing node and redeploy all instances "
                    "on other ones.")
    required_perms = ()
    online_required = False
    async_queue = "localhost.man.slow"

    def check_precond(self):
        super(ResetNodeOperation, self).check_precond()
        if not self.node.enabled or self.node.online:
            raise humanize_exception(ugettext_noop(
                "You cannot reset a disabled or online node."), Exception())

    def _operation(self, activity, user):
        if self.node.enabled:
            DisableOperation(self.node).call(parent_activity=activity,
                                             user=user)
        for i in self.node.instance_set.all():
            name = create_readable(ugettext_noop(
                "migrate %(instance)s (%(pk)s)"), instance=i.name, pk=i.pk)
            with activity.sub_activity('migrate_instance_%d' % i.pk,
                                       readable_name=name):
                i.redeploy(user=user)


@register_operation
1109 1110 1111
class FlushOperation(NodeOperation):
    id = 'flush'
    name = _("flush")
1112
    description = _("Passivate node and move all instances to other ones.")
1113
    required_perms = ()
1114
    async_queue = "localhost.man.slow"
1115

1116
    def _operation(self, activity, user):
1117 1118 1119
        if self.node.schedule_enabled:
            PassivateOperation(self.node).call(parent_activity=activity,
                                               user=user)
1120
        for i in self.node.instance_set.all():
1121 1122 1123 1124
            name = create_readable(ugettext_noop(
                "migrate %(instance)s (%(pk)s)"), instance=i.name, pk=i.pk)
            with activity.sub_activity('migrate_instance_%d' % i.pk,
                                       readable_name=name):
Bach Dániel committed
1125
                i.migrate(user=user)