node.py 10.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 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
from .activity import node_activity, NodeActivity
20

Gregory Nagy committed
21 22
from monitor.calvin.calvin import Query
from monitor.calvin.calvin import GraphiteHandler
23
from django.utils import timezone
24

Őry Máté committed
25 26 27
logger = getLogger(__name__)


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


Őry Máté committed
39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65
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
66 67 68
    def __unicode__(self):
        return self.name

69
    @method_cache(10)
70
    def get_online(self):
71
        """Check if the node is online.
Őry Máté committed
72

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

82 83
    online = property(get_online)

84
    @node_available
Őry Máté committed
85
    @method_cache(300)
86
    def get_num_cores(self):
Őry Máté committed
87 88
        """Number of CPU threads available to the virtual machines.
        """
89

90
        return self.remote_query(vm_tasks.get_core_num, default=0)
91

92 93
    num_cores = property(get_num_cores)

94 95 96 97 98
    STATES = {False: {False: ('OFFLINE', _('offline')),
                      True: ('DISABLED', _('disabled'))},
              True: {False: ('MISSING', _('missing')),
                     True: ('ONLINE', _('online'))}}

Őry Máté committed
99
    def get_state(self):
100
        """The state combined of online and enabled attributes.
101
        """
102 103
        return self.STATES[self.enabled][self.online][0]

Őry Máté committed
104 105
    state = property(get_state)

106 107
    def get_status_display(self):
        return self.STATES[self.enabled][self.online][1]
108 109 110

    def disable(self, user=None):
        ''' Disable the node.'''
111
        if self.enabled:
112 113 114
            with node_activity(code_suffix='disable', node=self, user=user):
                self.enabled = False
                self.save()
115 116 117

    def enable(self, user=None):
        ''' Enable the node. '''
118 119 120 121 122 123
        if self.enabled is not True:
            with node_activity(code_suffix='enable', node=self, user=user):
                self.enabled = True
                self.save()
            self.get_num_cores(invalidate_cache=True)
            self.get_ram_size(invalidate_cache=True)
Őry Máté committed
124

125
    @node_available
Őry Máté committed
126
    @method_cache(300)
127
    def get_ram_size(self):
Őry Máté committed
128 129
        """Bytes of total memory in the node.
        """
130
        return self.remote_query(vm_tasks.get_ram_size, default=0)
Őry Máté committed
131

132 133
    ram_size = property(get_ram_size)

Őry Máté committed
134
    @property
135
    @node_available
Őry Máté committed
136 137 138 139 140
    def ram_size_with_overcommit(self):
        """Bytes of total memory including overcommit margin.
        """
        return self.ram_size * self.overcommit

141
    @method_cache(30)
Őry Máté committed
142
    def get_remote_queue_name(self, queue_id):
143 144
        """Return the name of the remote celery queue for this node.

145
        throws Exception if there is no worker on the queue.
146
        Until the cache provide reult there can be dead queues.
147
        """
148

149
        if vm_tasks.check_queue(self.host.hostname, queue_id):
150
            self.node_online()
151 152
            return self.host.hostname + "." + queue_id
        else:
153
            if self.enabled:
154
                self.node_offline()
155
            raise WorkerNotFound()
Őry Máté committed
156

157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201
    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)
                self.get_num_cores(invalidate_cache=True)
                self.get_ram_size(invalidate_cache=True)

    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?

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

205 206 207 208
        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
209
        try:
210 211
            r = task.apply_async(
                queue=self.get_remote_queue_name('vm'), expires=timeout + 60)
Őry Máté committed
212
            return r.get(timeout=timeout)
213
        except (TimeoutError, WorkerNotFound):
Őry Máté committed
214 215 216 217 218
            if raise_:
                raise
            else:
                return default

219
    @node_available
220
    def get_monitor_info(self):
221 222
        try:
            handler = GraphiteHandler()
223 224 225
        except RuntimeError:
            return self.remote_query(vm_tasks.get_node_metrics, 30)

226
        query = Query()
Gregory Nagy committed
227 228 229
        query.set_target(self.host.hostname + ".circle")
        query.set_format("json")
        query.set_relative_start(5, "minutes")
230

231
        metrics = ["cpu.usage", "memory.usage"]
232
        for metric in metrics:
Gregory Nagy committed
233
            query.set_metric(metric)
234 235 236
            query.generate()
            handler.put(query)
            handler.send()
237 238

        collected = {}
239
        for metric in metrics:
Gregory Nagy committed
240
            response = handler.pop()
241 242 243 244
            try:
                cache = response[0]["datapoints"][-1][0]
            except (IndexError, KeyError):
                cache = 0
245 246 247 248 249
            if cache is None:
                cache = 0
            collected[metric] = cache
        return collected

250
    @property
251
    @node_available
252
    def cpu_usage(self):
253
        return float(self.get_monitor_info()["cpu.usage"]) / 100
254

255
    @property
256
    @node_available
257
    def ram_usage(self):
258
        return float(self.get_monitor_info()["memory.usage"]) / 100
259

260
    @property
261
    @node_available
262 263 264
    def byte_ram_usage(self):
        return self.ram_usage * self.ram_size

265
    @node_available
Őry Máté committed
266
    def update_vm_states(self):
267 268 269 270 271
        """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
272
        domains = {}
273 274 275 276 277
        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
278 279 280 281 282 283 284 285
            # [{'name': 'cloud-1234', 'state': 'RUNNING', ...}, ...]
            try:
                id = int(i['name'].split('-')[1])
            except:
                pass  # name format doesn't match
            else:
                domains[id] = i['state']

286 287
        instances = [{'id': i.id, 'state': i.state}
                     for i in self.instance_set.order_by('id').all()]
Őry Máté committed
288 289 290 291 292 293
        for i in instances:
            try:
                d = domains[i['id']]
            except KeyError:
                logger.info('Node %s update: instance %s missing from '
                            'libvirt', self, i['id'])
294 295
                # Set state to STOPPED when instance is missing
                self.instance_set.get(id=i['id']).vm_state_changed('STOPPED')
Őry Máté committed
296 297 298 299 300
            else:
                if d != i['state']:
                    logger.info('Node %s update: instance %s state changed '
                                '(libvirt: %s, db: %s)',
                                self, i['id'], d, i['state'])
301
                    self.instance_set.get(id=i['id']).vm_state_changed(d)
Őry Máté committed
302 303 304 305 306

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

    @classmethod
    def get_state_count(cls, online, enabled):
310 311
        return len([1 for i in
                    cls.objects.filter(enabled=enabled).select_related('host')
312
                    if i.online == online])
313 314 315 316

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