Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / logic / backend.py @ 3af1fb4b

History | View | Annotate | Download (42.1 kB)

1
# Copyright 2011-2013 GRNET S.A. All rights reserved.
2
#
3
# Redistribution and use in source and binary forms, with or
4
# without modification, are permitted provided that the following
5
# conditions are met:
6
#
7
#   1. Redistributions of source code must retain the above
8
#      copyright notice, this list of conditions and the following
9
#      disclaimer.
10
#
11
#   2. Redistributions in binary form must reproduce the above
12
#      copyright notice, this list of conditions and the following
13
#      disclaimer in the documentation and/or other materials
14
#      provided with the distribution.
15
#
16
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
17
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
18
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
19
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
20
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
23
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
24
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
26
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27
# POSSIBILITY OF SUCH DAMAGE.
28
#
29
# The views and conclusions contained in the software and
30
# documentation are those of the authors and should not be
31
# interpreted as representing official policies, either expressed
32
# or implied, of GRNET S.A.
33
from django.conf import settings
34
from django.db import transaction
35
from datetime import datetime, timedelta
36

    
37
from synnefo.db.models import (Backend, VirtualMachine, Network,
38
                               BackendNetwork, BACKEND_STATUSES,
39
                               pooled_rapi_client, VirtualMachineDiagnostic,
40
                               Flavor, IPAddress, IPAddressLog)
41
from synnefo.logic import utils, ips
42
from synnefo import quotas
43
from synnefo.api.util import release_resource
44
from synnefo.util.mac2eui64 import mac2eui64
45
from synnefo.logic import rapi
46

    
47
from logging import getLogger
48
log = getLogger(__name__)
49

    
50

    
51
_firewall_tags = {
52
    'ENABLED': settings.GANETI_FIREWALL_ENABLED_TAG,
53
    'DISABLED': settings.GANETI_FIREWALL_DISABLED_TAG,
54
    'PROTECTED': settings.GANETI_FIREWALL_PROTECTED_TAG}
55

    
56
_reverse_tags = dict((v.split(':')[3], k) for k, v in _firewall_tags.items())
57

    
58
SIMPLE_NIC_FIELDS = ["state", "mac", "network", "firewall_profile", "index"]
59
COMPLEX_NIC_FIELDS = ["ipv4_address", "ipv6_address"]
60
NIC_FIELDS = SIMPLE_NIC_FIELDS + COMPLEX_NIC_FIELDS
61
UNKNOWN_NIC_PREFIX = "unknown-"
62

    
63

    
64
def handle_vm_quotas(vm, job_id, job_opcode, job_status, job_fields):
65
    """Handle quotas for updated VirtualMachine.
66

67
    Update quotas for the updated VirtualMachine based on the job that run on
68
    the Ganeti backend. If a commission has been already issued for this job,
69
    then this commission is just accepted or rejected based on the job status.
70
    Otherwise, a new commission for the given change is issued, that is also in
71
    force and auto-accept mode. In this case, previous commissions are
72
    rejected, since they reflect a previous state of the VM.
73

74
    """
75
    if job_status not in rapi.JOB_STATUS_FINALIZED:
76
        return vm
77

    
78
    # Check successful completion of a job will trigger any quotable change in
79
    # the VM state.
80
    action = utils.get_action_from_opcode(job_opcode, job_fields)
81
    if action == "BUILD":
82
        # Quotas for new VMs are automatically accepted by the API
83
        return vm
84

    
85
    if vm.task_job_id == job_id and vm.serial is not None:
86
        # Commission for this change has already been issued. So just
87
        # accept/reject it. Special case is OP_INSTANCE_CREATE, which even
88
        # if fails, must be accepted, as the user must manually remove the
89
        # failed server
90
        serial = vm.serial
91
        if job_status == rapi.JOB_STATUS_SUCCESS:
92
            quotas.accept_serial(serial)
93
        elif job_status in [rapi.JOB_STATUS_ERROR, rapi.JOB_STATUS_CANCELED]:
94
            log.debug("Job %s failed. Rejecting related serial %s", job_id,
95
                      serial)
96
            quotas.reject_serial(serial)
97
        vm.serial = None
98
    elif job_status == rapi.JOB_STATUS_SUCCESS:
99
        commission_info = quotas.get_commission_info(resource=vm,
100
                                                     action=action,
101
                                                     action_fields=job_fields)
102
        if commission_info is not None:
103
            # Commission for this change has not been issued, or the issued
104
            # commission was unaware of the current change. Reject all previous
105
            # commissions and create a new one in forced mode!
106
            log.debug("Expected job was %s. Processing job %s.",
107
                      vm.task_job_id, job_id)
108
            reason = ("client: dispatcher, resource: %s, ganeti_job: %s"
109
                      % (vm, job_id))
110
            quotas.handle_resource_commission(vm, action,
111
                                              action_fields=job_fields,
112
                                              commission_name=reason,
113
                                              force=True,
114
                                              auto_accept=True)
115
            log.debug("Issued new commission: %s", vm.serial)
116

    
117
    return vm
118

    
119

    
120
@transaction.commit_on_success
121
def process_op_status(vm, etime, jobid, opcode, status, logmsg, nics=None,
122
                      job_fields=None):
123
    """Process a job progress notification from the backend
124

125
    Process an incoming message from the backend (currently Ganeti).
126
    Job notifications with a terminating status (sucess, error, or canceled),
127
    also update the operating state of the VM.
128

129
    """
130
    # See #1492, #1031, #1111 why this line has been removed
131
    #if (opcode not in [x[0] for x in VirtualMachine.BACKEND_OPCODES] or
132
    if status not in [x[0] for x in BACKEND_STATUSES]:
133
        raise VirtualMachine.InvalidBackendMsgError(opcode, status)
134

    
135
    vm.backendjobid = jobid
136
    vm.backendjobstatus = status
137
    vm.backendopcode = opcode
