node.py 16.8 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, 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
    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
136 137 138 139 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."))

    class Meta:
        app_label = 'vm'
        db_table = 'vm_node'
147 148 149
        permissions = (
            ('view_statistics', _('Can view Node box and statistics.')),
        )
150
        ordering = ('-enabled', 'normalized_name')
Őry Máté committed
151

Őry Máté committed
152 153 154
    def __unicode__(self):
        return self.name

155
    @method_cache(10)
156
    def get_online(self):
157
        """Check if the node is online.
Őry Máté committed
158

159
        Check if node is online by queue is available.
160 161
        """
        try:
162
            self.get_remote_queue_name("vm", "fast")
163 164
            self.get_remote_queue_name("vm", "slow")
            self.get_remote_queue_name("net", "fast")
165
        except:
166
            return False
167 168
        else:
            return True
Őry Máté committed
169

170 171
    online = property(get_online)

172 173 174 175
    @method_cache(20)
    def get_minion_online(self):
        name = self.host.hostname
        try:
Bach Dániel committed
176 177
            client = MyLocalClient()
            client.opts['timeout'] = 0.2
178
            return bool(client.cmd(name, 'test.ping')[name])
Bach Dániel committed
179
        except (KeyError, SaltClientError):
180 181 182 183
            return False

    minion_online = property(get_minion_online)

184
    @node_available
Őry Máté committed
185
    @method_cache(300)
186 187
    def get_info(self):
        return self.remote_query(vm_tasks.get_info,
188
                                 priority='fast',
189 190
                                 default={'core_num': 0,
                                          'ram_size': 0,
191
                                          'architecture': ''})
192

193
    info = property(get_info)
194

195
    @property
196 197 198 199 200
    def allocated_ram(self):
        return (self.instance_set.aggregate(
            r=Sum('ram_size'))['r'] or 0) * 1024 * 1024

    @property
201 202 203 204 205 206 207 208 209
    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']

210 211
    STATES = {None: ({True: ('MISSING', _('missing')),
                      False: ('OFFLINE', _('offline'))}),
212 213 214
              False: {False: ('DISABLED', _('disabled'))},
              True: {False: ('PASSIVE', _('passive')),
                     True: ('ACTIVE', _('active'))}}
215

216 217
    def _get_state(self):
        """The state tuple based on online and enabled attributes.
218
        """
219 220 221
        if self.online:
            return self.STATES[self.enabled][self.schedule_enabled]
        else:
222
            return self.STATES[None][self.enabled]
Őry Máté committed
223

224
    def get_status_display(self):
225 226 227 228 229 230
        return self._get_state()[1]

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

    state = property(get_state)
231

232
    def enable(self, user=None, base_activity=None):
233
        raise NotImplementedError("Use activate or passivate instead.")
Őry Máté committed
234 235

    @property
236
    @node_available
Őry Máté committed
237 238 239 240 241
    def ram_size_with_overcommit(self):
        """Bytes of total memory including overcommit margin.
        """
        return self.ram_size * self.overcommit

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

Dudás Ádám committed
246 247
        Throws Exception if there is no worker on the queue.
        The result may include dead queues because of caching.
248
        """
249

250 251 252 253
        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
254
            self.node_online()
255
            return queue_name
256
        else:
257
            if self.enabled:
258
                self.node_offline()
259
            raise WorkerNotFound()
Őry Máté committed
260

261 262 263 264 265 266 267 268 269 270 271
    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
272
                act = NodeActivity.create(code_suffix='monitor_success_online',
273 274 275 276 277 278
                                          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
279
                self.get_info(invalidate_cache=True)
280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304

    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?

305 306
    def remote_query(self, task, timeout=30, priority=None, raise_=False,
                     default=None):
Őry Máté committed
307 308
        """Query the given task, and get the result.

309 310 311 312
        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
313
        try:
314
            r = task.apply_async(
315 316
                queue=self.get_remote_queue_name('vm', priority),
                expires=timeout + 60)
Őry Máté committed
317
            return r.get(timeout=timeout)
318
        except (TimeoutError, WorkerNotFound, TaskRevokedError):
Őry Máté committed
319 320 321 322 323
            if raise_:
                raise
            else:
                return default

Bach Dániel committed
324
    @property
325
    @node_available
Bach Dániel committed
326
    @method_cache(10)
Bach Dániel committed
327
    def monitor_info(self):
328
        metrics = ('cpu.percent', 'memory.usage')
329
        prefix = 'circle.%s.' % self.host.hostname
