Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (43.8 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, volumes, 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
    disks = []
632
    for volume in volumes:
633
        disk = {}
634
        disk["size"] = volume.size * 1024
635
        provider = flavor.disk_provider
636
        if provider is not None:
637
            disk["provider"] = provider
638
            disk["origin"] = volume.source_image["checksum"]
639
            extra_disk_params = settings.GANETI_DISK_PROVIDER_KWARGS\
640
                                        .get(provider)
641
            if extra_disk_params is not None:
642
                disk.update(extra_disk_params)
643
        disks.append(disk)
644

    
645
    kw["disks"] = disks
646

    
647
    kw['nics'] = [{"name": nic.backend_uuid,
648
                   "network": nic.network.backend_id,
649
                   "ip": nic.ipv4_address}
650
                  for nic in nics]
651

    
652
    backend = vm.backend
653
    depend_jobs = []
654
    for nic in nics:
655
        bnet, job_ids = ensure_network_is_active(backend, nic.network_id)
656
        depend_jobs.extend(job_ids)
657

    
658
    kw["depends"] = create_job_dependencies(depend_jobs)
659

    
660
    # Defined in settings.GANETI_CREATEINSTANCE_KWARGS
661
    # kw['os'] = settings.GANETI_OS_PROVIDER
662
    kw['ip_check'] = False
663
    kw['name_check'] = False
664

    
665
    # Do not specific a node explicitly, have
666
    # Ganeti use an iallocator instead
667
    #kw['pnode'] = rapi.GetNodes()[0]
668

    
669
    kw['dry_run'] = settings.TEST
670

    
671
    kw['beparams'] = {
672
        'auto_balance': True,
673
        'vcpus': flavor.cpu,
674
        'memory': flavor.ram}
675

    
676
    kw['osparams'] = {
677
        'config_url': vm.config_url,
678
        # Store image id and format to Ganeti
679
        'img_id': image['backend_id'],
680
        'img_format': image['format']}
681

    
682
    # Use opportunistic locking
683
    kw['opportunistic_locking'] = settings.GANETI_USE_OPPORTUNISTIC_LOCKING
684

    
685
    # Defined in settings.GANETI_CREATEINSTANCE_KWARGS
686
    # kw['hvparams'] = dict(serial_console=False)
687

    
688
    log.debug("Creating instance %s", utils.hide_pass(kw))
689
    with pooled_rapi_client(vm) as client:
690
        return client.CreateInstance(**kw)
691

    
692

    
693
def delete_instance(vm):
694
    with pooled_rapi_client(vm) as client:
695
        return client.DeleteInstance(vm.backend_vm_id, dry_run=settings.TEST)
696

    
697

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

    
715

    
716
def startup_instance(vm):
717
    with pooled_rapi_client(vm) as client:
718
        return client.StartupInstance(vm.backend_vm_id, dry_run=settings.TEST)
719

    
720

    
721
def shutdown_instance(vm):
722
    with pooled_rapi_client(vm) as client:
723
        return client.ShutdownInstance(vm.backend_vm_id, dry_run=settings.TEST)
724

    
725

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

    
733

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

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

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

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

    
759
    return console
760

    
761

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

    
766

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

    
776

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

    
781

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

    
790

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

    
801

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

    
819

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

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

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

    
839
    return bnet, job_ids
840

    
841

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

    
846
    job_id = _create_network(network, backend)
847

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

    
854

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

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

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

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

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

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

    
905

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

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

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

    
926

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

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

    
935

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

    
941

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

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

    
953

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

    
959
    depends = create_job_dependencies(depend_jobs)
960

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

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

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

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

    
980

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

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

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

    
1001
        return jobID
1002

    
1003

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

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

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

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

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

    
1033

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

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

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

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

    
1060

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

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

    
1075

    
1076
def snapshot_instance(vm, snapshot_name):
1077
    #volume = instance.volumes.all()[0]
1078
    with pooled_rapi_client(vm) as client:
1079
        return client.SnapshotInstance(instance=vm.backend_vm_id,
1080
                                       snapshot_name=snapshot_name)
1081

    
1082

    
1083
def get_instances(backend, bulk=True):
1084
    with pooled_rapi_client(backend) as c:
1085
        return c.GetInstances(bulk=bulk)
1086

    
1087

    
1088
def get_nodes(backend, bulk=True):
1089
    with pooled_rapi_client(backend) as c:
1090
        return c.GetNodes(bulk=bulk)
1091

    
1092

    
1093
def get_jobs(backend, bulk=True):
1094
    with pooled_rapi_client(backend) as c:
1095
        return c.GetJobs(bulk=bulk)
1096

    
1097

    
1098
def get_physical_resources(backend):
1099
    """ Get the physical resources of a backend.
1100

1101
    Get the resources of a backend as reported by the backend (not the db).
1102

1103
    """
1104
    nodes = get_nodes(backend, bulk=True)
1105
    attr = ['mfree', 'mtotal', 'dfree', 'dtotal', 'pinst_cnt', 'ctotal']
1106
    res = {}
1107
    for a in attr:
1108
        res[a] = 0
1109
    for n in nodes:
1110
        # Filter out drained, offline and not vm_capable nodes since they will
1111
        # not take part in the vm allocation process
1112
        can_host_vms = n['vm_capable'] and not (n['drained'] or n['offline'])
1113
        if can_host_vms and n['cnodes']:
1114
            for a in attr:
1115
                res[a] += int(n[a] or 0)
1116
    return res
1117

    
1118

    
1119
def update_backend_resources(backend, resources=None):
1120
    """ Update the state of the backend resources in db.
1121

1122
    """
1123

    
1124
    if not resources:
1125
        resources = get_physical_resources(backend)
1126

    
1127
    backend.mfree = resources['mfree']
1128
    backend.mtotal = resources['mtotal']
1129
    backend.dfree = resources['dfree']
1130
    backend.dtotal = resources['dtotal']
1131
    backend.pinst_cnt = resources['pinst_cnt']
1132
    backend.ctotal = resources['ctotal']
1133
    backend.updated = datetime.now()
1134
    backend.save()
1135

    
1136

    
1137
def get_memory_from_instances(backend):
1138
    """ Get the memory that is used from instances.
1139

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

1143
    """
1144
    with pooled_rapi_client(backend) as client:
1145
        instances = client.GetInstances(bulk=True)
1146
    mem = 0
1147
    for i in instances:
1148
        mem += i['oper_ram']
1149
    return mem
1150

    
1151

    
1152
def get_available_disk_templates(backend):
1153
    """Get the list of available disk templates of a Ganeti backend.
1154

1155
    The list contains the disk templates that are enabled in the Ganeti backend
1156
    and also included in ipolicy-disk-templates.
1157

1158
    """
1159
    with pooled_rapi_client(backend) as c:
1160
        info = c.GetInfo()
1161
    ipolicy_disk_templates = info["ipolicy"]["disk-templates"]
1162
    try:
1163
        enabled_disk_templates = info["enabled_disk_templates"]
1164
        return [dp for dp in enabled_disk_templates
1165
                if dp in ipolicy_disk_templates]
1166
    except KeyError:
1167
        # Ganeti < 2.8 does not have 'enabled_disk_templates'
1168
        return ipolicy_disk_templates
1169

    
1170

    
1171
def update_backend_disk_templates(backend):
1172
    disk_templates = get_available_disk_templates(backend)
1173
    backend.disk_templates = disk_templates
1174
    backend.save()
1175

    
1176

    
1177
##
1178
## Synchronized operations for reconciliation
1179
##
1180

    
1181

    
1182
def create_network_synced(network, backend):
1183
    result = _create_network_synced(network, backend)
1184
    if result[0] != rapi.JOB_STATUS_SUCCESS:
1185
        return result
1186
    result = connect_network_synced(network, backend)
1187
    return result
1188

    
1189

    
1190
def _create_network_synced(network, backend):
1191
    with pooled_rapi_client(backend) as client:
1192
        job = _create_network(network, backend)
1193
        result = wait_for_job(client, job)
1194
    return result
1195

    
1196

    
1197
def connect_network_synced(network, backend):
1198
    with pooled_rapi_client(backend) as client:
1199
        for group in client.GetGroups():
1200
            job = client.ConnectNetwork(network.backend_id, group,
1201
                                        network.mode, network.link)
1202
            result = wait_for_job(client, job)
1203
            if result[0] != rapi.JOB_STATUS_SUCCESS:
1204
                return result
1205

    
1206
    return result
1207

    
1208

    
1209
def wait_for_job(client, jobid):
1210
    result = client.WaitForJobChange(jobid, ['status', 'opresult'], None, None)
1211
    status = result['job_info'][0]
1212
    while status not in rapi.JOB_STATUS_FINALIZED:
1213
        result = client.WaitForJobChange(jobid, ['status', 'opresult'],
1214
                                         [result], None)
1215
        status = result['job_info'][0]
1216

    
1217
    if status == rapi.JOB_STATUS_SUCCESS:
1218
        return (status, None)
1219
    else:
1220
        error = result['job_info'][1]
1221
        return (status, error)
1222

    
1223

    
1224
def create_job_dependencies(job_ids=[], job_states=None):
1225
    """Transform a list of job IDs to Ganeti 'depends' attribute."""
1226
    if job_states is None:
1227
        job_states = list(rapi.JOB_STATUS_FINALIZED)
1228
    assert(type(job_states) == list)
1229
    return [[job_id, job_states] for job_id in job_ids]