Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (43.7 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_resource_serial(vm)
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_resource_serial(vm)
97
    elif job_status == rapi.JOB_STATUS_SUCCESS:
98
        commission_info = quotas.get_commission_info(resource=vm,
99
                                                     action=action,
100
                                                     action_fields=job_fields)
101
        if commission_info is not None:
102
            # Commission for this change has not been issued, or the issued
103
            # commission was unaware of the current change. Reject all previous
104
            # commissions and create a new one in forced mode!
105
            log.debug("Expected job was %s. Processing job %s. "
106
                      "Attached serial %s",
107
                      vm.task_job_id, job_id, vm.serial)
108
            reason = ("client: dispatcher, resource: %s, ganeti_job: %s"
109
                      % (vm, job_id))
110
            serial = quotas.handle_resource_commission(
111
                vm, action,
112
                action_fields=job_fields,
113
                commission_name=reason,
114
                force=True,
115
                auto_accept=True)
116
            log.debug("Issued new commission: %s", serial)
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, shutdown_timeout=None):
687
    with pooled_rapi_client(vm) as client:
688
        return client.DeleteInstance(vm.backend_vm_id,
689
                                     shutdown_timeout=shutdown_timeout,
690
                                     dry_run=settings.TEST)
691

    
692

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

    
712

    
713
def startup_instance(vm):
714
    with pooled_rapi_client(vm) as client:
715
        return client.StartupInstance(vm.backend_vm_id, dry_run=settings.TEST)
716

    
717

    
718
def shutdown_instance(vm, shutdown_timeout=None):
719
    with pooled_rapi_client(vm) as client:
720
        return client.ShutdownInstance(vm.backend_vm_id,
721
                                       timeout=shutdown_timeout,
722
                                       dry_run=settings.TEST)
723

    
724

    
725
def resize_instance(vm, vcpus, memory):
726
    beparams = {"vcpus": int(vcpus),
727
                "minmem": int(memory),
728
                "maxmem": int(memory)}
729
    with pooled_rapi_client(vm) as client:
730
        return client.ModifyInstance(vm.backend_vm_id, beparams=beparams)
731

    
732

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

    
747
    console = {}
748
    console['kind'] = 'vnc'
749

    
750
    with pooled_rapi_client(vm) as client:
751
        i = client.GetInstance(vm.backend_vm_id)
752

    
753
    if vm.backend.hypervisor == "kvm" and i['hvparams']['serial_console']:
754
        raise Exception("hv parameter serial_console cannot be true")
755
    console['host'] = i['pnode']
756
    console['port'] = i['network_port']
757

    
758
    return console
759

    
760

    
761
def get_instance_info(vm):
762
    with pooled_rapi_client(vm) as client:
763
        return client.GetInstance(vm.backend_vm_id)
764

    
765

    
766
def vm_exists_in_backend(vm):
767
    try:
768
        get_instance_info(vm)
769
        return True
770
    except rapi.GanetiApiError as e:
771
        if e.code == 404:
772
            return False
773
        raise e
774

    
775

    
776
def get_network_info(backend_network):
777
    with pooled_rapi_client(backend_network) as client:
778
        return client.GetNetwork(backend_network.network.backend_id)
779

    
780

    
781
def network_exists_in_backend(backend_network):
782
    try:
783
        get_network_info(backend_network)
784
        return True
785
    except rapi.GanetiApiError as e:
786
        if e.code == 404:
787
            return False
788

    
789

    
790
def job_is_still_running(vm, job_id=None):
791
    with pooled_rapi_client(vm) as c:
792
        try:
793
            if job_id is None:
794
                job_id = vm.backendjobid
795
            job_info = c.GetJobStatus(job_id)
796
            return not (job_info["status"] in rapi.JOB_STATUS_FINALIZED)
797
        except rapi.GanetiApiError:
798
            return False