138
    vm.backendlogmsg = logmsg
139

    
140
    if status not in rapi.JOB_STATUS_FINALIZED:
141
        vm.save()
142
        return
143

    
144
    if job_fields is None:
145
        job_fields = {}
146

    
147
    new_operstate = None
148
    new_flavor = None
149
    state_for_success = VirtualMachine.OPER_STATE_FROM_OPCODE.get(opcode)
150

    
151
    if status == rapi.JOB_STATUS_SUCCESS:
152
        # If job succeeds, change operating state if needed
153
        if state_for_success is not None:
154
            new_operstate = state_for_success
155

    
156
        beparams = job_fields.get("beparams", None)
157
        if beparams:
158
            # Change the flavor of the VM
159
            new_flavor = _process_resize(vm, beparams)
160

    
161
        # Update backendtime only for jobs that have been successfully
162
        # completed, since only these jobs update the state of the VM. Else a
163
        # "race condition" may occur when a successful job (e.g.
164
        # OP_INSTANCE_REMOVE) completes before an error job and messages arrive
165
        # in reversed order.
166
        vm.backendtime = etime
167

    
168
    if status in rapi.JOB_STATUS_FINALIZED and nics is not None:
169
        # Update the NICs of the VM
170
        _process_net_status(vm, etime, nics)
171

    
172
    # Special case: if OP_INSTANCE_CREATE fails --> ERROR
173
    if opcode == 'OP_INSTANCE_CREATE' and status in (rapi.JOB_STATUS_CANCELED,
174
                                                     rapi.JOB_STATUS_ERROR):
175
        new_operstate = "ERROR"
176
        vm.backendtime = etime
177
        # Update state of associated NICs
178
        vm.nics.all().update(state="ERROR")
179
    elif opcode == 'OP_INSTANCE_REMOVE':
180
        # Special case: OP_INSTANCE_REMOVE fails for machines in ERROR,
181
        # when no instance exists at the Ganeti backend.
182
        # See ticket #799 for all the details.
183
        if (status == rapi.JOB_STATUS_SUCCESS or
184
           (status == rapi.JOB_STATUS_ERROR and not vm_exists_in_backend(vm))):
185
            # VM has been deleted
186
            for nic in vm.nics.all():
187
                # Release the IP
188
                remove_nic_ips(nic)
189
                # And delete the NIC.
190
                nic.delete()
191
            vm.deleted = True
192
            new_operstate = state_for_success
193
            vm.backendtime = etime
194
            status = rapi.JOB_STATUS_SUCCESS
195

    
196
    if status in rapi.JOB_STATUS_FINALIZED:
197
        # Job is finalized: Handle quotas/commissioning
198
        vm = handle_vm_quotas(vm, job_id=jobid, job_opcode=opcode,
199
                              job_status=status, job_fields=job_fields)
200
        # and clear task fields
201
        if vm.task_job_id == jobid:
202
            vm.task = None
203
            vm.task_job_id = None
204

    
205
    if new_operstate is not None:
206
        vm.operstate = new_operstate
207
    if new_flavor is not None:
208
        vm.flavor = new_flavor
209

    
210
    vm.save()
211

    
212

    
213
def _process_resize(vm, beparams):
214
    """Change flavor of a VirtualMachine based on new beparams."""
215
    old_flavor = vm.flavor
216
    vcpus = beparams.get("vcpus", old_flavor.cpu)
217
    ram = beparams.get("maxmem", old_flavor.ram)
218
    if vcpus == old_flavor.cpu and ram == old_flavor.ram:
219
        return
220
    try:
221
        new_flavor = Flavor.objects.get(cpu=vcpus, ram=ram,
222
                                        disk=old_flavor.disk,
223
                                        disk_template=old_flavor.disk_template)
224
    except Flavor.DoesNotExist:
225
        raise Exception("Cannot find flavor for VM")
226
    return new_flavor
227

    
228

    
229
@transaction.commit_on_success
230
def process_net_status(vm, etime, nics):
231
    """Wrap _process_net_status inside transaction."""
232
    _process_net_status(vm, etime, nics)
233

    
234

    
235
def _process_net_status(vm, etime, nics):
236
    """Process a net status notification from the backend
237

238
    Process an incoming message from the Ganeti backend,
239
    detailing the NIC configuration of a VM instance.
240

241
    Update the state of the VM in the DB accordingly.
242

243
    """
244
    ganeti_nics = process_ganeti_nics(nics)
245
    db_nics = dict([(nic.id, nic)
246
                    for nic in vm.nics.select_related("network")
247
                                      .prefetch_related("ips")])
248

    
249
    for nic_name in set(db_nics.keys()) | set(ganeti_nics.keys()):
250
        db_nic = db_nics.get(nic_name)
251
        ganeti_nic = ganeti_nics.get(nic_name)
252
        if ganeti_nic is None:
253
            if nic_is_stale(vm, nic):
254
                log.debug("Removing stale NIC '%s'" % db_nic)
255
                remove_nic_ips(db_nic)
256
                db_nic.delete()
257
            else:
258
                log.info("NIC '%s' is still being created" % db_nic)
259
        elif db_nic is None:
260
            msg = ("NIC/%s of VM %s does not exist in DB! Cannot automatically"
261
                   " fix this issue!" % (nic_name, vm))
262
            log.error(msg)
263
            continue
264
        elif not nics_are_equal(db_nic, ganeti_nic):
265
            for f in SIMPLE_NIC_FIELDS:
266
                # Update the NIC in DB with the values from Ganeti NIC
267
                setattr(db_nic, f, ganeti_nic[f])
268
                db_nic.save()
269

    
270
            # Special case where the IPv4 address has changed, because you
271
            # need to release the old IPv4 address and reserve the new one
272
            gnt_ipv4_address = ganeti_nic["ipv4_address"]
273
            db_ipv4_address = db_nic.ipv4_address
