node.py 12.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
Ő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,
Ő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 91
    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'
        permissions = ()
92
        ordering = ('-enabled', 'normalized_name')
Őry Máté committed
93

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

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

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

110 111
    online = property(get_online)

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

121
    info = property(get_info)
122

123 124 125 126 127 128 129 130 131 132
    @property
    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']

133 134 135 136 137
    STATES = {False: {False: ('OFFLINE', _('offline')),
                      True: ('DISABLED', _('disabled'))},
              True: {False: ('MISSING', _('missing')),
                     True: ('ONLINE', _('online'))}}

Őry Máté committed
138
    def get_state(self):
139
        """The state combined of online and enabled attributes.
140
        """
141 142
        return self.STATES[self.enabled][self.online][0]

Őry Máté committed
143 144
    state = property(get_state)

145 146
    def get_status_display(self):
        return self.STATES[self.enabled][self.online][1]
147

148
    def enable(self, user=None, base_activity=None):
149
        raise NotImplementedError("Use activate or passivate instead.")
Őry Máté committed
150 151

    @property
152
    @node_available
Őry Máté committed
153 154 155 156 157
    def ram_size_with_overcommit(self):
        """Bytes of total memory including overcommit margin.
        """
        return self.ram_size * self.overcommit

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

Dudás Ádám committed
162 163
        Throws Exception if there is no worker on the queue.
        The result may include dead queues because of caching.
164
        """
165

166 167 168 169
        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
170
            self.node_online()
171
            return queue_name
172
        else:
173
            if self.enabled:
174
                self.node_offline()
175
            raise WorkerNotFound()
Őry Máté committed
176

177 178 179 180 181 182 183 184 185 186 187
    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
188
                act = NodeActivity.create(code_suffix='monitor_success_online',
189 190 191 192 193 194
                                          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
195
                self.get_info(invalidate_cache=True)
196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220

    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?

221 222
    def remote_query(self, task, timeout=30, priority=None, raise_=False,
                     default=None):
Őry Máté committed
223 224
        """Query the given task, and get the result.

225 226 227 228
        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
229
        try:
230
            r = task.apply_async(
231 232
                queue=self.get_remote_queue_name('vm', priority),
                expires=timeout + 60)
Őry Máté committed
233
            return r.get(timeout=timeout)
234
        except (TimeoutError, WorkerNotFound):
Őry Máté committed
235 236 237 238 239
            if raise_:
                raise
            else:
                return default

Bach Dániel committed
240
    @property
241
    @node_available
Bach Dániel committed
242
    @method_cache(10)
Bach Dániel committed
243 244
    def monitor_info(self):
        metrics = ('cpu.usage', 'memory.usage')
245
        prefix = 'circle.%s.' % self.host.hostname
Bach Dániel committed
246 247 248 249 250
        params = [('target', '%s%s' % (prefix, metric))
                  for metric in metrics]
        params.append(('from', '-5min'))
        params.append(('format', 'json'))

251
        try:
Bach Dániel committed
252 253 254 255 256 257 258 259 260 261 262 263
            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):]
264 265
                    else:
                        continue
Bach Dániel committed
266 267 268 269 270 271 272 273
                    value = target['datapoints'][-2][0]
                    retval[metric] = float(value)
                except (KeyError, IndexError, ValueError):
                    continue

            return retval
        except:
            logger.exception('Unhandled exception: ')
274 275
            return self.remote_query(vm_tasks.get_node_metrics, timeout=30,
                                     priority="fast")
276

277
    @property
278
    @node_available
279
    def cpu_usage(self):
Bach Dániel committed
280
        return self.monitor_info.get('cpu.usage') / 100
281

282
    @property
283
    @node_available
284
    def ram_usage(self):
Bach Dániel committed
285
        return self.monitor_info.get('memory.usage') / 100
286

287
    @property
288
    @node_available
289 290 291
    def byte_ram_usage(self):
        return self.ram_usage * self.ram_size

292 293
    def get_status_icon(self):
        return {
294
            'OFFLINE': 'fa-minus-circle',
Kálmán Viktor committed
295
            'DISABLED': 'fa-moon-o',
296 297 298
            'MISSING': 'fa-warning',
            'ONLINE': 'fa-play-circle'}.get(self.get_state(),
                                            'fa-question-circle')
299

300 301
    def get_status_label(self):
        return {
302 303 304
            'OFFLINE': 'label-warning',
            'DISABLED': 'label-warning',
            'MISSING': 'label-danger',
305 306 307
            'ONLINE': 'label-success'}.get(self.get_state(),
                                           'label-danger')

308
    @node_available
Őry Máté committed
309
    def update_vm_states(self):
310 311 312 313 314
        """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
315
        domains = {}
316 317
        domain_list = self.remote_query(vm_tasks.list_domains_info, timeout=5,
                                        priority="fast")
318 319 320 321
        if domain_list is None:
            logger.info("Monitoring failed at: %s", self.name)
            return
        for i in domain_list:
Őry Máté committed
322 323 324 325 326 327 328 329
            # [{'name': 'cloud-1234', 'state': 'RUNNING', ...}, ...]
            try:
                id = int(i['name'].split('-')[1])
            except:
                pass  # name format doesn't match
            else:
                domains[id] = i['state']

330 331
        instances = [{'id': i.id, 'state': i.state}
                     for i in self.instance_set.order_by('id').all()]
Őry Máté committed
332 333 334 335 336 337
        for i in instances:
            try:
                d = domains[i['id']]
            except KeyError:
                logger.info('Node %s update: instance %s missing from '
                            'libvirt', self, i['id'])
338
                # Set state to STOPPED when instance is missing
339 340
                self.instance_set.get(id=i['id']).vm_state_changed(
                    'STOPPED', None)
Őry Máté committed
341 342 343 344 345
            else:
                if d != i['state']:
                    logger.info('Node %s update: instance %s state changed '
                                '(libvirt: %s, db: %s)',
                                self, i['id'], d, i['state'])
346
                    self.instance_set.get(id=i['id']).vm_state_changed(d)
Őry Máté committed
347 348

                del domains[i['id']]
Őry Máté committed
349
        for id, state in domains.iteritems():
350
            from .instance import Instance
Őry Máté committed
351 352 353
            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)
354 355 356

    @classmethod
    def get_state_count(cls, online, enabled):
357 358
        return len([1 for i in
                    cls.objects.filter(enabled=enabled).select_related('host')
359
                    if i.online == online])
360 361 362 363

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