node.py 13.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
Bach Dániel committed
21
import requests
Őry Máté committed
22

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

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

35
from common.models import method_cache, WorkerNotFound, HumanSortField
36
from common.operations import OperatedMixin
Őry Máté committed
37
from firewall.models import Host
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
            if base_activity:
144 145
                act_ctx = base_activity.sub_activity(
                    'disable', readable_name=ugettext_noop("disable node"))
146
            else:
147 148 149
                act_ctx = node_activity(
                    'disable', node=self, user=user,
                    readable_name=ugettext_noop("disable node"))
150
            with act_ctx:
151 152
                self.enabled = False
                self.save()
153

154
    def enable(self, user=None, base_activity=None):
155
        ''' Enable the node. '''
156
        if self.enabled is not True:
157 158 159 160 161
            if base_activity:
                act_ctx = base_activity.sub_activity('enable')
            else:
                act_ctx = node_activity('enable', node=self, user=user)
            with act_ctx:
162 163
                self.enabled = True
                self.save()
Bach Dániel committed
164
            self.get_info(invalidate_cache=True)
Őry Máté committed
165 166

    @property
167
    @node_available
Őry Máté committed
168 169 170 171 172
    def ram_size_with_overcommit(self):
        """Bytes of total memory including overcommit margin.
        """
        return self.ram_size * self.overcommit

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

Dudás Ádám committed
177 178
        Throws Exception if there is no worker on the queue.
        The result may include dead queues because of caching.
179
        """
180

181 182 183 184
        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
185
            self.node_online()
186
            return queue_name
187
        else:
188
            if self.enabled:
189
                self.node_offline()
190
            raise WorkerNotFound()
Őry Máté committed
191

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

    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?

236 237
    def remote_query(self, task, timeout=30, priority=None, raise_=False,
                     default=None):
Őry Máté committed
238 239
        """Query the given task, and get the result.

240 241 242 243
        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
244
        try:
245
            r = task.apply_async(
246 247
                queue=self.get_remote_queue_name('vm', priority),
                expires=timeout + 60)
Őry Máté committed
248
            return r.get(timeout=timeout)
249
        except (TimeoutError, WorkerNotFound):
Őry Máté committed
250 251 252 253 254
            if raise_:
                raise
            else:
                return default

Bach Dániel committed
255
    @property
256
    @node_available
Bach Dániel committed
257
    @method_cache(10)
Bach Dániel committed
258 259 260 261 262 263 264 265
    def monitor_info(self):
        metrics = ('cpu.usage', 'memory.usage')
        prefix = 'circle.%s.' % self.name
        params = [('target', '%s%s' % (prefix, metric))
                  for metric in metrics]
        params.append(('from', '-5min'))
        params.append(('format', 'json'))

266
        try:
Bach Dániel committed
267 268 269 270 271 272 273 274 275 276 277 278
            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):]
279 280
                    else:
                        continue
Bach Dániel committed
281 282 283 284 285 286 287 288
                    value = target['datapoints'][-2][0]
                    retval[metric] = float(value)
                except (KeyError, IndexError, ValueError):
                    continue

            return retval
        except:
            logger.exception('Unhandled exception: ')
289 290
            return self.remote_query(vm_tasks.get_node_metrics, timeout=30,
                                     priority="fast")
291

292
    @property
293
    @node_available
294
    def cpu_usage(self):
Bach Dániel committed
295
        return self.monitor_info.get('cpu.usage') / 100
296

297
    @property
298
    @node_available
299
    def ram_usage(self):
Bach Dániel committed
300
        return self.monitor_info.get('memory.usage') / 100
301

302
    @property
303
    @node_available
304 305 306
    def byte_ram_usage(self):
        return self.ram_usage * self.ram_size

307 308
    def get_status_icon(self):
        return {
309
            'OFFLINE': 'fa-minus-circle',
Kálmán Viktor committed
310
            'DISABLED': 'fa-moon-o',
311 312 313
            'MISSING': 'fa-warning',
            'ONLINE': 'fa-play-circle'}.get(self.get_state(),
                                            'fa-question-circle')
314

315 316
    def get_status_label(self):
        return {
317 318 319
            'OFFLINE': 'label-warning',
            'DISABLED': 'label-warning',
            'MISSING': 'label-danger',
320 321 322
            'ONLINE': 'label-success'}.get(self.get_state(),
                                           'label-danger')

323
    @node_available
Őry Máté committed
324
    def update_vm_states(self):
325 326 327 328 329
        """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
330
        domains = {}
331 332
        domain_list = self.remote_query(vm_tasks.list_domains_info, timeout=5,
                                        priority="fast")
333 334 335 336
        if domain_list is None:
            logger.info("Monitoring failed at: %s", self.name)
            return
        for i in domain_list:
Őry Máté committed
337 338 339 340 341 342 343 344
            # [{'name': 'cloud-1234', 'state': 'RUNNING', ...}, ...]
            try:
                id = int(i['name'].split('-')[1])
            except:
                pass  # name format doesn't match
            else:
                domains[id] = i['state']

345 346
        instances = [{'id': i.id, 'state': i.state}
                     for i in self.instance_set.order_by('id').all()]
Őry Máté committed
347 348 349 350 351 352
        for i in instances:
            try:
                d = domains[i['id']]
            except KeyError:
                logger.info('Node %s update: instance %s missing from '
                            'libvirt', self, i['id'])
353 354
                # Set state to STOPPED when instance is missing
                self.instance_set.get(id=i['id']).vm_state_changed('STOPPED')
Őry Máté committed
355 356 357 358 359
            else:
                if d != i['state']:
                    logger.info('Node %s update: instance %s state changed '
                                '(libvirt: %s, db: %s)',
                                self, i['id'], d, i['state'])
360
                    self.instance_set.get(id=i['id']).vm_state_changed(d)
Őry Máté committed
361 362 363 364 365

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

    @classmethod
    def get_state_count(cls, online, enabled):
369 370
        return len([1 for i in
                    cls.objects.filter(enabled=enabled).select_related('host')
371
                    if i.online == online])
372 373 374 375

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