274
            if db_ipv4_address != gnt_ipv4_address:
275
                change_address_of_port(db_nic, vm.userid,
276
                                       old_address=db_ipv4_address,
277
                                       new_address=gnt_ipv4_address,
278
                                       version=4)
279

    
280
            gnt_ipv6_address = ganeti_nic["ipv6_address"]
281
            db_ipv6_address = db_nic.ipv6_address
282
            if db_ipv6_address != gnt_ipv6_address:
283
                change_address_of_port(db_nic, vm.userid,
284
                                       old_address=db_ipv6_address,
285
                                       new_address=gnt_ipv6_address,
286
                                       version=6)
287

    
288
    vm.backendtime = etime
289
    vm.save()
290

    
291

    
292
def change_address_of_port(port, userid, old_address, new_address, version):
293
    """Change."""
294
    if old_address is not None:
295
        msg = ("IPv%s Address of server '%s' changed from '%s' to '%s'"
296
               % (version, port.machine_id, old_address, new_address))
297
        log.error(msg)
298

    
299
    # Remove the old IP address
300
    remove_nic_ips(port, version=version)
301

    
302
    if version == 4:
303
        ipaddress = ips.allocate_ip(port.network, userid, address=new_address)
304
        ipaddress.nic = port
305
        ipaddress.save()
306
    elif version == 6:
307
        subnet6 = port.network.subnet6
308
        ipaddress = IPAddress.objects.create(userid=userid,
309
                                             network=port.network,
310
                                             subnet=subnet6,
311
                                             nic=port,
312
                                             address=new_address,
313
                                             ipversion=6)
314
    else:
315
        raise ValueError("Unknown version: %s" % version)
316

    
317
    # New address log
318
    ip_log = IPAddressLog.objects.create(server_id=port.machine_id,
319
                                         network_id=port.network_id,
320
                                         address=new_address,
321
                                         active=True)
322
    log.info("Created IP log entry '%s' for address '%s' to server '%s'",
323
             ip_log.id, new_address, port.machine_id)
324

    
325
    return ipaddress
326

    
327

    
328
def nics_are_equal(db_nic, gnt_nic):
329
    for field in NIC_FIELDS:
330
        if getattr(db_nic, field) != gnt_nic[field]:
331
            return False
332
    return True
333

    
334

    
335
def process_ganeti_nics(ganeti_nics):
336
    """Process NIC dict from ganeti"""
337
    new_nics = []
338
    for index, gnic in enumerate(ganeti_nics):
339
        nic_name = gnic.get("name", None)
340
        if nic_name is not None:
341
            nic_id = utils.id_from_nic_name(nic_name)
342
        else:
343
            # Put as default value the index. If it is an unknown NIC to
344
            # synnefo it will be created automaticaly.
345
            nic_id = UNKNOWN_NIC_PREFIX + str(index)
346
        network_name = gnic.get('network', '')
347
        network_id = utils.id_from_network_name(network_name)
348
        network = Network.objects.get(id=network_id)
349

    
350
        # Get the new nic info
351
        mac = gnic.get('mac')
352
        ipv4 = gnic.get('ip')
353
        subnet6 = network.subnet6
354
        ipv6 = mac2eui64(mac, subnet6.cidr) if subnet6 else None
355

    
356
        firewall = gnic.get('firewall')
357
        firewall_profile = _reverse_tags.get(firewall)
358
        if not firewall_profile and network.public:
359
            firewall_profile = settings.DEFAULT_FIREWALL_PROFILE
360

    
361
        nic_info = {
362
            'index': index,
363
            'network': network,
364
            'mac': mac,
365
            'ipv4_address': ipv4,
366
            'ipv6_address': ipv6,
367
            'firewall_profile': firewall_profile,
368
            'state': 'ACTIVE'}
369

    
370
        new_nics.append((nic_id, nic_info))
371
    return dict(new_nics)
372

    
373

    
374
def remove_nic_ips(nic, version=None):
375
    """Remove IP addresses associated with a NetworkInterface.
376

377
    Remove all IP addresses that are associated with the NetworkInterface
378
    object, by returning them to the pool and deleting the IPAddress object. If
379
    the IP is a floating IP, then it is just disassociated from the NIC.
380
    If version is specified, then only IP addressses of that version will be
381
    removed.
382

383
    """
384
    for ip in nic.ips.all():
385
        if version and ip.ipversion != version:
386
            continue
387

    
388
        # Update the DB table holding the logging of all IP addresses
389
        terminate_active_ipaddress_log(nic, ip)
390

    
391
        if ip.floating_ip:
392
            ip.nic = None
393
            ip.save()
394
        else:
395
            # Release the IPv4 address
396
            ip.release_address()
397
            ip.delete()
398

    
399

    
400
def terminate_active_ipaddress_log(nic, ip):
401
    """Update DB logging entry for this IP address."""
402
    if not ip.network.public or nic.machine is None:
403
        return
404
    try:
405
        ip_log, created = \
406
            IPAddressLog.objects.get_or_create(server_id=nic.machine_id,
407
                                               network_id=ip.network_id,
408
                                               address=ip.address,
409
                                               active=True)
410
    except IPAddressLog.MultipleObjectsReturned:
411
        logmsg = ("Multiple active log entries for IP %s, Network %s,"
412
                  "Server %s. Cannot proceed!"
413
                  % (ip.address, ip.network, nic.machine))
414
        log.error(logmsg)
415
        raise
416

    
417
    if created:
418
        logmsg = ("No log entry for IP %s, Network %s, Server %s. Created new"
419
                  " but with wrong creation timestamp."
420
                  % (ip.address, ip.network, nic.machine))
421
        log.error(logmsg)
422
    ip_log.released_at = datetime.now()
423
    ip_log.active = False
424
    ip_log.save()
425

    
426

    
427
@transaction.commit_on_success
428
def process_network_status(back_network, etime, jobid, opcode, status, logmsg):
429
    if status not in [x[0] for x in BACKEND_STATUSES]:
430
        raise Network.InvalidBackendMsgError(opcode, status)
431

    
432
    back_network.backendjobid = jobid
433
    back_network.backendjobstatus = status
434
    back_network.backendopcode = opcode
435
    back_network.backendlogmsg = logmsg
436

    
437
    # Note: Network is already locked!
438
    network = back_network.network
439

    
440
    # Notifications of success change the operating state
441
    state_for_success = BackendNetwork.OPER_STATE_FROM_OPCODE.get(opcode, None)
442
    if status == rapi.JOB_STATUS_SUCCESS and state_for_success is not None:
443
        back_network.operstate = state_for_success
444

    
445
    if (status in (rapi.JOB_STATUS_CANCELED, rapi.JOB_STATUS_ERROR)
446
       and opcode == 'OP_NETWORK_ADD'):
447
        back_network.operstate = 'ERROR'
448
        back_network.backendtime = etime
449

    
450
    if opcode == 'OP_NETWORK_REMOVE':
451
        network_is_deleted = (status == rapi.JOB_STATUS_SUCCESS)
452
        if network_is_deleted or (status == rapi.JOB_STATUS_ERROR and not
453
                                  network_exists_in_backend(back_network)):
454
            back_network.operstate = state_for_success
455
            back_network.deleted = True
456
            back_network.backendtime = etime
457

    
458
    if status == rapi.JOB_STATUS_SUCCESS:
459
        back_network.backendtime = etime
460
    back_network.save()
461
    # Also you must update the state of the Network!!
462
    update_network_state(network)
463

    
464

    
465
def update_network_state(network):
466
    """Update the state of a Network based on BackendNetwork states.
467

468
    Update the state of a Network based on the operstate of the networks in the
469
    backends that network exists.
470

471
    The state of the network is:
472
    * ACTIVE: If it is 'ACTIVE' in at least one backend.
473
    * DELETED: If it is is 'DELETED' in all backends that have been created.
474

475
    This function also releases the resources (MAC prefix or Bridge) and the
476
    quotas for the network.
477

478
    """
479
    if network.deleted:
480
        # Network has already been deleted. Just assert that state is also
481
        # DELETED
482
        if not network.state == "DELETED":
483
            network.state = "DELETED"
484
            network.save()
485
        return
486

    
487
    backend_states = [s.operstate for s in network.backend_networks.all()]
488
    if not backend_states and network.action != "DESTROY":
489
        if network.state != "ACTIVE":
490
            network.state = "ACTIVE"
491
            network.save()
492
            return
493

    
494
    # Network is deleted when all BackendNetworks go to "DELETED" operstate
495
    deleted = reduce(lambda x, y: x == y and "DELETED", backend_states,
496
                     "DELETED")
497

    
498
    # Release the resources on the deletion of the Network
499
    if deleted:
500
        if network.ips.filter(deleted=False, floating_ip=True).exists():
501
            msg = "Cannot delete network %s! Floating IPs still in use!"
502
            log.error(msg % network)
503
            raise Exception(msg % network)
504
        log.info("Network %r deleted. Releasing link %r mac_prefix %r",
505
                 network.id, network.mac_prefix, network.link)
506
        network.deleted = True
507
        network.state = "DELETED"
508
        # Undrain the network, otherwise the network state will remain
509
        # as 'SNF:DRAINED'
510
        network.drained = False
511
        if network.mac_prefix:
512
            if network.FLAVORS[network.flavor]["mac_prefix"] == "pool":
513
                release_resource(res_type="mac_prefix",
514
                                 value=network.mac_prefix)
515
        if network.link:
516
            if network.FLAVORS[network.flavor]["link"] == "pool":
517
                release_resource(res_type="bridge", value=network.link)
518

    
519
        # Set all subnets as deleted
520
        network.subnets.update(deleted=True)
521
        # And delete the IP pools
522
        for subnet in network.subnets.all():
523
            if subnet.ipversion == 4:
524
                subnet.ip_pools.all().delete()
525
        # And all the backend networks since there are useless
526
        network.backend_networks.all().delete()
527

    
528
        # Issue commission
529
        if network.userid:
530
            quotas.issue_and_accept_commission(network, action="DESTROY")
531
            # the above has already saved the object and committed;
532
            # a second save would override others' changes, since the
533
            # object is now unlocked
534
            return
535
        elif not network.public:
536
            log.warning("Network %s does not have an owner!", network.id)
537
    network.save()
538

    
539

    
540
@transaction.commit_on_success
541
def process_network_modify(back_network, etime, jobid, opcode, status,
542
                           job_fields):
543
    assert (opcode == "OP_NETWORK_SET_PARAMS")
544
    if status not in [x[0] for x in BACKEND_STATUSES]:
545
        raise Network.InvalidBackendMsgError(opcode, status)
546

    
547
    back_network.backendjobid = jobid
548
    back_network.backendjobstatus = status
549
    back_network.opcode = opcode
550

    
551
    add_reserved_ips = job_fields.get("add_reserved_ips")
552
    if add_reserved_ips:
553
        network = back_network.network
554
        for ip in add_reserved_ips:
555
            network.reserve_address(ip, external=True)
556

    
557
    if status == rapi.JOB_STATUS_SUCCESS:
558
        back_network.backendtime = etime
559
    back_network.save()
560

    
561

    
562
@transaction.commit_on_success
563
def process_create_progress(vm, etime, progress):
564

    
565
    percentage = int(progress)
566

    
567
    # The percentage may exceed 100%, due to the way
568
    # snf-image:copy-progress tracks bytes read by image handling processes
569
    percentage = 100 if percentage > 100 else percentage
570
    if percentage < 0:
571
        raise ValueError("Percentage cannot be negative")
572

    
573
    # FIXME: log a warning here, see #1033