799

    
800

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

    
818

    
819
def ensure_network_is_active(backend, network_id):
820
    """Ensure that a network is active in the specified backend
821

822
    Check that a network exists and is active in the specified backend. If not
823
    (re-)create the network. Return the corresponding BackendNetwork object
824
    and the IDs of the Ganeti job to create the network.
825

826
    """
827
    job_ids = []
828
    try:
829
        bnet = BackendNetwork.objects.select_related("network")\
830
                                     .get(backend=backend, network=network_id)
831
        if bnet.operstate != "ACTIVE":
832
            job_ids = create_network(bnet.network, backend, connect=True)
833
    except BackendNetwork.DoesNotExist:
834
        network = Network.objects.select_for_update().get(id=network_id)
835
        bnet = BackendNetwork.objects.create(backend=backend, network=network)
836
        job_ids = create_network(network, backend, connect=True)
837

    
838
    return bnet, job_ids
839

    
840

    
841
def create_network(network, backend, connect=True):
842
    """Create a network in a Ganeti backend"""
843
    log.debug("Creating network %s in backend %s", network, backend)
844

    
845
    job_id = _create_network(network, backend)
846

    
847
    if connect:
848
        job_ids = connect_network(network, backend, depends=[job_id])
849
        return job_ids
850
    else:
851
        return [job_id]
852

    
853

    
854
def _create_network(network, backend):
855
    """Create a network."""
856

    
857
    tags = network.backend_tag
858
    subnet = None
859
    subnet6 = None
860
    gateway = None
861
    gateway6 = None
862
    for _subnet in network.subnets.all():
863
        if _subnet.dhcp and not "nfdhcpd" in tags:
864
            tags.append("nfdhcpd")
865
        if _subnet.ipversion == 4:
866
            subnet = _subnet.cidr
867
            gateway = _subnet.gateway
868
        elif _subnet.ipversion == 6:
869
            subnet6 = _subnet.cidr
870
            gateway6 = _subnet.gateway
871

    
872
    conflicts_check = False
873
    if network.public:
874
        tags.append('public')
875
        if subnet is not None:
876
            conflicts_check = True
877
    else:
878
        tags.append('private')
879

    
880
    # Use a dummy network subnet for IPv6 only networks. Currently Ganeti does
881
    # not support IPv6 only networks. To bypass this limitation, we create the
882
    # network with a dummy network subnet, and make Cyclades connect instances
883
    # to such networks, with address=None.
884
    if subnet is None:
885
        subnet = "10.0.0.0/29"
886

    
887
    try:
888
        bn = BackendNetwork.objects.get(network=network, backend=backend)
889
        mac_prefix = bn.mac_prefix
890
    except BackendNetwork.DoesNotExist:
891
        raise Exception("BackendNetwork for network '%s' in backend '%s'"
892
                        " does not exist" % (network.id, backend.id))
893

    
894
    with pooled_rapi_client(backend) as client:
895
        return client.CreateNetwork(network_name=network.backend_id,
896
                                    network=subnet,
897
                                    network6=subnet6,
898
                                    gateway=gateway,
899
                                    gateway6=gateway6,
900
                                    mac_prefix=mac_prefix,
901
                                    conflicts_check=conflicts_check,
902
                                    tags=tags)
903

    
904

    
905
def connect_network(network, backend, depends=[], group=None):
906
    """Connect a network to nodegroups."""
907
    log.debug("Connecting network %s to backend %s", network, backend)
908

    
909
    conflicts_check = False
910
    if network.public and (network.subnet4 is not None):
911
        conflicts_check = True
912

    
913
    depends = create_job_dependencies(depends)
914
    with pooled_rapi_client(backend) as client:
915
        groups = [group] if group is not None else client.GetGroups()
916
        job_ids = []
917
        for group in groups:
918
            job_id = client.ConnectNetwork(network.backend_id, group,
919
                                           network.mode, network.link,
920
                                           conflicts_check,
921
                                           depends=depends)
