node.py 13.6 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
19
from functools import update_wrapper
Őry Máté committed
20
from logging import getLogger
21
from warnings import warn
Bach Dániel committed
22
import requests
Őry Máté committed
23

Bach Dániel committed
24
from django.conf import settings
Őry Máté committed
25
from django.db.models import (
26
    CharField, IntegerField, ForeignKey, BooleanField, ManyToManyField,
27
    FloatField, permalink, Sum
Őry Máté committed
28
)
29
from django.utils import timezone
30
from django.utils.translation import ugettext_lazy as _
Őry Máté committed
31 32 33 34 35

from celery.exceptions import TimeoutError
from model_utils.models import TimeStampedModel
from taggit.managers import TaggableManager

36
from common.models import method_cache, WorkerNotFound, HumanSortField
37
from common.operations import OperatedMixin
Őry Máté committed
38
from firewall.models import Host
39
from ..tasks import vm_tasks
40
from .activity import NodeActivity
41 42
from .common import Trait

43

Őry Máté committed
44 45 46
logger = getLogger(__name__)


47 48 49 50
def node_available(function):
    """Decorate methods to ignore disabled Nodes.
    """
    def decorate(self, *args, **kwargs):
51
        if self.enabled and self.online:
52 53 54
            return function(self, *args, **kwargs)
        else:
            return None
55 56
    update_wrapper(decorate, function)
    decorate._original = function
57 58 59
    return decorate


60
class Node(OperatedMixin, TimeStampedModel):
Őry Máté committed
61 62 63 64 65 66

    """A VM host machine, a hypervisor.
    """
    name = CharField(max_length=50, unique=True,
                     verbose_name=_('name'),
                     help_text=_('Human readable name of node.'))
67
    normalized_name = HumanSortField(monitor='name', max_length=100)
Őry Máté committed
68 69 70 71 72 73 74
    priority = IntegerField(verbose_name=_('priority'),
                            help_text=_('Node usage priority.'))
    host = ForeignKey(Host, verbose_name=_('host'),
                      help_text=_('Host in firewall.'))
    enabled = BooleanField(verbose_name=_('enabled'), default=False,
                           help_text=_('Indicates whether the node can '
                                       'be used for hosting.'))
75 76 77 78 79
    schedule_enabled = BooleanField(verbose_name=_('schedule enabled'),
                                    default=False, help_text=_(
                                        'Indicates whether a vm can be '
                                        'automatically scheduled to this '
                                        'node.'))
Őry Máté committed
80 81 82 83 84 85 86 87 88 89 90
    traits = ManyToManyField(Trait, blank=True,
                             help_text=_("Declared traits."),
                             verbose_name=_('traits'))
    tags = TaggableManager(blank=True, verbose_name=_("tags"))
    overcommit = FloatField(default=1.0, verbose_name=_("overcommit ratio"),
                            help_text=_("The ratio of total memory with "
                                        "to without overcommit."))

    class Meta:
        app_label = 'vm'
        db_table = 'vm_node'
91 92 93
        permissions = (
            ('view_statistics', _('Can view Node box and statistics.')),
        )
94
        ordering = ('-enabled', 'normalized_name')
Őry Máté committed
95

Őry Máté committed
96 97 98
    def __unicode__(self):
        return self.name

99
    @method_cache(10)
100
    def get_online(self):
101
        """Check if the node is online.
Őry Máté committed
102

103
        Check if node is online by queue is available.
104 105
        """
        try:
106
            self.get_remote_queue_name("vm", "fast")
107
        except:
108
            return False
109 110
        else:
            return True
Őry Máté committed
111

112 113
    online = property(get_online)

114
    @node_available
Őry Máté committed
115
    @method_cache(300)
116 117
    def get_info(self):
        return self.remote_query(vm_tasks.get_info,
118
                                 priority='fast',
119 120
                                 default={'core_num': 0,
                                          'ram_size': 0,
121
                                          'architecture': ''})
122

123
    info = property(get_info)
124

125
    @property
126 127 128 129 130
    def allocated_ram(self):
        return (self.instance_set.aggregate(
            r=Sum('ram_size'))['r'] or 0) * 1024 * 1024

    @property
131 132 133 134 135 136 137 138 139
    def ram_size(self):
        warn('Use Node.info["ram_size"]', DeprecationWarning)
        return self.info['ram_size']

    @property
    def num_cores(self):
        warn('Use Node.info["core_num"]', DeprecationWarning)
        return self.info['core_num']