574
#   if last_update > percentage:
575
#       raise ValueError("Build percentage should increase monotonically " \
576
#                        "(old = %d, new = %d)" % (last_update, percentage))
577

    
578
    # This assumes that no message of type 'ganeti-create-progress' is going to
579
    # arrive once OP_INSTANCE_CREATE has succeeded for a Ganeti instance and
580
    # the instance is STARTED.  What if the two messages are processed by two
581
    # separate dispatcher threads, and the 'ganeti-op-status' message for
582
    # successful creation gets processed before the 'ganeti-create-progress'
583
    # message? [vkoukis]
584
    #
585
    #if not vm.operstate == 'BUILD':
586
    #    raise VirtualMachine.IllegalState("VM is not in building state")
587

    
588
    vm.buildpercentage = percentage
589
    vm.backendtime = etime
590
    vm.save()
591

    
592

    
593
@transaction.commit_on_success
594
def create_instance_diagnostic(vm, message, source, level="DEBUG", etime=None,
595
                               details=None):
596
    """
597
    Create virtual machine instance diagnostic entry.
598

599
    :param vm: VirtualMachine instance to create diagnostic for.
600
    :param message: Diagnostic message.
601
    :param source: Diagnostic source identifier (e.g. image-helper).
602
    :param level: Diagnostic level (`DEBUG`, `INFO`, `WARNING`, `ERROR`).
603
    :param etime: The time the message occured (if available).
604
    :param details: Additional details or debug information.
605
    """
606
    VirtualMachineDiagnostic.objects.create_for_vm(vm, level, source=source,
607
                                                   source_date=etime,
608
                                                   message=message,
609
                                                   details=details)
610

    
611

    
612
def create_instance(vm, nics, flavor, image):
613
    """`image` is a dictionary which should contain the keys:
614
            'backend_id', 'format' and 'metadata'
615

616
        metadata value should be a dictionary.
617
    """
618

    
619
    # Handle arguments to CreateInstance() as a dictionary,
620
    # initialize it based on a deployment-specific value.
621
    # This enables the administrator to override deployment-specific
622
    # arguments, such as the disk template to use, name of os provider
623
    # and hypervisor-specific parameters at will (see Synnefo #785, #835).
624
    #
625
    kw = vm.backend.get_create_params()
626
    kw['mode'] = 'create'
627
    kw['name'] = vm.backend_vm_id
628
    # Defined in settings.GANETI_CREATEINSTANCE_KWARGS
629

    
630
    kw['disk_template'] = flavor.disk_template
631
    kw['disks'] = [{"size": flavor.disk * 1024}]
632
    provider = flavor.disk_provider
633
    if provider:
634
        kw['disks'][0]['provider'] = provider
635
        kw['disks'][0]['origin'] = flavor.disk_origin
636
        extra_disk_params = settings.GANETI_DISK_PROVIDER_KWARGS.get(provider)
637
        if extra_disk_params is not None:
638
            kw["disks"][0].update(extra_disk_params)
639

    
640
    kw['nics'] = [{"name": nic.backend_uuid,
641
                   "network": nic.network.backend_id,
642
                   "ip": nic.ipv4_address}
643
                  for nic in nics]
644

    
645
    backend = vm.backend
646
    depend_jobs = []
647
    for nic in nics:
648
        bnet, job_ids = ensure_network_is_active(backend, nic.network_id)
649
        depend_jobs.extend(job_ids)
650

    
651
    kw["depends"] = create_job_dependencies(depend_jobs)
652

    
653
    # Defined in settings.GANETI_CREATEINSTANCE_KWARGS
654
    # kw['os'] = settings.GANETI_OS_PROVIDER
655
    kw['ip_check'] = False
656
    kw['name_check'] = False
657

    
658
    # Do not specific a node explicitly, have
659
    # Ganeti use an iallocator instead
660
    #kw['pnode'] = rapi.GetNodes()[0]
661

    
662
    kw['dry_run'] = settings.TEST
663

    
664
    kw['beparams'] = {
665
        'auto_balance': True,
666
        'vcpus': flavor.cpu,
667
        'memory': flavor.ram}
668

    
669
    kw['osparams'] = {
670
        'config_url': vm.config_url,
671
        # Store image id and format to Ganeti
672
        'img_id': image['backend_id'],
673
        'img_format': image['format']}
674

    
675
    # Use opportunistic locking
676
    kw['opportunistic_locking'] = settings.GANETI_USE_OPPORTUNISTIC_LOCKING
677

    
678
    # Defined in settings.GANETI_CREATEINSTANCE_KWARGS
679
    # kw['hvparams'] = dict(serial_console=False)
680

    
681
    log.debug("Creating instance %s", utils.hide_pass(kw))
682
    with pooled_rapi_client(vm) as client:
683
        return client.CreateInstance(**kw)
684

    
685

    
686
def delete_instance(vm):
687
    with pooled_rapi_client(vm) as client:
688
        return client.DeleteInstance(vm.backend_vm_id, dry_run=settings.TEST)
689

    
690

    
691
def reboot_instance(vm, reboot_type):
692
    assert reboot_type in ('soft', 'hard')
693
    # Note that reboot type of Ganeti job must be always hard. The 'soft' and
694
    # 'hard' type of OS API is different from the one in Ganeti, and maps to
695
    # 'shutdown_timeout'.
696
    kwargs = {"instance": vm.backend_vm_id,
697
              "reboot_type": "hard"}
698
    # 'shutdown_timeout' parameter is only support from snf-ganeti>=2.8.2 and
699
    # Ganeti > 2.10. In other versions this parameter will be ignored and
700
    # we will fallback to default timeout of Ganeti (120s).
701
    if reboot_type == "hard":
702
        kwargs["shutdown_timeout"] = 0
703
    if settings.TEST:
704
        kwargs["dry_run"] = True
705
    with pooled_rapi_client(vm) as client:
706
        return client.RebootInstance(**kwargs)