922
            job_ids.append(job_id)
923
    return job_ids
924

    
925

    
926
def delete_network(network, backend, disconnect=True):
927
    log.debug("Deleting network %s from backend %s", network, backend)
928

    
929
    depends = []
930
    if disconnect:
931
        depends = disconnect_network(network, backend)
932
    _delete_network(network, backend, depends=depends)
933

    
934

    
935
def _delete_network(network, backend, depends=[]):
936
    depends = create_job_dependencies(depends)
937
    with pooled_rapi_client(backend) as client:
938
        return client.DeleteNetwork(network.backend_id, depends)
939

    
940

    
941
def disconnect_network(network, backend, group=None):
942
    log.debug("Disconnecting network %s to backend %s", network, backend)
943

    
944
    with pooled_rapi_client(backend) as client:
945
        groups = [group] if group is not None else client.GetGroups()
946
        job_ids = []
947
        for group in groups:
948
            job_id = client.DisconnectNetwork(network.backend_id, group)
949
            job_ids.append(job_id)
950
    return job_ids
951

    
952

    
953
def connect_to_network(vm, nic):
954
    network = nic.network
955
    backend = vm.backend
956
    bnet, depend_jobs = ensure_network_is_active(backend, network.id)
957

    
958
    depends = create_job_dependencies(depend_jobs)
959

    
960
    nic = {'name': nic.backend_uuid,
961
           'network': network.backend_id,
962
           'ip': nic.ipv4_address}
963

    
964
    log.debug("Adding NIC %s to VM %s", nic, vm)
965

    
966
    kwargs = {
967
        "instance": vm.backend_vm_id,
968
        "nics": [("add", "-1", nic)],
969
        "depends": depends,
970
    }
971
    if vm.backend.use_hotplug():
972
        kwargs["hotplug_if_possible"] = True
973
    if settings.TEST:
974
        kwargs["dry_run"] = True
975

    
976
    with pooled_rapi_client(vm) as client:
977
        return client.ModifyInstance(**kwargs)
978

    
979

    
980
def disconnect_from_network(vm, nic):
981
    log.debug("Removing NIC %s of VM %s", nic, vm)
982

    
983
    kwargs = {
984
        "instance": vm.backend_vm_id,
985
        "nics": [("remove", nic.backend_uuid, {})],
986
    }
987
    if vm.backend.use_hotplug():
988
        kwargs["hotplug_if_possible"] = True
989
    if settings.TEST:
990
        kwargs["dry_run"] = True
991

    
992
    with pooled_rapi_client(vm) as client:
993
        jobID = client.ModifyInstance(**kwargs)
994
        firewall_profile = nic.firewall_profile
995
        if firewall_profile and firewall_profile != "DISABLED":
996
            tag = _firewall_tags[firewall_profile] % nic.backend_uuid
997
            client.DeleteInstanceTags(vm.backend_vm_id, [tag],
998
                                      dry_run=settings.TEST)
999

    
1000
        return jobID
1001

    
1002

    
1003
def set_firewall_profile(vm, profile, nic):
1004
    uuid = nic.backend_uuid
1005
    try:
1006
        tag = _firewall_tags[profile] % uuid
1007
    except KeyError:
1008
        raise ValueError("Unsopported Firewall Profile: %s" % profile)
1009

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

    
1012
    with pooled_rapi_client(vm) as client:
1013
        # Delete previous firewall tags
1014
        old_tags = client.GetInstanceTags(vm.backend_vm_id)
1015
        delete_tags = [(t % uuid) for t in _firewall_tags.values()
1016
                       if (t % uuid) in old_tags]
1017
        if delete_tags:
1018
            client.DeleteInstanceTags(vm.backend_vm_id, delete_tags,
1019
                                      dry_run=settings.TEST)
1020

    
1021
        if profile != "DISABLED":