140 141
    STATES = {None: ({True: ('MISSING', _('missing')),
                      False: ('OFFLINE', _('offline'))}),
142 143 144
              False: {False: ('DISABLED', _('disabled'))},
              True: {False: ('PASSIVE', _('passive')),
                     True: ('ACTIVE', _('active'))}}
145

146 147
    def _get_state(self):
        """The state tuple based on online and enabled attributes.
148
        """
149 150 151
        if self.online:
            return self.STATES[self.enabled][self.schedule_enabled]
        else:
152
            return self.STATES[None][self.enabled]
Őry Máté committed
153

154
    def get_status_display(self):
155 156 157 158 159 160
        return self._get_state()[1]

    def get_state(self):
        return self._get_state()[0]

    state = property(get_state)
161

162
    def enable(self, user=None, base_activity=None):
163
        raise NotImplementedError("Use activate or passivate instead.")
Őry Máté committed
164 165

    @property
166
    @node_available
Őry Máté committed
167 168 169 170 171
    def ram_size_with_overcommit(self):
        """Bytes of total memory including overcommit margin.
        """
        return self.ram_size * self.overcommit

172
    @method_cache(30)
173
    def get_remote_queue_name(self, queue_id, priority=None):
Dudás Ádám committed
174
        """Returns the name of the remote celery queue for this node.
175

Dudás Ádám committed
176 177
        Throws Exception if there is no worker on the queue.
        The result may include dead queues because of caching.
178
        """
179

180 181 182 183
        if vm_tasks.check_queue(self.host.hostname, queue_id, priority):
            queue_name = self.host.hostname + "." + queue_id
            if priority is not None:
                queue_name = queue_name + "." + priority
184
            self.node_online()
185
            return queue_name
186
        else:
187
            if self.enabled:
188
                self.node_offline()
189
            raise WorkerNotFound()
Őry Máté committed
190

191 192 193 194 195 196 197 198 199 200 201
    def node_online(self):
        """Create activity and log entry when node reappears.
        """

        try:
            act = self.activity_log.order_by('-pk')[0]
        except IndexError:
            pass  # no monitoring activity at all
        else:
            logger.debug("The last activity was %s" % act)
            if act.activity_code.endswith("offline"):
Dudás Ádám committed
202
                act = NodeActivity.create(code_suffix='monitor_success_online',
203 204 205 206 207 208
                                          node=self, user=None)
                act.started = timezone.now()
                act.finished = timezone.now()
                act.succeeded = True
                act.save()
                logger.info("Node %s is ONLINE." % self.name)
Bach Dániel committed
209
                self.get_info(invalidate_cache=True)
210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234

    def node_offline(self):
        """Called when a node disappears.

        If the node is not already offline, record an activity and a log entry.
        """

        try:
            act = self.activity_log.order_by('-pk')[0]
        except IndexError:
            pass  # no activity at all
        else:
            logger.debug("The last activity was %s" % act)
            if act.activity_code.endswith("offline"):
                return
        act = NodeActivity.create(code_suffix='monitor_failed_offline',
                                  node=self, user=None)
        act.started = timezone.now()
        act.finished = timezone.now()
        act.succeeded = False
        act.save()
        logger.critical("Node %s is OFFLINE%s.", self.name,
                        ", but enabled" if self.enabled else "")
        # TODO: check if we should reschedule any VMs?

235 236
    def remote_query(self, task, timeout=30, priority=None, raise_=False,
                     default=None):
Őry Máté committed
237 238
        """Query the given task, and get the result.

239 240 241 242
        If the result is not ready or worker not reachable
        in timeout secs, return default value or raise a
        TimeoutError or WorkerNotFound exception.
        """
Őry Máté committed
243
        try:
244
            r = task.apply_async(
245 246
                queue=self.get_remote_queue_name('vm', priority),
                expires=timeout + 60)
Őry Máté committed
247
            return r.get(timeout=timeout)
248
        except (TimeoutError, WorkerNotFound):
Őry Máté committed
249 250 251 252 253
            if raise_:
                raise
            else:
                return default

Bach Dániel committed
254
    @property
255
    @node_available
Bach Dániel committed
256
    @method_cache(10)
Bach Dániel committed
257
    def monitor_info(self):
258
        metrics = ('cpu.percent', 'memory.usage')
259
        prefix = 'circle.%s.' % self.host.hostname
Bach Dániel committed
260 261 262 263 264
        params = [('target', '%s%s' % (prefix, metric))
                  for metric in metrics]
        params.append(('from', '-5min'))
        params.append(('format', 'json'))

265
        try:
Bach Dániel committed
266 267 268 269 270 271 272 273 274 275 276 277
            logger.info('%s %s', settings.GRAPHITE_URL, params)
            response = requests.get(settings.GRAPHITE_URL, params=params)

            retval = {}
            for target in response.json():
                # Example:
                # {"target": "circle.szianode.cpu.usage",
                #  "datapoints": [[0.6, 1403045700], [0.5, 1403045760]
                try:
                    metric = target['target']
                    if metric.startswith(prefix):
                        metric = metric[len(prefix):]
278 279
                    else:
                        continue
Bach Dániel committed
280 281 282 283 284 285 286 287
                    value = target['datapoints'][-2][0]
                    retval[metric] = float(value)
                except (KeyError, IndexError, ValueError):
                    continue

            return retval
        except:
            logger.exception('Unhandled exception: ')
288 289
            return self.remote_query(vm_tasks.get_node_metrics, timeout=30,
                                     priority="fast")
290

291
    @property
292
    @node_available
293
    def cpu_usage(self):
294
        return self.monitor_info.get('cpu.percent') / 100
295

296
    @property
297
    @node_available
298
    def ram_usage(self):
Bach Dániel committed
299
        return self.monitor_info.get('memory.usage') / 100
300

301
    @property
302
    @node_available
303 304 305
    def byte_ram_usage(self):
        return self.ram_usage * self.ram_size

306 307
    def get_status_icon(self):
        return {
308 309
            'DISABLED': 'fa-times-circle-o',
            'OFFLINE': 'fa-times-circle',
310
            'MISSING': 'fa-warning',
311 312
            'PASSIVE': 'fa-play-circle-o',
            'ACTIVE': 'fa-play-circle'}.get(self.get_state(),
313
                                            'fa-question-circle')
314

315 316
    def get_status_label(self):
        return {
317
            'OFFLINE': 'label-warning',
318
            'DISABLED': 'label-danger',
319
            'MISSING': 'label-danger',
320 321 322
            'ACTIVE': 'label-success',
            'PASSIVE': 'label-warning',
        }.get(self.get_state(), 'label-danger')
323

324
    @node_available
Őry Máté committed
325
    def update_vm_states(self):
326 327 328 329 330
        """Update state of Instances running on this Node.

        Query state of all libvirt domains, and notify Instances by their
        vm_state_changed hook.
        """
Őry Máté committed
331
        domains = {}
332 333
        domain_list = self.remote_query(vm_tasks.list_domains_info, timeout=5,
                                        priority="fast")
334 335 336 337
        if domain_list is None:
            logger.info("Monitoring failed at: %s", self.name)
            return
        for i in domain_list:
Őry Máté committed
338 339 340 341 342 343 344 345
            # [{'name': 'cloud-1234', 'state': 'RUNNING', ...}, ...]
            try:
                id = int(i['name'].split('-')[1])
            except:
                pass  # name format doesn't match
            else:
                domains[id] = i['state']

346 347
        instances = [{'id': i.id, 'state': i.state}
                     for i in self.instance_set.order_by('id').all()]
Őry Máté committed
348 349 350 351 352 353
        for i in instances:
            try:
                d = domains[i['id']]
            except KeyError:
                logger.info('Node %s update: instance %s missing from '
                            'libvirt', self, i['id'])
354
                # Set state to STOPPED when instance is missing
355 356
                self.instance_set.get(id=i['id']).vm_state_changed(
                    'STOPPED', None)
Őry Máté committed
357 358 359 360 361
            else:
                if d != i['state']:
                    logger.info('Node %s update: instance %s state changed '
                                '(libvirt: %s, db: %s)',
                                self, i['id'], d, i['state'])
362
                    self.instance_set.get(id=i['id']).vm_state_changed(d)
Őry Máté committed
363 364

                del domains[i['id']]
Őry Máté committed
365
        for id, state in domains.iteritems():
366
            from .instance import Instance
Őry Máté committed
367 368 369
            logger.error('Node %s update: domain %s in libvirt but not in db.',
                         self, id)
            Instance.objects.get(id=id).vm_state_changed(state, self)
370 371 372

    @classmethod
    def get_state_count(cls, online, enabled):
373 374
        return len([1 for i in
                    cls.objects.filter(enabled=enabled).select_related('host')
375
                    if i.online == online])
376 377 378 379

    @permalink
    def get_absolute_url(self):
        return ('dashboard.views.node-detail', None, {'pk': self.id})
380 381 382 383 384

    def save(self, *args, **kwargs):
        if not self.enabled:
            self.schedule_enabled = False
        super(Node, self).save(*args, **kwargs)
385

386 387 388
    @property
    def metric_prefix(self):
        return 'circle.%s' % self.host.hostname