models.py 16.2 KB
Newer Older
1 2
# coding=utf-8

3
from contextlib import contextmanager
4
import logging
5 6
import uuid

7
from django.db.models import (Model, BooleanField, CharField, DateTimeField,
8
                              ForeignKey)
9
from django.utils import timezone
10
from django.utils.translation import ugettext_lazy as _
11
from model_utils.models import TimeStampedModel
12
from sizefield.models import FileSizeField
13
from datetime import timedelta
14

15
from acl.models import AclBase
16
from .tasks import local_tasks, remote_tasks
17
from celery.exceptions import TimeoutError
18
from common.models import ActivityModel, activitycontextimpl, WorkerNotFound
19 20 21 22

logger = logging.getLogger(__name__)


23
class DataStore(Model):
Guba Sándor committed
24

25 26
    """Collection of virtual disks.
    """
27 28 29 30
    name = CharField(max_length=100, unique=True, verbose_name=_('name'))
    path = CharField(max_length=200, unique=True, verbose_name=_('path'))
    hostname = CharField(max_length=40, unique=True,
                         verbose_name=_('hostname'))
Guba Sándor committed
31

32 33 34 35 36 37 38 39
    class Meta:
        ordering = ['name']
        verbose_name = _('datastore')
        verbose_name_plural = _('datastores')

    def __unicode__(self):
        return u'%s (%s)' % (self.name, self.path)

40
    def get_remote_queue_name(self, queue_id, check_worker=True):
41 42
        logger.debug("Checking for storage queue %s.%s",
                     self.hostname, queue_id)
43 44
        if not check_worker or local_tasks.check_queue(self.hostname,
                                                       queue_id):
45 46 47
            return self.hostname + '.' + queue_id
        else:
            raise WorkerNotFound()
48

49 50 51 52 53
    def get_deletable_disks(self):
        return [disk.filename for disk in
                self.disk_set.filter(
                    destroyed__isnull=False) if disk.is_deletable()]

54

55
class Disk(AclBase, TimeStampedModel):
Guba Sándor committed
56

57 58
    """A virtual disk.
    """
59 60 61 62 63
    ACL_LEVELS = (
        ('user', _('user')),          # see all details
        ('operator', _('operator')),
        ('owner', _('owner')),        # superuser, can delete, delegate perms
    )
64 65
    TYPES = [('qcow2-norm', 'qcow2 normal'), ('qcow2-snap', 'qcow2 snapshot'),
             ('iso', 'iso'), ('raw-ro', 'raw read-only'), ('raw-rw', 'raw')]
66 67 68 69 70
    name = CharField(blank=True, max_length=100, verbose_name=_("name"))
    filename = CharField(max_length=256, verbose_name=_("filename"))
    datastore = ForeignKey(DataStore, verbose_name=_("datastore"),
                           help_text=_("The datastore that holds the disk."))
    type = CharField(max_length=10, choices=TYPES)
71
    size = FileSizeField()
72 73
    base = ForeignKey('self', blank=True, null=True,
                      related_name='derivatives')
74 75
    ready = BooleanField(default=False,
                         help_text=_("The associated resource is ready."))
76
    dev_num = CharField(default='a', max_length=1,
77
                        verbose_name=_("device number"))
78
    destroyed = DateTimeField(blank=True, default=None, null=True)
79 80 81 82 83 84

    class Meta:
        ordering = ['name']
        verbose_name = _('disk')
        verbose_name_plural = _('disks')

85 86
    class WrongDiskTypeError(Exception):

87 88 89 90 91 92 93 94
        def __init__(self, type, message=None):
            if message is None:
                message = ("Operation can't be invoked on a disk of type '%s'."
                           % type)

            Exception.__init__(self, message)

            self.type = type
95

96 97
    class DiskInUseError(Exception):

98 99 100 101
        def __init__(self, disk, message=None):
            if message is None:
                message = ("The requested operation can't be performed on "
                           "disk '%s (%s)' because it is in use." %
Dudás Ádám committed
102
                           (disk.name, disk.filename))
