node.py 11.4 KB
Newer Older
1
from __future__ import absolute_import, unicode_literals
Őry Máté committed
2 3 4
from logging import getLogger

from django.db.models import (
5
    CharField, IntegerField, ForeignKey, BooleanField, ManyToManyField,
6
    FloatField, permalink,
Őry Máté committed
7 8 9
)
from django.utils.translation import ugettext_lazy as _

10 11
from warnings import warn

Őry Máté committed
12 13 14 15
from celery.exceptions import TimeoutError
from model_utils.models import TimeStampedModel
from taggit.managers import TaggableManager

16
from common.models import method_cache, WorkerNotFound, HumanSortField
Őry Máté committed
17
from firewall.models import Host
Őry Máté committed
18
from ..tasks import vm_tasks, local_tasks
19
from .common import Trait
Őry Máté committed
20

21
from .activity import node_activity, NodeActivity
22

Gregory Nagy committed
23 24
from monitor.calvin.calvin import Query
from monitor.calvin.calvin import GraphiteHandler
25
from django.utils import timezone
26

Őry Máté committed
27 28 29
logger = getLogger(__name__)


30 31 32 33
def node_available(function):
    """Decorate methods to ignore disabled Nodes.
    """
    def decorate(self, *args, **kwargs):
34
        if self.enabled and self.online:
35 36 37 38 39 40
            return function(self, *args, **kwargs)
        else:
            return None
    return decorate


Őry Máté committed
41 42 43 44 45 46 47
class Node(TimeStampedModel):

    """A VM host machine, a hypervisor.
    """
    name = CharField(max_length=50, unique=True,
                     verbose_name=_('name'),
                     help_text=_('Human readable name of node.'))
48
    normalized_name = HumanSortField(monitor='name', max_length=100)
Őry Máté committed
49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67
    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 = ()
68
        ordering = ('-enabled', 'normalized_name')
Őry Máté committed
69

Őry Máté committed
70 71 72
    def __unicode__(self):
        return self.name

73
    @method_cache(10)
74
    def get_online(self):
75
        """Check if the node is online.
Őry Máté committed
76

77
        Check if node is online by queue is available.
78 79
        """
        try:
80 81
            self.get_remote_queue_name("vm")
        except:
82
            return False
83 84
        else:
            return True
Őry Máté committed
85

86 87
    online = property(get_online)

88
    @node_available
Őry Máté committed
89
    @method_cache(300)
90 91
    def get_info(self):
        return self.remote_query(vm_tasks.get_info,
92 93 94
                                 default={'core_num': '',
                                          'ram_size': '0',
                                          'architecture': ''})
95

96
    info = property(get_info)
97

98 99 100 101 102 103 104 105 106 107
    @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']

108 109 110 111 112
    STATES = {False: {False: ('OFFLINE', _('offline')),
                      True: ('DISABLED', _('disabled'))},
              True: {False: ('MISSING', _('missing')),
                     True: ('ONLINE', _('online'))}}

Őry Máté committed
113
    def get_state(self):
114
        """The state combined of online and enabled attributes.
115
        """
116 117
        return self.STATES[self.enabled][self.online][0]

Őry Máté committed
118 119
    state = property(get_state)

120 121
    def get_status_display(self):
        return self.STATES[self.enabled][self.online][1]
122

123
    def disable(self, user=None, base_activity=None):
124
        ''' Disable the node.'''
125
        if self.enabled:
126 127 128 129 130
            if base_activity:
                act_ctx = base_activity.sub_activity('disable')
            else:
                act_ctx = node_activity('disable', node=self, user=user)
            with act_ctx:
131 132
                self.enabled = False
                self.save()
133

Őry Máté committed
134
    def flush(self, user=None, task_uuid=None):
Őry Máté committed
135 136
        """Disable node and move all instances to other ones.
        """
Őry Máté committed
137 138
        with node_activity('flush', node=self, user=user,
                           task_uuid=task_uuid) as act:
Őry Máté committed
139 140 141 142 143 144 145 146 147 148 149
            self.disable(user, act)
            for i in self.instance_set.all():
                with act.sub_activity('migrate_instance_%d' % i.pk):
                    i.migrate()

    def flush_async(self, user=None):
        """Execute flush asynchronously.
        """
        return local_tasks.flush.apply_async(args=[self, user],
                                             queue="localhost.man")

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):
167 168
        """Return the name of the remote celery queue for this node.

169
        throws Exception if there is no worker on the queue.
170
        Until the cache provide reult there can be dead queues.
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 192 193 194 195 196 197 198
    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"):
                act = NodeActivity.create(code_suffix='monitor_succes_online',
                                          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
    @node_available
Őry Máté committed
289
    def update_vm_states(self):
290 291 292 293 294
        """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
295
        domains = {}
296 297 298 299 300
        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
301 302 303 304 305 306 307 308
            # [{'name': 'cloud-1234', 'state': 'RUNNING', ...}, ...]
            try:
                id = int(i['name'].split('-')[1])
            except:
                pass  # name format doesn't match
            else:
                domains[id] = i['state']

309 310
        instances = [{'id': i.id, 'state': i.state}
                     for i in self.instance_set.order_by('id').all()]
Őry Máté committed
311 312 313 314 315 316
        for i in instances:
            try:
                d = domains[i['id']]
            except KeyError:
                logger.info('Node %s update: instance %s missing from '
                            'libvirt', self, i['id'])
317 318
                # Set state to STOPPED when instance is missing
                self.instance_set.get(id=i['id']).vm_state_changed('STOPPED')
Őry Máté committed
319 320 321 322 323
            else:
                if d != i['state']:
                    logger.info('Node %s update: instance %s state changed '
                                '(libvirt: %s, db: %s)',
                                self, i['id'], d, i['state'])
324
                    self.instance_set.get(id=i['id']).vm_state_changed(d)
Őry Máté committed
325 326 327 328 329

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

    @classmethod
    def get_state_count(cls, online, enabled):
333 334
        return len([1 for i in
                    cls.objects.filter(enabled=enabled).select_related('host')
335
                    if i.online == online])
336 337 338 339

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