node.py 16.1 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 26 27
from salt.client import LocalClient
import salt.utils
from time import time, sleep
Őry Máté committed
28

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

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

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

48

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


52 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 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101
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
                    logger.debug("jid %s found all minions", jid)
                    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


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


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

    """A VM host machine, a hypervisor.
    """
    name = CharField(max_length=50, unique=True,
                     verbose_name=_('name'),
                     help_text=_('Human readable name of node.'))
122
    normalized_name = HumanSortField(monitor='name', max_length=100)
Őry Máté committed
123 124 125 126 127 128 129
    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.'))
130 131 132 133 134
    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
135 136 137 138 139 140 141 142 143 144 145
    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'
146 147 148
        permissions = (
            ('view_statistics', _('Can view Node box and statistics.')),
        )
149
        ordering = ('-enabled', 'normalized_name')
Őry Máté committed
150

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

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

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

167 168
    online = property(get_online)

169 170 171 172 173 174 175 176 177 178 179 180
    @method_cache(20)
    def get_minion_online(self):
        name = self.host.hostname
        client = MyLocalClient()
        client.opts['timeout'] = 0.2
        try:
            return bool(client.cmd(name, 'test.ping')[name])
        except KeyError:
            return False

    minion_online = property(get_minion_online)

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

190
    info = property(get_info)
191

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

    @property
198 199 200 201 202 203 204 205 206
    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']

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

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

221
    def get_status_display(self):
222 223 224 225 226 227
        return self._get_state()[1]

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

    state = property(get_state)
228

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

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

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

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

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

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

    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?

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

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

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

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

            return retval
        except:
            logger.exception('Unhandled exception: ')
355 356
            return self.remote_query(vm_tasks.get_node_metrics, timeout=30,
                                     priority="fast")
357

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

    @property
    @node_available
365
    def cpu_usage(self):
366
        return self.monitor_info.get('cpu.percent') / 100
367

368
    @property
369
    @node_available
370
    def ram_usage(self):
Bach Dániel committed
371
        return self.monitor_info.get('memory.usage') / 100
372

373
    @property
374
    @node_available
375 376 377
    def byte_ram_usage(self):
        return self.ram_usage * self.ram_size

378 379
    def get_status_icon(self):
        return {
380 381
            'DISABLED': 'fa-times-circle-o',
            'OFFLINE': 'fa-times-circle',
382
            'MISSING': 'fa-warning',
383 384
            'PASSIVE': 'fa-play-circle-o',
            'ACTIVE': 'fa-play-circle'}.get(self.get_state(),
385
                                            'fa-question-circle')
386

387 388
    def get_status_label(self):
        return {
389
            'OFFLINE': 'label-warning',
390
            'DISABLED': 'label-danger',
391
            'MISSING': 'label-danger',
392 393 394
            'ACTIVE': 'label-success',
            'PASSIVE': 'label-warning',
        }.get(self.get_state(), 'label-danger')
395

396
    @node_available
Őry Máté committed
397
    def update_vm_states(self):
398 399 400 401 402
        """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
403
        domains = {}
404 405
        domain_list = self.remote_query(vm_tasks.list_domains_info, timeout=5,
                                        priority="fast")
406 407 408 409
        if domain_list is None:
            logger.info("Monitoring failed at: %s", self.name)
            return
        for i in domain_list:
Őry Máté committed
410 411 412 413 414 415 416 417
            # [{'name': 'cloud-1234', 'state': 'RUNNING', ...}, ...]
            try:
                id = int(i['name'].split('-')[1])
            except:
                pass  # name format doesn't match
            else:
                domains[id] = i['state']

418 419
        instances = [{'id': i.id, 'state': i.state}
                     for i in self.instance_set.order_by('id').all()]
Őry Máté committed
420 421 422 423 424 425
        for i in instances:
            try:
                d = domains[i['id']]
            except KeyError:
                logger.info('Node %s update: instance %s missing from '
                            'libvirt', self, i['id'])
426
                # Set state to STOPPED when instance is missing
427 428
                self.instance_set.get(id=i['id']).vm_state_changed(
                    'STOPPED', None)
Őry Máté committed
429 430 431 432 433
            else:
                if d != i['state']:
                    logger.info('Node %s update: instance %s state changed '
                                '(libvirt: %s, db: %s)',
                                self, i['id'], d, i['state'])
434
                    self.instance_set.get(id=i['id']).vm_state_changed(d)
Őry Máté committed
435 436

                del domains[i['id']]
Őry Máté committed
437
        for id, state in domains.iteritems():
438
            from .instance import Instance
Őry Máté committed
439 440 441
            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)
442 443 444

    @classmethod
    def get_state_count(cls, online, enabled):
445 446
        return len([1 for i in
                    cls.objects.filter(enabled=enabled).select_related('host')
447
                    if i.online == online])
448 449 450 451

    @permalink
    def get_absolute_url(self):
        return ('dashboard.views.node-detail', None, {'pk': self.id})
452 453 454 455 456

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

458 459 460
    @property
    def metric_prefix(self):
        return 'circle.%s' % self.host.hostname