1022
            client.AddInstanceTags(vm.backend_vm_id, [tag],
1023
                                   dry_run=settings.TEST)
1024

    
1025
        # XXX NOP ModifyInstance call to force process_net_status to run
1026
        # on the dispatcher
1027
        os_name = settings.GANETI_CREATEINSTANCE_KWARGS['os']
1028
        client.ModifyInstance(vm.backend_vm_id,
1029
                              os_name=os_name)
1030
    return None
1031

    
1032

    
1033
def attach_volume(vm, volume, depends=[]):
1034
    log.debug("Attaching volume %s to vm %s", vm, volume)
1035

    
1036
    disk = {"size": volume.size,
1037
            "name": volume.backend_volume_uuid,
1038
            "volume_name": volume.backend_volume_uuid}
1039
    if volume.source_volume_id is not None:
1040
        disk["origin"] = volume.source_volume.backend_volume_uuid
1041
    elif volume.source_snapshot is not None:
1042
        disk["origin"] = volume.source_snapshot["checksum"]
1043
    elif volume.source_image is not None:
1044
        disk["origin"] = volume.source_image["checksum"]
1045

    
1046
    kwargs = {
1047
        "instance": vm.backend_vm_id,
1048
        "disks": [("add", "-1", disk)],
1049
        "depends": depends,
1050
    }
1051
    if vm.backend.use_hotplug():
1052
        kwargs["hotplug"] = True
1053
    if settings.TEST:
1054
        kwargs["dry_run"] = True
1055

    
1056
    with pooled_rapi_client(vm) as client:
1057
        return client.ModifyInstance(**kwargs)
1058

    
1059

    
1060
def detach_volume(vm, volume):
1061
    log.debug("Removing volume %s from vm %s", volume, vm)
1062
    kwargs = {
1063
        "instance": vm.backend_vm_id,
1064
        "disks": [("remove", volume.backend_volume_uuid, {})],
1065
    }
1066
    if vm.backend.use_hotplug():
1067
        kwargs["hotplug"] = True
1068
    if settings.TEST:
1069
        kwargs["dry_run"] = True
1070

    
1071
    with pooled_rapi_client(vm) as client:
1072
        return client.ModifyInstance(**kwargs)
1073

    
1074

    
1075
def get_instances(backend, bulk=True):
1076
    with pooled_rapi_client(backend) as c:
1077
        return c.GetInstances(bulk=bulk)
1078

    
1079

    
1080
def get_nodes(backend, bulk=True):
1081
    with pooled_rapi_client(backend) as c:
1082
        return c.GetNodes(bulk=bulk)
1083

    
1084

    
1085
def get_jobs(backend, bulk=True):
1086
    with pooled_rapi_client(backend) as c:
1087
        return c.GetJobs(bulk=bulk)
1088

    
1089

    
1090
def get_physical_resources(backend):
1091
    """ Get the physical resources of a backend.
1092

1093
    Get the resources of a backend as reported by the backend (not the db).
1094

1095
    """
1096
    nodes = get_nodes(backend, bulk=True)
1097
    attr = ['mfree', 'mtotal', 'dfree', 'dtotal', 'pinst_cnt', 'ctotal']
1098
    res = {}
1099
    for a in attr:
1100
        res[a] = 0
1101
    for n in nodes:
1102
        # Filter out drained, offline and not vm_capable nodes since they will
1103
        # not take part in the vm allocation process
1104
        can_host_vms = n['vm_capable'] and not (n['drained'] or n['offline'])
1105
        if can_host_vms and n['cnodes']:
1106
            for a in attr:
1107
                res[a] += int(n[a] or 0)
1108
    return res
1109

    
1110

    
1111
def update_backend_resources(backend, resources=None):
1112
    """ Update the state of the backend resources in db.
1113

1114
    """
1115

    
1116
    if not resources:
1117
        resources = get_physical_resources(backend)