707

    
708

    
709
def startup_instance(vm):
710
    with pooled_rapi_client(vm) as client:
711
        return client.StartupInstance(vm.backend_vm_id, dry_run=settings.TEST)
712

    
713

    
714
def shutdown_instance(vm):
715
    with pooled_rapi_client(vm) as client:
716
        return client.ShutdownInstance(vm.backend_vm_id, dry_run=settings.TEST)
717

    
718

    
719
def resize_instance(vm, vcpus, memory):
720
    beparams = {"vcpus": int(vcpus),
721
                "minmem": int(memory),
722
                "maxmem": int(memory)}
723
    with pooled_rapi_client(vm) as client:
724
        return client.ModifyInstance(vm.backend_vm_id, beparams=beparams)
725

    
726

    
727
def get_instance_console(vm):
728
    # RAPI GetInstanceConsole() returns endpoints to the vnc_bind_address,
729
    # which is a cluster-wide setting, either 0.0.0.0 or 127.0.0.1, and pretty
730
    # useless (see #783).
731
    #
732
    # Until this is fixed on the Ganeti side, construct a console info reply
733
    # directly.
734
    #
735
    # WARNING: This assumes that VNC runs on port network_port on
736
    #          the instance's primary node, and is probably
737
    #          hypervisor-specific.
738
    #
739
    log.debug("Getting console for vm %s", vm)
740

    
741
    console = {}
742
    console['kind'] = 'vnc'
743

    
744
    with pooled_rapi_client(vm) as client:
745
        i = client.GetInstance(vm.backend_vm_id)
746

    
747
    if vm.backend.hypervisor == "kvm" and i['hvparams']['serial_console']:
748
        raise Exception("hv parameter serial_console cannot be true")
749
    console['host'] = i['pnode']
750
    console['port'] = i['network_port']
751

    
752
    return console
753

    
754

    
755
def get_instance_info(vm):
756
    with pooled_rapi_client(vm) as client:
757
        return client.GetInstance(vm.backend_vm_id)
758

    
759

    
760
def vm_exists_in_backend(vm):
761
    try:
762
        get_instance_info(vm)
763
        return True
764
    except rapi.GanetiApiError as e:
765
        if e.code == 404:
766
            return False
767
        raise e
768

    
769

    
770
def get_network_info(backend_network):
771
    with pooled_rapi_client(backend_network) as client:
772
        return client.GetNetwork(backend_network.network.backend_id)
773

    
774

    
775
def network_exists_in_backend(backend_network):
776
    try:
777
        get_network_info(backend_network)
778
        return True
779
    except rapi.GanetiApiError as e:
780
        if e.code == 404:
781
            return False
782

    
783

    
784
def job_is_still_running(vm, job_id=None):
785
    with pooled_rapi_client(vm) as c:
786
        try:
787
            if job_id is None:
788
                job_id = vm.backendjobid
789
            job_info = c.GetJobStatus(job_id)
790
            return not (job_info["status"] in rapi.JOB_STATUS_FINALIZED)
791
        except rapi.GanetiApiError:
792
            return False
793

    
794

    
795
def nic_is_stale(vm, nic, timeout=60):
796
    """Check if a NIC is stale or exists in the Ganeti backend."""
797
    # First check the state of the NIC and if there is a pending CONNECT
798
    if nic.state == "BUILD" and vm.task == "CONNECT":
799
        if datetime.now() < nic.created + timedelta(seconds=timeout):
800
            # Do not check for too recent NICs to avoid the time overhead
801
            return False
802
        if job_is_still_running(vm, job_id=vm.task_job_id):
803
            return False
804
        else:
805
            # If job has finished, check that the NIC exists, because the
806
            # message may have been lost or stuck in the queue.
807
            vm_info = get_instance_info(vm)
808
            if nic.backend_uuid in vm_info["nic.names"]:
809
                return False
810
    return True
811

    
812

    
813
def ensure_network_is_active(backend, network_id):
814
    """Ensure that a network is active in the specified backend
815

816
    Check that a network exists and is active in the specified backend. If not
817
    (re-)create the network. Return the corresponding BackendNetwork object
818
    and the IDs of the Ganeti job to create the network.
819

820
    """
821
    job_ids = []
822
    try:
823
        bnet = BackendNetwork.objects.select_related("network")\
824
                                     .get(backend=backend, network=network_id)
825
        if bnet.operstate != "ACTIVE":
826
            job_ids = create_network(bnet.network, backend, connect=True)
827
    except BackendNetwork.DoesNotExist:
828
        network = Network.objects.select_for_update().get(id=network_id)
829
        bnet = BackendNetwork.objects.create(backend=backend, network=network)
830
        job_ids = create_network(network, backend, connect=True)
831

    
832
    return bnet, job_ids
833

    
834

    
835
def create_network(network, backend, connect=True):
836
    """Create a network in a Ganeti backend"""
837
    log.debug("Creating network %s in backend %s", network, backend)
838

    
839
    job_id = _create_network(network, backend)
840

    
841
    if connect:
842
        job_ids = connect_network(network, backend, depends=[job_id])
843
        return job_ids
844
    else:
845
        return [job_id]
846

    
847

    
848
def _create_network(network, backend):
849
    """Create a network."""
850

    
851
    tags = network.backend_tag
852
    subnet = None
853
    subnet6 = None
854
    gateway = None
855
    gateway6 = None
856
    for _subnet in network.subnets.all():
857
        if _subnet.dhcp and not "nfdhcpd" in tags:
858
            tags.append("nfdhcpd")
859
        if _subnet.ipversion == 4:
860
            subnet = _subnet.cidr
861
            gateway = _subnet.gateway
862
        elif _subnet.ipversion == 6:
863
            subnet6 = _subnet.cidr
864
            gateway6 = _subnet.gateway
865

    
866
    conflicts_check = False
867
    if network.public:
868
        tags.append('public')
869
        if subnet is not None:
870
            conflicts_check = True