103 104 105 106

            Exception.__init__(self, message)

            self.disk = disk
107

108 109
    @property
    def path(self):
110
        """Get the path where the files are stored."""
111 112 113 114
        return self.datastore.path + '/' + self.filename

    @property
    def format(self):
115
        """Returns the proper file format for different type of images."""
116 117 118
        return {
            'qcow2-norm': 'qcow2',
            'qcow2-snap': 'qcow2',
119
            'iso': 'raw',
120 121 122 123
            'raw-ro': 'raw',
            'raw-rw': 'raw',
        }[self.type]

124 125
    @property
    def device_type(self):
126
        """Returns the proper device prefix for different file format."""
127
        return {
128 129
            'qcow2-norm': 'vd',
            'qcow2-snap': 'vd',
130
            'iso': 'hd',
131 132 133
            'raw-ro': 'vd',
            'raw-rw': 'vd',
        }[self.type]
134

135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151
    def is_deletable(self):
        """Returns True if no child and disk is destroyed."""
        time_before = timezone.now() - timedelta(days=1)
        if self.destroyed > time_before or self.has_active_child():
            return False
        else:
            return True

    def has_active_child(self):
        """Returns True if disk have iactive childs."""
        time_before = timezone.now() - timedelta(days=1)
        for child in self.derivatives.all():
            if child.destroyed > time_before or None:
                return True
        else:
            return False

152
    def is_in_use(self):
153
        """Returns True if disc is attached to an active VM else False"""
154
        return any([i.state != 'STOPPED' for i in self.instance_set.all()])
155

156 157
    def get_exclusive(self):
        """Get an instance of the disk for exclusive usage.
158

159 160 161
        This method manipulates the database only.
        """
        type_mapping = {
162 163 164
            'qcow2-norm': 'qcow2-snap',
            'iso': 'iso',
            'raw-ro': 'raw-rw',
165 166 167 168 169 170 171
        }

        if self.type not in type_mapping.keys():
            raise self.WrongDiskTypeError(self.type)

        filename = self.filename if self.type == 'iso' else str(uuid.uuid4())
        new_type = type_mapping[self.type]
172

173 174 175
        return Disk.objects.create(base=self, datastore=self.datastore,
                                   filename=filename, name=self.name,
                                   size=self.size, type=new_type)
176 177

    def get_vmdisk_desc(self):
178
        """Serialize disk object to the vmdriver."""
179
        return {
180
            'source': self.path,
181 182
            'driver_type': self.format,
            'driver_cache': 'default',
183
            'target_device': self.device_type + self.dev_num,
184
            'disk_device': 'cdrom' if self.type == 'iso' else 'disk'
185 186
        }

187
    def get_disk_desc(self):
188
        """Serialize disk object to the storage driver."""
189 190 191 192 193 194 195 196 197
        return {
            'name': self.filename,
            'dir': self.datastore.path,
            'format': self.format,
            'size': self.size,
            'base_name': self.base.filename if self.base else None,
            'type': 'snapshot' if self.type == 'qcow2-snap' else 'normal'
        }

198
    def get_remote_queue_name(self, queue_id):
199
        """Returns the proper queue name based on the datastore."""
200 201 202 203 204
        if self.datastore:
            return self.datastore.get_remote_queue_name(queue_id)
        else:
            return None

205 206 207
    def __unicode__(self):
        return u"%s (#%d)" % (self.name, self.id)

208 209 210 211 212
    def clean(self, *args, **kwargs):
        if self.size == "" and self.base:
            self.size = self.base.size
        super(Disk, self).clean(*args, **kwargs)

213
    def deploy(self, user=None, task_uuid=None):
214 215 216 217 218
        """Reify the disk model on the associated data store.

        :param self: the disk model to reify
        :type self: storage.models.Disk

219 220 221 222 223 224 225
        :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

226 227 228 229
        :return: True if a new reification of the disk has been created;
                 otherwise, False.
        :rtype: bool
        """
230 231 232 233
        if self.destroyed:
            self.destroyed = None
            self.save()

234
        if self.ready:
235
            return False
236

237 238 239 240
        with disk_activity(code_suffix='deploy', disk=self,
                           task_uuid=task_uuid, user=user) as act:

            # Delegate create / snapshot jobs
241
            queue_name = self.get_remote_queue_name('storage')
242 243 244 245 246 247 248 249 250 251 252 253
            disk_desc = self.get_disk_desc()
            if self.type == 'qcow2-snap':
                with act.sub_activity('creating_snapshot'):
                    remote_tasks.snapshot.apply_async(args=[disk_desc],
                                                      queue=queue_name).get()
            else:
                with act.sub_activity('creating_disk'):
                    remote_tasks.create.apply_async(args=[disk_desc],
                                                    queue=queue_name).get()

            self.ready = True
            self.save()
254

255
            return True
256

257
    def deploy_async(self, user=None):
258 259
        """Execute deploy asynchronously.
        """
260 261
        return local_tasks.deploy.apply_async(args=[self, user],
                                              queue="localhost.man")
262

263
    @classmethod
264 265 266 267 268 269 270 271 272 273 274 275
    def create_empyt_async(instance=None, params=None, user=None):
        """Create empty Disk object asynchronusly.

        :param instance: instnace object to connect disk
        :type instane: vm.models.Instance
        :param params: disk custom parameters
        :type params: dict
        :param user: owner of the disk
        :type user: django.contrib.auth.User
        """
        return local_tasks.create_empty.apply_async(
            args=[instance, params, user], queue="localhost.man")
276 277

    @classmethod
278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299
    def create_empty(cls, instance=None, params=None,
                     user=None, task_uuid=None):
        """Create empty Disk object.

        :param instance: instnace object to connect disk
        :type instane: vm.models.Instance
        :param params: disk custom parameters
        :type params: dict
        :param user: owner of the disk
        :type user: django.contrib.auth.User
        """
        with disk_activity(code_suffix="create", task_uuid=task_uuid,
                           user=user) as act:
            disk = cls()
            if params:
                disk.__dict__.update(params)
            disk.save()
            act.disk = disk
            act.save()
            if instance:
                instance.disks.add(disk)
            return disk
300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316

    @classmethod
    def create_from_url_async(cls, url, instance=None, params=None, user=None):
        """Create disk object and download data from url asynchronusly.

        :param url: image url to download.
        :type url: url
        :param instance: instnace object to connect disk
        :type instane: vm.models.Instance
        :param params: disk custom parameters
        :type params: dict
        :param user: owner of the disk
        :type user: django.contrib.auth.User

        :return: Task
        :rtype: AsyncResult
        """
317
        return local_tasks.create_from_url.apply_async(kwargs={
318 319
            'cls': cls, 'url': url, 'instance': instance,
            'params': params, 'user': user},
320
            queue='localhost.man')
321

322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338
    @classmethod
    def create_from_url(cls, url, instance=None, params=None, user=None,
                        task_uuid=None, abortable_task=None):
        """Create disk object and download data from url synchronusly.

        :param url: image url to download.
        :type url: url
        :param instance: instnace object to connect disk
        :type instane: vm.models.Instance
        :param params: disk custom parameters
        :type params: dict
        :param user: owner of the disk
        :type user: django.contrib.auth.User

        :return: Task
        :rtype: AsyncResult
        """
339 340 341 342
        disk = cls()
        disk.filename = str(uuid.uuid4())
        disk.type = "iso"
        disk.size = 1
343
        # TODO get proper datastore
344
        disk.datastore = DataStore.objects.all()[0]
345 346
        if params:
            disk.__dict__.update(params)
347
        disk.save()
348 349
        if instance:
            instance.disks.add(disk)
350 351
        queue_name = disk.get_remote_queue_name('storage')

352 353 354 355 356 357 358 359 360
        def __on_abort(activity, error):
            activity.disk.destroyed = timezone.now()
            activity.disk.save()

        if abortable_task:
            from celery.contrib.abortable import AbortableAsyncResult

            class AbortException(Exception):
                pass
