node.py 12.5 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
Őry Máté committed
19
from logging import getLogger
20
from warnings import warn
Őry Máté committed
21 22

from django.db.models import (
23
    CharField, IntegerField, ForeignKey, BooleanField, ManyToManyField,
24
    FloatField, permalink,
Őry Máté committed
25
)
26
from django.utils import timezone
Őry Máté committed
27 28 29 30 31 32
from django.utils.translation import ugettext_lazy as _

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

33
from common.models import method_cache, WorkerNotFound, HumanSortField
34
from common.operations import OperatedMixin
Őry Máté committed
35
from firewall.models import Host
Gregory Nagy committed
36 37
from monitor.calvin.calvin import Query
from monitor.calvin.calvin import GraphiteHandler
38 39 40 41
from ..tasks import vm_tasks
from .activity import node_activity, NodeActivity
from .common import Trait

42

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


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


57
class Node(OperatedMixin, TimeStampedModel):
Őry Máté committed
58 59 60 61 62 63

    """A VM host machine, a hypervisor.
    """
    name = CharField(max_length=50, unique=True,
                     verbose_name=_('name'),
                     help_text=_('Human readable name of node.'))
64
    normalized_name = HumanSortField(monitor='name', max_length=100)
Őry Máté committed
65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83
    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.'))
    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 = ()
84
        ordering = ('-enabled', 'normalized_name')
Őry Máté committed
85

Őry Máté committed
86 87 88
    def __unicode__(self):
        return self.name

89
    @method_cache(10)
90
    def get_online(self):
91
        """Check if the node is online.
Őry Máté committed
92

93
        Check if node is online by queue is available.
94 95
        """
        try:
96
            self.get_remote_queue_name("vm", "fast")
97
        except:
98
            return False
99 100
        else:
            return True
Őry Máté committed
101

102 103
    online = property(get_online)

104
    @node_available
Őry Máté committed
105
    @method_cache(300)
106 107
    def get_info(self):
        return self.remote_query(vm_tasks.get_info,
108
                                 priority='fast',
109 110 111
                                 default={'core_num': '',
                                          'ram_size': '0',
                                          'architecture': ''})
112

113
    info = property(get_info)
114

115 116 117 118 119 120 121 122 123 124
    @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']

125 126 127 128 129
    STATES = {False: {False: ('OFFLINE', _('offline')),
                      True: ('DISABLED', _('disabled'))},
              True: {False: ('MISSING', _('missing')),
                     True: ('ONLINE', _('online'))}}

Őry Máté committed
130
    def get_state(self):
131
        """The state combined of online and enabled attributes.
132
        """
133 134
        return self.STATES[self.enabled][self.online][0]

Őry Máté committed
135 136
    state = property(get_state)

137 138
    def get_status_display(self):
        return self.STATES[self.enabled][self.online][1]
139

140
    def disable(self, user=None, base_activity=None):
141
        ''' Disable the node.'''
142
        if self.enabled:
143 144 145 146 147
            if base_activity:
                act_ctx = base_activity.sub_activity('disable')
            else:
                act_ctx = node_activity('disable', node=self, user=user)
            with act_ctx:
148 149
                self.enabled = False
                self.save()
150 151 152

    def enable(self, user=None):
        ''' Enable the node. '''
153 154 155 156
        if self.enabled is not True:
            with node_activity(code_suffix='enable', node=self, user=user):
                self.enabled = True
                self.save()
Bach Dániel committed
157
            self.get_info(invalidate_cache=True)
Őry Máté committed
158 159

    @property
160
    @node_available
Őry Máté committed
161 162 163 164 165
    def ram_size_with_overcommit(self):
        """Bytes of total memory including overcommit margin.
        """
        return self.ram_size * self.overcommit

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

Dudás Ádám committed
170 171
        Throws Exception if there is no worker on the queue.
        The result may include dead queues because of caching.
172
        """
173

174 175 176 177
        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
178
            self.node_online()
179
            return queue_name
180
        else:
181
            if self.enabled:
182
                self.node_offline()
183
            raise WorkerNotFound()
Őry Máté committed
184

185 186 187 188 189 190 191 192 193 194 195
    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
196
                act = NodeActivity.create(code_suffix='monitor_success_online',
197 198 199 200 201 202
                                          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
203
                self.get_info(invalidate_cache=True)
204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228

    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?

229 230
    def remote_query(self, task, timeout=30, priority=None, raise_=False,
                     default=None):
Őry Máté committed
231 232
        """Query the given task, and get the result.

233 234 235 236
        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
237
        try:
238
            r = task.apply_async(
239 240
                queue=self.get_remote_queue_name('vm', priority),
                expires=timeout + 60)
Őry Máté committed
241
            return r.get(timeout=timeout)
242
        except (TimeoutError, WorkerNotFound):
Őry Máté committed
243 244 245 246 247
            if raise_:
                raise
            else:
                return default

248
    @node_available
Bach Dániel committed
249
    @method_cache(10)
250
    def get_monitor_info(self):
251 252
        try:
            handler = GraphiteHandler()
253
        except RuntimeError:
254 255
            return self.remote_query(vm_tasks.get_node_metrics, timeout=30,
                                     priority="fast")
256

257
        query = Query()
Gregory Nagy committed
258 259 260
        query.set_target(self.host.hostname + ".circle")
        query.set_format("json")
        query.set_relative_start(5, "minutes")
261

262
        metrics = ["cpu.usage", "memory.usage"]
263
        for metric in metrics:
Gregory Nagy committed
264
            query.set_metric(metric)
265 266 267
            query.generate()
            handler.put(query)
            handler.send()
268 269

        collected = {}
270
        for metric in metrics:
Gregory Nagy committed
271
            response = handler.pop()
272
            try:
Bach Dániel committed
273
                cache = response[0]["datapoints"][-2][0]
274 275
            except (IndexError, KeyError):
                cache = 0
276 277 278 279 280
            if cache is None:
                cache = 0
            collected[metric] = cache
        return collected

281
    @property
282
    @node_available
283
    def cpu_usage(self):
284
        return float(self.get_monitor_info()["cpu.usage"]) / 100
285

286
    @property
287
    @node_available
288
    def ram_usage(self):
289
        return float(self.get_monitor_info()["memory.usage"]) / 100
290

291
    @property
292
    @node_available
293 294 295
    def byte_ram_usage(self):
        return self.ram_usage * self.ram_size

296 297 298 299 300 301 302 303
    def get_status_icon(self):
        return {
            'OFFLINE': 'icon-minus-sign',
            'DISABLED': 'icon-moon',
            'MISSING': 'icon-warning-sign',
            'ONLINE': 'icon-play-sign'}.get(self.get_state(),
                                            'icon-question-sign')

304 305
    def get_status_label(self):
        return {
306 307 308
            'OFFLINE': 'label-warning',
            'DISABLED': 'label-warning',
            'MISSING': 'label-danger',
309 310 311
            'ONLINE': 'label-success'}.get(self.get_state(),
                                           'label-danger')

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

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

                del domains[i['id']]
        for i in domains.keys():
            logger.info('Node %s update: domain %s in libvirt but not in db.',
                        self, i)
355 356 357

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

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