871
    else:
872
        tags.append('private')
873

    
874
    # Use a dummy network subnet for IPv6 only networks. Currently Ganeti does
875
    # not support IPv6 only networks. To bypass this limitation, we create the
876
    # network with a dummy network subnet, and make Cyclades connect instances
877
    # to such networks, with address=None.
878
    if subnet is None:
879
        subnet = "10.0.0.0/29"
880

    
881
    try:
882
        bn = BackendNetwork.objects.get(network=network, backend=backend)
883
        mac_prefix = bn.mac_prefix
884
    except BackendNetwork.DoesNotExist:
885
        raise Exception("BackendNetwork for network '%s' in backend '%s'"
886
                        " does not exist" % (network.id, backend.id))
887

    
888
    with pooled_rapi_client(backend) as client:
889
        return client.CreateNetwork(network_name=network.backend_id,
890
                                    network=subnet,
891
                                    network6=subnet6,
892
                                    gateway=gateway,
893
                                    gateway6=gateway6,
894
                                    mac_prefix=mac_prefix,
895
                                    conflicts_check=conflicts_check,
896
                                    tags=tags)
897

    
898

    
899
def connect_network(network, backend, depends=[], group=None):
900
    """Connect a network to nodegroups."""
901
    log.debug("Connecting network %s to backend %s", network, backend)
902

    
903
    conflicts_check = False
904
    if network.public and (network.subnet4 is not None):
905
        conflicts_check = True
906

    
907
    depends = create_job_dependencies(depends)
908
    with pooled_rapi_client(backend) as client:
909
        groups = [group] if group is not None else client.GetGroups()
910
        job_ids = []
911
        for group in groups:
912
            job_id = client.ConnectNetwork(network.backend_id, group,
913
                                           network.mode, network.link,
914
                                           conflicts_check,
915
                                           depends=depends)
916
            job_ids.append(job_id)
917
    return job_ids
918

    
919

    
920
def delete_network(network, backend, disconnect=True):
921
    log.debug("Deleting network %s from backend %s", network, backend)
922

    
923
    depends = []
924
    if disconnect:
925
        depends = disconnect_network(network, backend)
926
    _delete_network(network, backend, depends=depends)
927

    
928

    
929
def _delete_network(network, backend, depends=[]):
930
    depends = create_job_dependencies(depends)
931
    with pooled_rapi_client(backend) as client:
932
        return client.DeleteNetwork(network.backend_id, depends)
933

    
934

    
935
def disconnect_network(network, backend, group=None):
936
    log.debug("Disconnecting network %s to backend %s", network, backend)
937

    
938
    with pooled_rapi_client(backend) as client:
939
        groups = [group] if group is not None else client.GetGroups()
940
        job_ids = []
941
        for group in groups:
942
            job_id = client.DisconnectNetwork(network.backend_id, group)
943
            job_ids.append(job_id)
944
    return job_ids
945

    
946

    
947
def connect_to_network(vm, nic):
948
    network = nic.network
949
    backend = vm.backend
950
    bnet, depend_jobs = ensure_network_is_active(backend, network.id)
951

    
952
    depends = create_job_dependencies(depend_jobs)
953

    
954
    nic = {'name': nic.backend_uuid,
955
           'network': network.backend_id,
956
           'ip': nic.ipv4_address}
957

    
958
    log.debug("Adding NIC %s to VM %s", nic, vm)
959

    
960
    kwargs = {
961
        "instance": vm.backend_vm_id,
962
        "nics": [("add", "-1", nic)],
963
        "depends": depends,
964
    }
965
    if vm.backend.use_hotplug():
966
        kwargs["hotplug_if_possible"] = True
967
    if settings.TEST:
968
        kwargs["dry_run"] = True
969

    
970
    with pooled_rapi_client(vm) as client:
971
        return client.ModifyInstance(**kwargs)
972

    
973

    
974
def disconnect_from_network(vm, nic):
975
    log.debug("Removing NIC %s of VM %s", nic, vm)
976

    
977
    kwargs = {
978
        "instance": vm.backend_vm_id,
979
        "nics": [("remove", nic.backend_uuid, {})],
980
    }
981
    if vm.backend.use_hotplug():
982
        kwargs["hotplug_if_possible"] = True
983
    if settings.TEST:
984
        kwargs["dry_run"] = True
985

    
986
    with pooled_rapi_client(vm) as client:
987
        jobID = client.ModifyInstance(**kwargs)
988
        firewall_profile = nic.firewall_profile
989
        if firewall_profile and firewall_profile != "DISABLED":
990
            tag = _firewall_tags[firewall_profile] % nic.backend_uuid
991
            client.DeleteInstanceTags(vm.backend_vm_id, [tag],
992
                                      dry_run=settings.TEST)
993

    
994
        return jobID
995

    
996

    
997
def set_firewall_profile(vm, profile, nic):
998
    uuid = nic.backend_uuid
999
    try:
1000
        tag = _firewall_tags[profile] % uuid
1001
    except KeyError:
1002
        raise ValueError("Unsopported Firewall Profile: %s" % profile)
1003

    
1004
    log.debug("Setting tag of VM %s, NIC %s, to %s", vm, nic, profile)
1005

    
1006
    with pooled_rapi_client(vm) as client:
1007
        # Delete previous firewall tags
1008
        old_tags = client.GetInstanceTags(vm.backend_vm_id)
1009
        delete_tags = [(t % uuid) for t in _firewall_tags.values()
1010
                       if (t % uuid) in old_tags]
1011
        if delete_tags:
1012
            client.DeleteInstanceTags(vm.backend_vm_id, delete_tags,
1013
                                      dry_run=settings.TEST)
1014

    
1015
        if profile != "DISABLED":
1016
            client.AddInstanceTags(vm.backend_vm_id, [tag],
1017
                                   dry_run=settings.TEST)
1018

    
1019
        # XXX NOP ModifyInstance call to force process_net_status to run