361 362

        with disk_activity(code_suffix='download', disk=disk,
363 364 365 366 367 368 369 370 371 372 373 374 375 376
                           task_uuid=task_uuid, user=user,
                           on_abort=__on_abort):
            result = remote_tasks.download.apply_async(
                kwargs={'url': url, 'parent_id': task_uuid,
                        'disk': disk.get_disk_desc()},
                queue=queue_name)
            while True:
                try:
                    size = result.get(timeout=5)
                    break
                except TimeoutError:
                    if abortable_task and abortable_task.is_aborted():
                        AbortableAsyncResult(result.id).abort()
                        raise AbortException("Download aborted by user.")
377
            disk.size = size
378
            disk.ready = True
379 380
            disk.save()

381
    def destroy(self, user=None, task_uuid=None):
382 383 384
        if self.destroyed:
            return False

385 386 387 388
        with disk_activity(code_suffix='destroy', disk=self,
                           task_uuid=task_uuid, user=user):
            self.destroyed = timezone.now()
            self.save()
389

390
            return True
391

392
    def destroy_async(self, user=None):
393 394
        """Execute destroy asynchronously.
        """
395 396
        return local_tasks.destroy.apply_async(args=[self, user],
                                               queue='localhost.man')
397

398
    def restore(self, user=None, task_uuid=None):
399
        """Restore destroyed disk.
400 401 402 403 404 405 406 407
        """
        # TODO
        pass

    def restore_async(self, user=None):
        local_tasks.restore.apply_async(args=[self, user],
                                        queue='localhost.man')

408
    def save_as(self, user=None, task_uuid=None):
409 410 411 412 413 414 415 416 417 418 419 420
        mapping = {
            'qcow2-snap': ('qcow2-norm', self.base),
        }
        if self.type not in mapping.keys():
            raise self.WrongDiskTypeError(self.type)

        if self.is_in_use():
            raise self.DiskInUseError(self)

        # from this point on, the caller has to guarantee that the disk is not
        # going to be used until the operation is complete

421 422 423 424 425 426 427
        with disk_activity(code_suffix='save_as', disk=self,
                           task_uuid=task_uuid, user=user):

            filename = str(uuid.uuid4())
            new_type, new_base = mapping[self.type]

            disk = Disk.objects.create(base=new_base, datastore=self.datastore,
428
                                       filename=filename, name=self.name,
429 430
                                       size=self.size, type=new_type)

431
            queue_name = self.get_remote_queue_name('storage')
432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447
            remote_tasks.merge.apply_async(args=[self.get_disk_desc(),
                                                 disk.get_disk_desc()],
                                           queue=queue_name).get()

            disk.ready = True
            disk.save()

            return disk


class DiskActivity(ActivityModel):
    disk = ForeignKey(Disk, related_name='activity_log',
                      help_text=_('Disk this activity works on.'),
                      verbose_name=_('disk'))

    @classmethod
448
    def create(cls, code_suffix, disk, task_uuid=None, user=None):
449
        act = cls(activity_code='storage.Disk.' + code_suffix,
450
                  disk=disk, parent=None, started=timezone.now(),
451
                  task_uuid=task_uuid, user=user)
452
        act.save()
453
        return act
454

455 456 457
    def create_sub(self, code_suffix, task_uuid=None):
        act = DiskActivity(
            activity_code=self.activity_code + '.' + code_suffix,
458
            disk=self.disk, parent=self, started=timezone.now(),
459 460 461
            task_uuid=task_uuid, user=self.user)
        act.save()
        return act
462

463 464 465
    @contextmanager
    def sub_activity(self, code_suffix, task_uuid=None):
        act = self.create_sub(code_suffix, task_uuid)
466
        return activitycontextimpl(act)
467

468

469
@contextmanager
470 471
def disk_activity(code_suffix, disk, task_uuid=None, user=None,
                  on_abort=None, on_commit=None):
472
    act = DiskActivity.create(code_suffix, disk, task_uuid, user)
473
    return activitycontextimpl(act, on_abort=on_abort, on_commit=on_commit)