Bach Dániel committed
330 331 332 333 334
        params = [('target', '%s%s' % (prefix, metric))
                  for metric in metrics]
        params.append(('from', '-5min'))
        params.append(('format', 'json'))

335
        try:
Bach Dániel committed
336 337 338 339 340 341 342 343 344 345 346 347
            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):]
348 349
                    else:
                        continue
Bach Dániel committed
350 351
                    value = target['datapoints'][-2][0]
                    retval[metric] = float(value)
352
                except (KeyError, IndexError, ValueError, TypeError):
Bach Dániel committed
353 354 355 356 357
                    continue

            return retval
        except:
            logger.exception('Unhandled exception: ')
358 359
            return self.remote_query(vm_tasks.get_node_metrics, timeout=30,
                                     priority="fast")
360

361
    @property
362
    @node_available
363
    def driver_version(self):
364
        return self.info.get('driver_version')
365

366
    def get_monitor_info(self, metric):
367 368
        # return with the metric value if the monitor info not none
        # or return 0 if its None or the metric unreachable
369
        if self.monitor_info is None:
370
            logger.warning('Monitor info is None')
371 372
            return 0
        elif self.monitor_info.get(metric) is None:
Chif Gergő committed
373
            logger.warning('Unreachable monitor info of: ' + metric)
374 375 376 377
            return 0
        else:
            return self.monitor_info.get(metric)

378 379
    @property
    @node_available
380
    def cpu_usage(self):
381
        return self.get_monitor_info('cpu.percent') / 100
382

383
    @property
384
    @node_available
385
    def ram_usage(self):
386
        return self.get_monitor_info('memory.usage') / 100
387

388
    @property
389
    @node_available
390 391 392
    def byte_ram_usage(self):
        return self.ram_usage * self.ram_size

393 394
    def get_status_icon(self):
        return {
395 396
            'DISABLED': 'fa-times-circle-o',
            'OFFLINE': 'fa-times-circle',
397
            'MISSING': 'fa-warning',
398 399
            'PASSIVE': 'fa-play-circle-o',
            'ACTIVE': 'fa-play-circle'}.get(self.get_state(),
400
                                            'fa-question-circle')
401

402 403
    def get_status_label(self):
        return {
404
            'OFFLINE': 'label-warning',
405
            'DISABLED': 'label-danger',
406
            'MISSING': 'label-danger',
407 408 409
            'ACTIVE': 'label-success',
            'PASSIVE': 'label-warning',
        }.get(self.get_state(), 'label-danger')
410

411
    @node_available
Őry Máté committed
412
    def update_vm_states(self):
413 414 415 416 417
        """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
418
        domains = {}
419 420
        domain_list = self.remote_query(vm_tasks.list_domains_info, timeout=5,
                                        priority="fast")
421 422 423 424
        if domain_list is None:
            logger.info("Monitoring failed at: %s", self.name)
            return
        for i in domain_list:
Őry Máté committed
425 426 427 428 429 430 431 432
            # [{'name': 'cloud-1234', 'state': 'RUNNING', ...}, ...]
            try:
                id = int(i['name'].split('-')[1])
            except:
                pass  # name format doesn't match
            else:
                domains[id] = i['state']

433 434
        instances = [{'id': i.id, 'state': i.state}
                     for i in self.instance_set.order_by('id').all()]
Őry Máté committed
435 436 437 438 439 440
        for i in instances:
            try:
                d = domains[i['id']]
            except KeyError:
                logger.info('Node %s update: instance %s missing from '
                            'libvirt', self, i['id'])
441
                # Set state to STOPPED when instance is missing
442 443
                self.instance_set.get(id=i['id']).vm_state_changed(
                    'STOPPED', None)
Őry Máté committed
444 445 446 447 448
            else:
                if d != i['state']:
                    logger.info('Node %s update: instance %s state changed '
                                '(libvirt: %s, db: %s)',
                                self, i['id'], d, i['state'])
449
                    self.instance_set.get(id=i['id']).vm_state_changed(d)
Őry Máté committed
450 451

                del domains[i['id']]
Őry Máté committed
452
        for id, state in domains.iteritems():
453
            from .instance import Instance
Őry Máté committed
454 455 456
            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)
457 458 459

    @classmethod
    def get_state_count(cls, online, enabled):
460 461
        return len([1 for i in
                    cls.objects.filter(enabled=enabled).select_related('host')
462
                    if i.online == online])
463 464 465 466

    @permalink
    def get_absolute_url(self):
        return ('dashboard.views.node-detail', None, {'pk': self.id})
467 468 469 470 471

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

473 474 475
    @property
    def metric_prefix(self):
        return 'circle.%s' % self.host.hostname