node.py 17.7 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
20
from glob import glob
Őry Máté committed
21
from logging import getLogger
22
import os.path
23
from warnings import warn
Bach Dániel committed
24
import requests
25
from salt.client import LocalClient
Bach Dániel committed
26
from salt.exceptions import SaltClientError
27 28
import salt.utils
from time import time, sleep
Őry Máté committed
29

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

38
from celery.exceptions import TimeoutError, TaskRevokedError
Őry Máté committed
39 40 41
from model_utils.models import TimeStampedModel
from taggit.managers import TaggableManager

42
from common.models import method_cache, WorkerNotFound, HumanSortField
43
from common.operations import OperatedMixin
Őry Máté committed
44
from firewall.models import Host
45
from ..tasks import vm_tasks
46
from .activity import NodeActivity
47 48
from .common import Trait

49

Őry Máté committed
50 51 52
logger = getLogger(__name__)


53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82
class MyLocalClient(LocalClient):
    def get_returns(self, jid, minions, timeout=None):
        '''
        Get the returns for the command line interface via the event system
        '''
        minions = set(minions)
        if timeout is None:
            timeout = self.opts['timeout']
        jid_dir = salt.utils.jid_dir(jid,
                                     self.opts['cachedir'],
                                     self.opts['hash_type'])
        start = time()
        timeout_at = start + timeout

        found = set()
        ret = {}
        wtag = os.path.join(jid_dir, 'wtag*')
        # Check to see if the jid is real, if not return the empty dict
        if not os.path.isdir(jid_dir):
            logger.warning("jid_dir (%s) does not exist", jid_dir)
            return ret
        # Wait for the hosts to check in
        while True:
            time_left = timeout_at - time()
            raw = self.event.get_event(time_left, jid)
            if raw is not None and 'return' in raw:
                found.add(raw['id'])
                ret[raw['id']] = raw['return']
                if len(found.intersection(minions)) >= len(minions):
                    # All minions have returned, break out of the loop
Turcsik Máté committed
83
                    logger.debug("jid %s found all minions", jid)
84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102
                    break
                continue
            # Then event system timeout was reached and nothing was returned
            if len(found.intersection(minions)) >= len(minions):
                # All minions have returned, break out of the loop
                logger.debug("jid %s found all minions", jid)
                break
            if glob(wtag) and time() <= timeout_at + 1:
                # The timeout +1 has not been reached and there is still a
                # write tag for the syndic
                continue
            if time() > timeout_at:
                logger.info('jid %s minions %s did not return in time',
                            jid, (minions - found))
                break
            sleep(0.01)
        return ret


103 104 105 106
def node_available(function):
    """Decorate methods to ignore disabled Nodes.
    """
    def decorate(self, *args, **kwargs):
107
        if self.enabled and self.online:
108 109 110
            return function(self, *args, **kwargs)
        else:
            return None
111 112
    update_wrapper(decorate, function)
    decorate._original = function
113 114 115
    return decorate


116
class Node(OperatedMixin, TimeStampedModel):
Őry Máté committed
117 118 119 120 121 122

    """A VM host machine, a hypervisor.
    """
    name = CharField(max_length=50, unique=True,
                     verbose_name=_('name'),
                     help_text=_('Human readable name of node.'))
123
    normalized_name = HumanSortField(monitor='name', max_length=100)
Őry Máté committed
124 125 126 127 128 129 130
    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.'))
131 132 133 134 135 136 137 138 139
    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
140 141 142 143 144 145 146
    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."))
147 148 149 150 151 152 153 154 155 156 157 158 159 160 161
    ram_weight = FloatField(
        default=1.0,
        help_text=_("Indicates the relative RAM quantity of this node."),
        verbose_name=_("RAM Weight")
    )
    cpu_weight = FloatField(
        default=1.0,
        help_text=_("Indicates the relative CPU power of this node."),
        verbose_name=_("CPU Weight")
    )
    time_stamp = DateTimeField(
        auto_now_add=True,
        help_text=_("A timestamp for the node, used by the scheduler."),
        verbose_name=_("Last Scheduled Time Stamp")
    )
Őry Máté committed
162 163 164 165

    class Meta:
        app_label = 'vm'
        db_table = 'vm_node'
166 167 168
        permissions = (
            ('view_statistics', _('Can view Node box and statistics.')),
        )
169
        ordering = ('-enabled', 'normalized_name')
Őry Máté committed
170

Őry Máté committed
171 172 173
    def __unicode__(self):
        return self.name

174
    @method_cache(10)
175
    def get_online(self):
176
        """Check if the node is online.
Őry Máté committed
177

178
        Check if node is online by queue is available.
179 180
        """
        try:
181
            self.get_remote_queue_name("vm", "fast")
182 183
            self.get_remote_queue_name("vm", "slow")
            self.get_remote_queue_name("net", "fast")
184
        except Exception:
185
            return False
186 187
        else:
            return True
Őry Máté committed
188

189 190
    online = property(get_online)

191 192 193 194
    @method_cache(20)
    def get_minion_online(self):
        name = self.host.hostname
        try:
Bach Dániel committed
195 196
            client = MyLocalClient()
            client.opts['timeout'] = 0.2
197
            return bool(client.cmd(name, 'test.ping')[name])
Bach Dániel committed
198
        except (KeyError, SaltClientError):
199 200 201 202
            return False

    minion_online = property(get_minion_online)

203
    @node_available
Őry Máté committed
204
    @method_cache(300)
205 206
    def get_info(self):
        return self.remote_query(vm_tasks.get_info,
207
                                 priority='fast',
208 209
                                 default={'core_num': 0,
                                          'ram_size': 0,
210
                                          'architecture': ''})
211

212
    info = property(get_info)
213

214
    @property
215 216 217 218 219
    def allocated_ram(self):
        return (self.instance_set.aggregate(
            r=Sum('ram_size'))['r'] or 0) * 1024 * 1024

    @property
220 221 222 223 224 225 226 227 228
    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']

229 230
    STATES = {None: ({True: ('MISSING', _('missing')),
                      False: ('OFFLINE', _('offline'))}),
231 232 233
              False: {False: ('DISABLED', _('disabled'))},
              True: {False: ('PASSIVE', _('passive')),
                     True: ('ACTIVE', _('active'))}}
234

235 236
    def _get_state(self):
        """The state tuple based on online and enabled attributes.
237
        """
238 239 240
        if self.online:
            return self.STATES[self.enabled][self.schedule_enabled]
        else:
241
            return self.STATES[None][self.enabled]
Őry Máté committed
242

243
    def get_status_display(self):
244 245 246 247 248 249
        return self._get_state()[1]

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

    state = property(get_state)
250

251
    def enable(self, user=None, base_activity=None):
252
        raise NotImplementedError("Use activate or passivate instead.")
Őry Máté committed
253 254

    @property
255
    @node_available
Őry Máté committed
256 257 258 259 260
    def ram_size_with_overcommit(self):
        """Bytes of total memory including overcommit margin.
        """
        return self.ram_size * self.overcommit

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

Dudás Ádám committed
265 266
        Throws Exception if there is no worker on the queue.
        The result may include dead queues because of caching.
267
        """
268

269 270 271 272
        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
273
            self.node_online()
274
            return queue_name
275
        else:
276
            if self.enabled:
277
                self.node_offline()
278
            raise WorkerNotFound()
Őry Máté committed
279

280 281 282 283 284 285 286 287 288 289 290
    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
291
                act = NodeActivity.create(code_suffix='monitor_success_online',
292 293 294 295 296 297
                                          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
298
                self.get_info(invalidate_cache=True)
299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323

    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?

324 325
    def remote_query(self, task, timeout=30, priority=None, raise_=False,
                     default=None):
Őry Máté committed
326 327
        """Query the given task, and get the result.

328 329 330 331
        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
332
        try:
333
            r = task.apply_async(
334 335
                queue=self.get_remote_queue_name('vm', priority),
                expires=timeout + 60)
Őry Máté committed
336
            return r.get(timeout=timeout)
337
        except (TimeoutError, WorkerNotFound, TaskRevokedError):
Őry Máté committed
338 339 340 341 342
            if raise_:
                raise
            else:
                return default

Bach Dániel committed
343
    @property
344
    @node_available
Bach Dániel committed
345
    @method_cache(10)
Bach Dániel committed
346
    def monitor_info(self):
347
        metrics = ('cpu.percent', 'memory.usage')
348
        prefix = 'circle.%s.' % self.host.hostname
Bach Dániel committed
349 350 351 352 353
        params = [('target', '%s%s' % (prefix, metric))
                  for metric in metrics]
        params.append(('from', '-5min'))
        params.append(('format', 'json'))

354
        try:
Bach Dániel committed
355 356 357 358 359 360 361 362
            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]
363
                logger.info('MONITOR_TARGET: %s', target)
Bach Dániel committed
364 365 366 367
                try:
                    metric = target['target']
                    if metric.startswith(prefix):
                        metric = metric[len(prefix):]
368
                    else:
369
            	        logger.info('MONITOR_MET: %s %s', target, metric)
370
                        continue
