node.py 7.13 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 10 11 12 13
)
from django.utils.translation import ugettext_lazy as _

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

14
from common.models import method_cache, WorkerNotFound
Őry Máté committed
15 16
from firewall.models import Host
from ..tasks import vm_tasks
17
from .common import Trait
Őry Máté committed
18

19 20
from .activity import node_activity

Gregory Nagy committed
21 22
from monitor.calvin.calvin import Query
from monitor.calvin.calvin import GraphiteHandler
23

Őry Máté committed
24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53
logger = getLogger(__name__)


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.'))
    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 = ()

Őry Máté committed
54 55 56
    def __unicode__(self):
        return self.name

Őry Máté committed
57 58 59 60
    @property
    @method_cache(10, 5)
    def online(self):

Őry Máté committed
61
        return self.remote_query(vm_tasks.ping, timeout=1, default=False)
Őry Máté committed
62 63 64 65

    @property
    @method_cache(300)
    def num_cores(self):
Őry Máté committed
66 67
        """Number of CPU threads available to the virtual machines.
        """
68

Őry Máté committed
69
        return self.remote_query(vm_tasks.get_core_num)
70

71 72 73 74 75
    @property
    def state(self):
        """Node state.
        """

76
        if self.enabled and self.online:
77
            return 'Online'
78
        elif self.enabled and not self.online:
79
            return 'Missing'
80
        elif not self.enabled and self.online:
81
            return 'Disabled'
82
        else:
83 84 85 86 87 88 89 90 91 92 93 94 95
            return 'Offline'

    def disable(self, user=None):
        ''' Disable the node.'''
        with node_activity(code_suffix='disable', node=self, user=user):
            self.enabled = False
            self.save()

    def enable(self, user=None):
        ''' Enable the node. '''
        with node_activity(code_suffix='enable', node=self, user=user):
            self.enabled = True
            self.save()
Őry Máté committed
96 97 98 99

    @property
    @method_cache(300)
    def ram_size(self):
Őry Máté committed
100 101
        """Bytes of total memory in the node.
        """
Őry Máté committed
102 103 104 105 106 107 108 109 110

        return self.remote_query(vm_tasks.get_ram_size)

    @property
    def ram_size_with_overcommit(self):
        """Bytes of total memory including overcommit margin.
        """
        return self.ram_size * self.overcommit

111
    @method_cache(30)
Őry Máté committed
112
    def get_remote_queue_name(self, queue_id):
113 114 115 116 117 118 119
        """ Return the remote queue name
        throws Exception if there is no worker on the queue.
        Until the cache provide reult there can be dead quques.
        """
        if vm_tasks.check_queue(self.host.hostname, queue_id):
            return self.host.hostname + "." + queue_id
        else:
120
            raise WorkerNotFound()
Őry Máté committed
121

122
    def remote_query(self, task, timeout=30, raise_=False, default=None):
Őry Máté committed
123 124 125 126 127 128 129 130 131 132 133 134 135 136
        """Query the given task, and get the result.

        If the result is not ready in timeout secs, return default value or
        raise a TimeoutError."""
        r = task.apply_async(
            queue=self.get_remote_queue_name('vm'), expires=timeout + 60)
        try:
            return r.get(timeout=timeout)
        except TimeoutError:
            if raise_:
                raise
            else:
                return default

137
    def get_monitor_info(self):
138 139 140 141
        collected = {}
        try:
            handler = GraphiteHandler()
        except:
Gregory Nagy committed
142
            response = self.remote_query(vm_tasks.get_node_metrics, 30)
143 144
            collected['cpu.usage'] = response['cpu.usage']
            collected['memory.usage'] = response['memory.usage']
145
            return collected
146
        query = Query()
Gregory Nagy committed
147 148 149
        query.set_target(self.host.hostname + ".circle")
        query.set_format("json")
        query.set_relative_start(5, "minutes")
150
        metrics = ["cpu.usage", "memory.usage"]
151 152
        collected = {}
        for metric in metrics:
Gregory Nagy committed
153
            query.set_metric(metric)
154 155 156 157
            query.generate()
            handler.put(query)
            handler.send()
        for metric in metrics:
Gregory Nagy committed
158
            response = handler.pop()
159 160 161 162 163 164 165
            length = len(response[0]["datapoints"])
            cache = response[0]["datapoints"][length - 1][0]
            if cache is None:
                cache = 0
            collected[metric] = cache
        return collected

166
    @property
167
    def cpu_usage(self):
168
        return float(self.get_monitor_info()["cpu.usage"]) / 100
169

170
    @property
171
    def ram_usage(self):
172
        return float(self.get_monitor_info()["memory.usage"]) / 100
173

Őry Máté committed
174 175
    def update_vm_states(self):
        domains = {}
176 177 178 179 180
        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
181 182 183 184 185 186 187 188
            # [{'name': 'cloud-1234', 'state': 'RUNNING', ...}, ...]
            try:
                id = int(i['name'].split('-')[1])
            except:
                pass  # name format doesn't match
            else:
                domains[id] = i['state']

189 190
        instances = [{'id': i.id, 'state': i.state}
                     for i in self.instance_set.order_by('id').all()]
Őry Máté committed
191 192 193 194 195 196 197 198 199 200 201
        for i in instances:
            try:
                d = domains[i['id']]
            except KeyError:
                logger.info('Node %s update: instance %s missing from '
                            'libvirt', self, i['id'])
            else:
                if d != i['state']:
                    logger.info('Node %s update: instance %s state changed '
                                '(libvirt: %s, db: %s)',
                                self, i['id'], d, i['state'])
202
                    self.instance_set.get(id=i['id']).vm_state_changed(d)
Őry Máté committed
203 204 205 206 207

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

    @classmethod
    def get_state_count(cls, online, enabled):
211 212
        return len([1 for i in cls.objects.filter(enabled=enabled).all()
                    if i.online == online])
213 214 215 216

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