1020
        # on the dispatcher
1021
        os_name = settings.GANETI_CREATEINSTANCE_KWARGS['os']
1022
        client.ModifyInstance(vm.backend_vm_id,
1023
                              os_name=os_name)
1024
    return None
1025

    
1026

    
1027
def get_instances(backend, bulk=True):
1028
    with pooled_rapi_client(backend) as c:
1029
        return c.GetInstances(bulk=bulk)
1030

    
1031

    
1032
def get_nodes(backend, bulk=True):
1033
    with pooled_rapi_client(backend) as c:
1034
        return c.GetNodes(bulk=bulk)
1035

    
1036

    
1037
def get_jobs(backend, bulk=True):
1038
    with pooled_rapi_client(backend) as c:
1039
        return c.GetJobs(bulk=bulk)
1040

    
1041

    
1042
def get_physical_resources(backend):
1043
    """ Get the physical resources of a backend.
1044

1045
    Get the resources of a backend as reported by the backend (not the db).
1046

1047
    """
1048
    nodes = get_nodes(backend, bulk=True)
1049
    attr = ['mfree', 'mtotal', 'dfree', 'dtotal', 'pinst_cnt', 'ctotal']
1050
    res = {}
1051
    for a in attr:
1052
        res[a] = 0
1053
    for n in nodes:
1054
        # Filter out drained, offline and not vm_capable nodes since they will
1055
        # not take part in the vm allocation process
1056
        can_host_vms = n['vm_capable'] and not (n['drained'] or n['offline'])
1057
        if can_host_vms and n['cnodes']:
1058
            for a in attr:
1059
                res[a] += int(n[a] or 0)
1060
    return res
1061

    
1062

    
1063
def update_backend_resources(backend, resources=None):
1064
    """ Update the state of the backend resources in db.
1065

1066
    """
1067

    
1068
    if not resources:
1069
        resources = get_physical_resources(backend)
1070

    
1071
    backend.mfree = resources['mfree']
1072
    backend.mtotal = resources['mtotal']
1073
    backend.dfree = resources['dfree']
1074
    backend.dtotal = resources['dtotal']
1075
    backend.pinst_cnt = resources['pinst_cnt']
1076
    backend.ctotal = resources['ctotal']
1077
    backend.updated = datetime.now()
1078
    backend.save()
1079

    
1080

    
1081
def get_memory_from_instances(backend):
1082
    """ Get the memory that is used from instances.
1083

1084
    Get the used memory of a backend. Note: This is different for
1085
    the real memory used, due to kvm's memory de-duplication.
1086

1087
    """
1088
    with pooled_rapi_client(backend) as client:
1089
        instances = client.GetInstances(bulk=True)
1090
    mem = 0
1091
    for i in instances:
1092
        mem += i['oper_ram']
1093
    return mem
1094

    
1095

    
1096
def get_available_disk_templates(backend):
1097
    """Get the list of available disk templates of a Ganeti backend.
1098

1099
    The list contains the disk templates that are enabled in the Ganeti backend
1100
    and also included in ipolicy-disk-templates.
1101

1102
    """
1103
    with pooled_rapi_client(backend) as c:
1104
        info = c.GetInfo()
1105
    ipolicy_disk_templates = info["ipolicy"]["disk-templates"]
1106
    try:
1107
        enabled_disk_templates = info["enabled_disk_templates"]
1108
        return [dp for dp in enabled_disk_templates
1109
                if dp in ipolicy_disk_templates]
1110
    except KeyError:
1111
        # Ganeti < 2.8 does not have 'enabled_disk_templates'
1112
        return ipolicy_disk_templates
1113

    
1114

    
1115
def update_backend_disk_templates(backend):
1116
    disk_templates = get_available_disk_templates(backend)
1117
    backend.disk_templates = disk_templates
1118
    backend.save()
1119

    
1120

    
1121
##
1122
## Synchronized operations for reconciliation
1123
##
1124

    
1125

    
1126
def create_network_synced(network, backend):
1127
    result = _create_network_synced(network, backend)
1128
    if result[0] != rapi.JOB_STATUS_SUCCESS:
1129
        return result
1130
    result = connect_network_synced(network, backend)
1131
    return result
1132

    
1133

    
1134
def _create_network_synced(network, backend):
1135
    with pooled_rapi_client(backend) as client:
1136
        job = _create_network(network, backend)
1137
        result = wait_for_job(client, job)
1138
    return result
1139

    
1140

    
1141
def connect_network_synced(network, backend):
1142
    with pooled_rapi_client(backend) as client:
1143
        for group in client.GetGroups():
1144
            job = client.ConnectNetwork(network.backend_id, group,
1145
                                        network.mode, network.link)
1146
            result = wait_for_job(client, job)
1147
            if result[0] != rapi.JOB_STATUS_SUCCESS:
1148
                return result
1149

    
1150
    return result
1151

    
1152

    
1153
def wait_for_job(client, jobid):
1154
    result = client.WaitForJobChange(jobid, ['status', 'opresult'], None, None)
1155
    status = result['job_info'][0]
1156
    while status not in rapi.JOB_STATUS_FINALIZED:
1157
        result = client.WaitForJobChange(jobid, ['status', 'opresult'],
1158
                                         [result], None)
1159
        status = result['job_info'][0]
1160

    
1161
    if status == rapi.JOB_STATUS_SUCCESS:
1162
        return (status, None)
1163
    else:
1164
        error = result['job_info'][1]
1165
        return (status, error)
1166

    
1167

    
1168
def create_job_dependencies(job_ids=[], job_states=None):
1169
    """Transform a list of job IDs to Ganeti 'depends' attribute."""
1170
    if job_states is None:
1171
        job_states = list(rapi.JOB_STATUS_FINALIZED)
1172
    assert(type(job_states) == list)
1173
    return [[job_id, job_states] for job_id in job_ids]