node.py 12.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
Ő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 97
            self.get_remote_queue_name("vm")
        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 109 110
                                 default={'core_num': '',
                                          'ram_size': '0',
                                          'architecture': ''})
111

112
    info = property(get_info)
113

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

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

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

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

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

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

    def enable(self, user=None):
        ''' Enable the node. '''
152 153 154 155
        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
156
            self.get_info(invalidate_cache=True)
Őry Máté committed
157 158

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

165
    @method_cache(30)
Őry Máté committed
166
    def get_remote_queue_name(self, queue_id):
Dudás Ádám committed
167
        """Returns the name of the remote celery queue for this node.
168

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

173
        if vm_tasks.check_queue(self.host.hostname, queue_id):
174
            self.node_online()
175 176
            return self.host.hostname + "." + queue_id
        else:
177
            if self.enabled:
178
                self.node_offline()
179
            raise WorkerNotFound()
Őry Máté committed
180

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

    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?

225
    def remote_query(self, task, timeout=30, raise_=False, default=None):
Őry Máté committed
226 227
        """Query the given task, and get the result.

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

242
    @node_available
243
    def get_monitor_info(self):
244 245
        try:
            handler = GraphiteHandler()
246 247 248
        except RuntimeError:
            return self.remote_query(vm_tasks.get_node_metrics, 30)

249
        query = Query()
Gregory Nagy committed
250 251 252
        query.set_target(self.host.hostname + ".circle")
        query.set_format("json")
        query.set_relative_start(5, "minutes")
253

254
        metrics = ["cpu.usage", "memory.usage"]
255
        for metric in metrics:
Gregory Nagy committed
256
            query.set_metric(metric)
257 258 259
            query.generate()
            handler.put(query)
            handler.send()
260 261

        collected = {}
262
        for metric in metrics:
Gregory Nagy committed
263
            response = handler.pop()
264 265 266 267
            try:
                cache = response[0]["datapoints"][-1][0]
            except (IndexError, KeyError):
                cache = 0
268 269 270 271 272
            if cache is None:
                cache = 0
            collected[metric] = cache
        return collected

273
    @property
274
    @node_available
275
    def cpu_usage(self):
276
        return float(self.get_monitor_info()["cpu.usage"]) / 100
277

278
    @property
279
    @node_available
280
    def ram_usage(self):
281
        return float(self.get_monitor_info()["memory.usage"]) / 100
282

283
    @property
284
    @node_available
285 286 287
    def byte_ram_usage(self):
        return self.ram_usage * self.ram_size

288 289 290 291 292 293 294 295
    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')

296 297
    def get_status_label(self):
        return {
298 299 300
            'OFFLINE': 'label-warning',
            'DISABLED': 'label-warning',
            'MISSING': 'label-danger',
301 302 303
            'ONLINE': 'label-success'}.get(self.get_state(),
                                           'label-danger')

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

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

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

    @classmethod
    def get_state_count(cls, online, enabled):
349 350
        return len([1 for i in
                    cls.objects.filter(enabled=enabled).select_related('host')
351
                    if i.online == online])
352 353 354 355

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