1118

    
1119
    backend.mfree = resources['mfree']
1120
    backend.mtotal = resources['mtotal']
1121
    backend.dfree = resources['dfree']
1122
    backend.dtotal = resources['dtotal']
1123
    backend.pinst_cnt = resources['pinst_cnt']
1124
    backend.ctotal = resources['ctotal']
1125
    backend.updated = datetime.now()
1126
    backend.save()
1127

    
1128

    
1129
def get_memory_from_instances(backend):
1130
    """ Get the memory that is used from instances.
1131

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

1135
    """
1136
    with pooled_rapi_client(backend) as client:
1137
        instances = client.GetInstances(bulk=True)
1138
    mem = 0
1139
    for i in instances:
1140
        mem += i['oper_ram']
1141
    return mem
1142

    
1143

    
1144
def get_available_disk_templates(backend):
1145
    """Get the list of available disk templates of a Ganeti backend.
1146

1147
    The list contains the disk templates that are enabled in the Ganeti backend
1148
    and also included in ipolicy-disk-templates.
1149

1150
    """
1151
    with pooled_rapi_client(backend) as c:
1152
        info = c.GetInfo()
1153
    ipolicy_disk_templates = info["ipolicy"]["disk-templates"]
1154
    try:
1155
        enabled_disk_templates = info["enabled_disk_templates"]
1156
        return [dp for dp in enabled_disk_templates
1157
                if dp in ipolicy_disk_templates]
1158
    except KeyError:
1159
        # Ganeti < 2.8 does not have 'enabled_disk_templates'
1160
        return ipolicy_disk_templates
1161

    
1162

    
1163
def update_backend_disk_templates(backend):
1164
    disk_templates = get_available_disk_templates(backend)
1165
    backend.disk_templates = disk_templates
1166
    backend.save()
1167

    
1168

    
1169
##
1170
## Synchronized operations for reconciliation
1171
##
1172

    
1173

    
1174
def create_network_synced(network, backend):
1175
    result = _create_network_synced(network, backend)
1176
    if result[0] != rapi.JOB_STATUS_SUCCESS:
1177
        return result
1178
    result = connect_network_synced(network, backend)
1179
    return result
1180

    
1181

    
1182
def _create_network_synced(network, backend):
1183
    with pooled_rapi_client(backend) as client:
1184
        job = _create_network(network, backend)
1185
        result = wait_for_job(client, job)
1186
    return result
1187

    
1188

    
1189
def connect_network_synced(network, backend):
1190
    with pooled_rapi_client(backend) as client:
1191
        for group in client.GetGroups():
1192
            job = client.ConnectNetwork(network.backend_id, group,
1193
                                        network.mode, network.link)
1194
            result = wait_for_job(client, job)
1195
            if result[0] != rapi.JOB_STATUS_SUCCESS:
1196
                return result
1197

    
1198
    return result
1199

    
1200

    
1201
def wait_for_job(client, jobid):
1202
    result = client.WaitForJobChange(jobid, ['status', 'opresult'], None, None)
1203
    status = result['job_info'][0]
1204
    while status not in rapi.JOB_STATUS_FINALIZED:
1205
        result = client.WaitForJobChange(jobid, ['status', 'opresult'],
1206
                                         [result], None)
1207
        status = result['job_info'][0]
1208

    
1209
    if status == rapi.JOB_STATUS_SUCCESS:
1210
        return (status, None)
1211
    else:
1212
        error = result['job_info'][1]
1213
        return (status, error)
1214

    
1215

    
1216
def create_job_dependencies(job_ids=[], job_states=None):
1217
    """Transform a list of job IDs to Ganeti 'depends' attribute."""
1218
    if job_states is None:
1219
        job_states = list(rapi.JOB_STATUS_FINALIZED)
1220
    assert(type(job_states) == list)
1221
    return [[job_id, job_states] for job_id in job_ids]