node.py 10.7 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
from firewall.models import Host
Őry Máté committed
16
from ..tasks import vm_tasks, local_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 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 54
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
55 56 57
    def __unicode__(self):
        return self.name

Őry Máté committed
58
    @method_cache(10, 5)
59
    def get_online(self):
60
        """Check if the node is online.
Őry Máté committed
61

62 63 64 65 66 67
        Runs a remote ping task if the worker is running.
        """
        try:
            return self.remote_query(vm_tasks.ping, timeout=1, default=False)
        except WorkerNotFound:
            return False
Őry Máté committed
68

69 70
    online = property(get_online)

Őry Máté committed
71
    @method_cache(300)
72
    def get_num_cores(self):
Őry Máté committed
73 74
        """Number of CPU threads available to the virtual machines.
        """
75

76
        return self.remote_query(vm_tasks.get_core_num, default=0)
77

78 79
    num_cores = property(get_num_cores)

80 81
    @property
    def state(self):
82
        """The state combined of online and enabled attributes.
83
        """
84
        if self.enabled and self.online:
85
            return 'ONLINE'
86
        elif self.enabled and not self.online:
87
            return 'MISSING'
88
        elif not self.enabled and self.online:
89
            return 'DISABLED'
90
        else:
91
            return 'OFFLINE'
92

93
    def disable(self, user=None, base_activity=None):
94
        ''' Disable the node.'''
95
        if self.enabled:
96 97 98 99 100
            if base_activity:
                act_ctx = base_activity.sub_activity('disable')
            else:
                act_ctx = node_activity('disable', node=self, user=user)
            with act_ctx:
101 102
                self.enabled = False
                self.save()
103

Őry Máté committed
104
    def flush(self, user=None, task_uuid=None):
Őry Máté committed
105 106
        """Disable node and move all instances to other ones.
        """
Őry Máté committed
107 108
        with node_activity('flush', node=self, user=user,
                           task_uuid=task_uuid) as act:
Őry Máté committed
109 110 111 112 113 114 115 116 117 118 119
            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")

120 121
    def enable(self, user=None):
        ''' Enable the node. '''
122 123 124 125 126 127
        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
128 129

    @method_cache(300)
130
    def get_ram_size(self):
Őry Máté committed
131 132
        """Bytes of total memory in the node.
        """
133
        return self.remote_query(vm_tasks.get_ram_size, default=0)
Őry Máté committed
134

135 136
    ram_size = property(get_ram_size)

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

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

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

151
        if vm_tasks.check_queue(self.host.hostname, queue_id):
152
            self.node_online()
153 154
            return self.host.hostname + "." + queue_id
        else:
155
            if self.enabled:
156
                self.node_offline()
157
            raise WorkerNotFound()
Őry Máté committed
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 202 203
    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?

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

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

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

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

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

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

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

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

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

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

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

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

    @classmethod
    def get_state_count(cls, online, enabled):
307 308
        return len([1 for i in cls.objects.filter(enabled=enabled).all()
                    if i.online == online])
309 310 311 312

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