371 372 373
                    value = target['datapoints'][-1][0]
                    if value is None: 
                        value = target['datapoints'][-2][0]
Bach Dániel committed
374
                    retval[metric] = float(value)
375
                    logger.info('MONITOR_RETVAL: %s %s, %s', target['target'], metric, retval[metric])
376
                except (KeyError, IndexError, ValueError, TypeError):
377
                    logger.info('MONITOR_ERR: %s %s', metric, value)
Bach Dániel committed
378 379 380
                    continue

            return retval
381
        except Exception:
Bach Dániel committed
382
            logger.exception('Unhandled exception: ')
383 384
            return self.remote_query(vm_tasks.get_node_metrics, timeout=30,
                                     priority="fast")
385

386
    @property
387
    @node_available
388
    def driver_version(self):
389
        return self.info.get('driver_version')
390

391
    def get_monitor_info(self, metric):
392 393
        # return with the metric value if the monitor info not none
        # or return 0 if its None or the metric unreachable
394
        if self.monitor_info is None:
395
            logger.warning('Monitor info is None')
396 397
            return 0
        elif self.monitor_info.get(metric) is None:
Chif Gergő committed
398
            logger.warning('Unreachable monitor info of: ' + metric)
399 400 401 402
            return 0
        else:
            return self.monitor_info.get(metric)

403 404
    @property
    @node_available
405
    def cpu_usage(self):
406
        return self.get_monitor_info('cpu.percent') / 100
407

408
    @property
409
    @node_available
410
    def ram_usage(self):
411
        return self.get_monitor_info('memory.usage') / 100
412

413
    @property
414
    @node_available
415 416 417
    def byte_ram_usage(self):
        return self.ram_usage * self.ram_size

418 419
    def get_status_icon(self):
        return {
420 421
            'DISABLED': 'fa-times-circle-o',
            'OFFLINE': 'fa-times-circle',
422
            'MISSING': 'fa-warning',
423 424
            'PASSIVE': 'fa-play-circle-o',
            'ACTIVE': 'fa-play-circle'}.get(self.get_state(),
425
                                            'fa-question-circle')
426

427 428
    def get_status_label(self):
        return {
429
            'OFFLINE': 'label-warning',
430
            'DISABLED': 'label-danger',
431
            'MISSING': 'label-danger',
432 433 434
            'ACTIVE': 'label-success',
            'PASSIVE': 'label-warning',
        }.get(self.get_state(), 'label-danger')
435

436
    @node_available
Őry Máté committed
437
    def update_vm_states(self):
438 439 440 441 442
        """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
443
        domains = {}
444
        domain_list = self.remote_query(vm_tasks.list_domains_info, timeout=10,
445
                                        priority="fast")
446 447 448 449
        if domain_list is None:
            logger.info("Monitoring failed at: %s", self.name)
            return
        for i in domain_list:
Őry Máté committed
450 451 452
            # [{'name': 'cloud-1234', 'state': 'RUNNING', ...}, ...]
            try:
                id = int(i['name'].split('-')[1])
453
            except Exception:
Őry Máté committed
454 455 456 457
                pass  # name format doesn't match
            else:
                domains[id] = i['state']

458 459
        instances = [{'id': i.id, 'state': i.state}
                     for i in self.instance_set.order_by('id').all()]
Őry Máté committed
460 461 462 463 464 465
        for i in instances:
            try:
                d = domains[i['id']]
            except KeyError:
                logger.info('Node %s update: instance %s missing from '
                            'libvirt', self, i['id'])
466
                # Set state to STOPPED when instance is missing
467 468
                self.instance_set.get(id=i['id']).vm_state_changed(
                    'STOPPED', None)
Őry Máté committed
469 470 471 472 473
            else:
                if d != i['state']:
                    logger.info('Node %s update: instance %s state changed '
                                '(libvirt: %s, db: %s)',
                                self, i['id'], d, i['state'])
474
                    self.instance_set.get(id=i['id']).vm_state_changed(d)
Őry Máté committed
475 476

                del domains[i['id']]
Őry Máté committed
477
        for id, state in domains.iteritems():
478
            from .instance import Instance
Őry Máté committed
479 480 481
            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)
482 483 484

    @classmethod
    def get_state_count(cls, online, enabled):
485 486
        return len([1 for i in
                    cls.objects.filter(enabled=enabled).select_related('host')
487
                    if i.online == online])
488 489 490 491

    @permalink
    def get_absolute_url(self):
        return ('dashboard.views.node-detail', None, {'pk': self.id})
492 493 494 495 496

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

498 499 500
    @property
    def metric_prefix(self):
        return 'circle.%s' % self.host.hostname