Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (40.6 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
# Timeout in seconds for building NICs. After this period the NICs considered
59
# stale and removed from DB.
60
BUILDING_NIC_TIMEOUT = timedelta(seconds=180)
61

    
62
SIMPLE_NIC_FIELDS = ["state", "mac", "network", "firewall_profile", "index"]
63
COMPLEX_NIC_FIELDS = ["ipv4_address", "ipv6_address"]
64
NIC_FIELDS = SIMPLE_NIC_FIELDS + COMPLEX_NIC_FIELDS
65
UNKNOWN_NIC_PREFIX = "unknown-"
66

    
67

    
68
def handle_vm_quotas(vm, job_id, job_opcode, job_status, job_fields):
69
    """Handle quotas for updated VirtualMachine.
70

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

78
    """
79
    if job_status not in rapi.JOB_STATUS_FINALIZED:
80
        return vm
81

    
82
    # Check successful completion of a job will trigger any quotable change in
83
    # the VM state.
84
    action = utils.get_action_from_opcode(job_opcode, job_fields)
85
    if action == "BUILD":
86
        # Quotas for new VMs are automatically accepted by the API
87
        return vm
88
    commission_info = quotas.get_commission_info(vm, action=action,
89
                                                 action_fields=job_fields)
90

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

    
119
    return vm
120

    
121

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

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

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

    
137
    vm.backendjobid = jobid
138
    vm.backendjobstatus = status
139
    vm.backendopcode = opcode
140
    vm.backendlogmsg = logmsg
141

    
142
    if status not in rapi.JOB_STATUS_FINALIZED:
143
        vm.save()
144
        return
145

    
146
    if job_fields is None:
147
        job_fields = {}
148
    state_for_success = VirtualMachine.OPER_STATE_FROM_OPCODE.get(opcode)
149

    
150
    # Notifications of success change the operating state
151
    if status == rapi.JOB_STATUS_SUCCESS:
152
        if state_for_success is not None:
153
            vm.operstate = state_for_success
154
        beparams = job_fields.get("beparams", None)
155
        if beparams:
156
            # Change the flavor of the VM
157
            _process_resize(vm, beparams)
158
        # Update backendtime only for jobs that have been successfully
159
        # completed, since only these jobs update the state of the VM. Else a
160
        # "race condition" may occur when a successful job (e.g.
161
        # OP_INSTANCE_REMOVE) completes before an error job and messages arrive
162
        # in reversed order.
163
        vm.backendtime = etime
164

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

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

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

    
202
    vm.save()
203

    
204

    
205
def _process_resize(vm, beparams):
206
    """Change flavor of a VirtualMachine based on new beparams."""
207
    old_flavor = vm.flavor
208
    vcpus = beparams.get("vcpus", old_flavor.cpu)
209
    ram = beparams.get("maxmem", old_flavor.ram)
210
    if vcpus == old_flavor.cpu and ram == old_flavor.ram:
211
        return
212
    try:
213
        new_flavor = Flavor.objects.get(cpu=vcpus, ram=ram,
214
                                        disk=old_flavor.disk,
215
                                        disk_template=old_flavor.disk_template)
216
    except Flavor.DoesNotExist:
217
        raise Exception("Cannot find flavor for VM")
218
    vm.flavor = new_flavor
219
    vm.save()
220

    
221

    
222
@transaction.commit_on_success
223
def process_net_status(vm, etime, nics):
224
    """Wrap _process_net_status inside transaction."""
225
    _process_net_status(vm, etime, nics)
226

    
227

    
228
def _process_net_status(vm, etime, nics):
229
    """Process a net status notification from the backend
230

231
    Process an incoming message from the Ganeti backend,
232
    detailing the NIC configuration of a VM instance.
233

234
    Update the state of the VM in the DB accordingly.
235

236
    """
237
    ganeti_nics = process_ganeti_nics(nics)
238
    db_nics = dict([(nic.id, nic)
239
                    for nic in vm.nics.prefetch_related("ips__subnet")])
240

    
241
    # Get X-Lock on backend before getting X-Lock on network IP pools, to
242
    # guarantee that no deadlock will occur with Backend allocator.
243
    Backend.objects.select_for_update().get(id=vm.backend_id)
244

    
245
    for nic_name in set(db_nics.keys()) | set(ganeti_nics.keys()):
246
        db_nic = db_nics.get(nic_name)
247
        ganeti_nic = ganeti_nics.get(nic_name)
248
        if ganeti_nic is None:
249
            # NIC exists in DB but not in Ganeti. If the NIC is in 'building'
250
            # state for more than 5 minutes, then we remove the NIC.
251
            # TODO: This is dangerous as the job may be stack in the queue, and
252
            # releasing the IP may lead to duplicate IP use.
253
            if db_nic.state != "BUILD" or\
254
                (db_nic.state == "BUILD" and
255
                 etime > db_nic.created + BUILDING_NIC_TIMEOUT):
256
                remove_nic_ips(db_nic)
257
                db_nic.delete()
258
            else:
259
                log.warning("Ignoring recent building NIC: %s", db_nic)
260
        elif db_nic is None:
261
            msg = ("NIC/%s of VM %s does not exist in DB! Cannot automatically"
262
                   " fix this issue!" % (nic_name, vm))
263
            log.error(msg)
264
            continue
265
        elif not nics_are_equal(db_nic, ganeti_nic):
266
            for f in SIMPLE_NIC_FIELDS:
267
                # Update the NIC in DB with the values from Ganeti NIC
268
                setattr(db_nic, f, ganeti_nic[f])
269
                db_nic.save()
270

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

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

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

    
290

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

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

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

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

    
323
    return ipaddress
324

    
325

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

    
332

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

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

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

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

    
368
        new_nics.append((nic_id, nic_info))
369
    return dict(new_nics)
370

    
371

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

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

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

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

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

    
397

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

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

    
424

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

    
430
    back_network.backendjobid = jobid
431
    back_network.backendjobstatus = status
432
    back_network.backendopcode = opcode
433
    back_network.backendlogmsg = logmsg
434

    
435
    # Note: Network is already locked!
436
    network = back_network.network
437

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

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

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

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

    
462

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

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

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

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

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

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

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

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

    
514
        # Set all subnets as deleted
515
        network.subnets.update(deleted=True)
516
        # And delete the IP pools
517
        for subnet in network.subnets.all():
518
            if subnet.ipversion == 4:
519
                subnet.ip_pools.all().delete()
520
        # And all the backend networks since there are useless
521
        network.backend_networks.all().delete()
522

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

    
534

    
535
@transaction.commit_on_success
536
def process_network_modify(back_network, etime, jobid, opcode, status,
537
                           job_fields):
538
    assert (opcode == "OP_NETWORK_SET_PARAMS")
539
    if status not in [x[0] for x in BACKEND_STATUSES]:
540
        raise Network.InvalidBackendMsgError(opcode, status)
541

    
542
    back_network.backendjobid = jobid
543
    back_network.backendjobstatus = status
544
    back_network.opcode = opcode
545

    
546
    add_reserved_ips = job_fields.get("add_reserved_ips")
547
    if add_reserved_ips:
548
        network = back_network.network
549
        for ip in add_reserved_ips:
550
            network.reserve_address(ip, external=True)
551

    
552
    if status == rapi.JOB_STATUS_SUCCESS:
553
        back_network.backendtime = etime
554
    back_network.save()
555

    
556

    
557
@transaction.commit_on_success
558
def process_create_progress(vm, etime, progress):
559

    
560
    percentage = int(progress)
561

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

    
568
    # FIXME: log a warning here, see #1033
569
#   if last_update > percentage:
570
#       raise ValueError("Build percentage should increase monotonically " \
571
#                        "(old = %d, new = %d)" % (last_update, percentage))
572

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

    
583
    vm.buildpercentage = percentage
584
    vm.backendtime = etime
585
    vm.save()
586

    
587

    
588
@transaction.commit_on_success
589
def create_instance_diagnostic(vm, message, source, level="DEBUG", etime=None,
590
                               details=None):
591
    """
592
    Create virtual machine instance diagnostic entry.
593

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

    
606

    
607
def create_instance(vm, nics, flavor, image):
608
    """`image` is a dictionary which should contain the keys:
609
            'backend_id', 'format' and 'metadata'
610

611
        metadata value should be a dictionary.
612
    """
613

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

    
625
    kw['disk_template'] = flavor.disk_template
626
    kw['disks'] = [{"size": flavor.disk * 1024}]
627
    provider = flavor.disk_provider
628
    if provider:
629
        kw['disks'][0]['provider'] = provider
630
        kw['disks'][0]['origin'] = flavor.disk_origin
631

    
632
    kw['nics'] = [{"name": nic.backend_uuid,
633
                   "network": nic.network.backend_id,
634
                   "ip": nic.ipv4_address}
635
                  for nic in nics]
636

    
637
    backend = vm.backend
638
    depend_jobs = []
639
    for nic in nics:
640
        bnet, job_ids = ensure_network_is_active(backend, nic.network_id)
641
        depend_jobs.extend(job_ids)
642

    
643
    kw["depends"] = create_job_dependencies(depend_jobs)
644

    
645
    # Defined in settings.GANETI_CREATEINSTANCE_KWARGS
646
    # kw['os'] = settings.GANETI_OS_PROVIDER
647
    kw['ip_check'] = False
648
    kw['name_check'] = False
649

    
650
    # Do not specific a node explicitly, have
651
    # Ganeti use an iallocator instead
652
    #kw['pnode'] = rapi.GetNodes()[0]
653

    
654
    kw['dry_run'] = settings.TEST
655

    
656
    kw['beparams'] = {
657
        'auto_balance': True,
658
        'vcpus': flavor.cpu,
659
        'memory': flavor.ram}
660

    
661
    kw['osparams'] = {
662
        'config_url': vm.config_url,
663
        # Store image id and format to Ganeti
664
        'img_id': image['backend_id'],
665
        'img_format': image['format']}
666

    
667
    # Use opportunistic locking
668
    kw['opportunistic_locking'] = True
669

    
670
    # Defined in settings.GANETI_CREATEINSTANCE_KWARGS
671
    # kw['hvparams'] = dict(serial_console=False)
672

    
673
    log.debug("Creating instance %s", utils.hide_pass(kw))
674
    with pooled_rapi_client(vm) as client:
675
        return client.CreateInstance(**kw)
676

    
677

    
678
def delete_instance(vm):
679
    with pooled_rapi_client(vm) as client:
680
        return client.DeleteInstance(vm.backend_vm_id, dry_run=settings.TEST)
681

    
682

    
683
def reboot_instance(vm, reboot_type):
684
    assert reboot_type in ('soft', 'hard')
685
    kwargs = {"instance": vm.backend_vm_id,
686
              "reboot_type": "hard"}
687
    # XXX: Currently shutdown_timeout parameter is not supported from the
688
    # Ganeti RAPI. Until supported, we will fallback for both reboot types
689
    # to the default shutdown timeout of Ganeti (120s). Note that reboot
690
    # type of Ganeti job must be always hard. The 'soft' and 'hard' type
691
    # of OS API is different from the one in Ganeti, and maps to
692
    # 'shutdown_timeout'.
693
    #if reboot_type == "hard":
694
    #    kwargs["shutdown_timeout"] = 0
695
    if settings.TEST:
696
        kwargs["dry_run"] = True
697
    with pooled_rapi_client(vm) as client:
698
        return client.RebootInstance(**kwargs)
699

    
700

    
701
def startup_instance(vm):
702
    with pooled_rapi_client(vm) as client:
703
        return client.StartupInstance(vm.backend_vm_id, dry_run=settings.TEST)
704

    
705

    
706
def shutdown_instance(vm):
707
    with pooled_rapi_client(vm) as client:
708
        return client.ShutdownInstance(vm.backend_vm_id, dry_run=settings.TEST)
709

    
710

    
711
def resize_instance(vm, vcpus, memory):
712
    beparams = {"vcpus": int(vcpus),
713
                "minmem": int(memory),
714
                "maxmem": int(memory)}
715
    with pooled_rapi_client(vm) as client:
716
        return client.ModifyInstance(vm.backend_vm_id, beparams=beparams)
717

    
718

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

    
733
    console = {}
734
    console['kind'] = 'vnc'
735

    
736
    with pooled_rapi_client(vm) as client:
737
        i = client.GetInstance(vm.backend_vm_id)
738

    
739
    if vm.backend.hypervisor == "kvm" and i['hvparams']['serial_console']:
740
        raise Exception("hv parameter serial_console cannot be true")
741
    console['host'] = i['pnode']
742
    console['port'] = i['network_port']
743

    
744
    return console
745

    
746

    
747
def get_instance_info(vm):
748
    with pooled_rapi_client(vm) as client:
749
        return client.GetInstance(vm.backend_vm_id)
750

    
751

    
752
def vm_exists_in_backend(vm):
753
    try:
754
        get_instance_info(vm)
755
        return True
756
    except rapi.GanetiApiError as e:
757
        if e.code == 404:
758
            return False
759
        raise e
760

    
761

    
762
def get_network_info(backend_network):
763
    with pooled_rapi_client(backend_network) as client:
764
        return client.GetNetwork(backend_network.network.backend_id)
765

    
766

    
767
def network_exists_in_backend(backend_network):
768
    try:
769
        get_network_info(backend_network)
770
        return True
771
    except rapi.GanetiApiError as e:
772
        if e.code == 404:
773
            return False
774

    
775

    
776
def ensure_network_is_active(backend, network_id):
777
    """Ensure that a network is active in the specified backend
778

779
    Check that a network exists and is active in the specified backend. If not
780
    (re-)create the network. Return the corresponding BackendNetwork object
781
    and the IDs of the Ganeti job to create the network.
782

783
    """
784
    network = Network.objects.select_for_update().get(id=network_id)
785
    bnet, created = BackendNetwork.objects.get_or_create(backend=backend,
786
                                                         network=network)
787
    job_ids = []
788
    if bnet.operstate != "ACTIVE":
789
        job_ids = create_network(network, backend, connect=True)
790

    
791
    return bnet, job_ids
792

    
793

    
794
def create_network(network, backend, connect=True):
795
    """Create a network in a Ganeti backend"""
796
    log.debug("Creating network %s in backend %s", network, backend)
797

    
798
    job_id = _create_network(network, backend)
799

    
800
    if connect:
801
        job_ids = connect_network(network, backend, depends=[job_id])
802
        return job_ids
803
    else:
804
        return [job_id]
805

    
806

    
807
def _create_network(network, backend):
808
    """Create a network."""
809

    
810
    tags = network.backend_tag
811
    subnet = None
812
    subnet6 = None
813
    gateway = None
814
    gateway6 = None
815
    for _subnet in network.subnets.all():
816
        if _subnet.dhcp and not "nfdhcpd" in tags:
817
            tags.append("nfdhcpd")
818
        if _subnet.ipversion == 4:
819
            subnet = _subnet.cidr
820
            gateway = _subnet.gateway
821
        elif _subnet.ipversion == 6:
822
            subnet6 = _subnet.cidr
823
            gateway6 = _subnet.gateway
824

    
825
    if network.public:
826
        conflicts_check = True
827
        tags.append('public')
828
    else:
829
        conflicts_check = False
830
        tags.append('private')
831

    
832
    # Use a dummy network subnet for IPv6 only networks. Currently Ganeti does
833
    # not support IPv6 only networks. To bypass this limitation, we create the
834
    # network with a dummy network subnet, and make Cyclades connect instances
835
    # to such networks, with address=None.
836
    if subnet is None:
837
        subnet = "10.0.0.0/30"
838

    
839
    try:
840
        bn = BackendNetwork.objects.get(network=network, backend=backend)
841
        mac_prefix = bn.mac_prefix
842
    except BackendNetwork.DoesNotExist:
843
        raise Exception("BackendNetwork for network '%s' in backend '%s'"
844
                        " does not exist" % (network.id, backend.id))
845

    
846
    with pooled_rapi_client(backend) as client:
847
        return client.CreateNetwork(network_name=network.backend_id,
848
                                    network=subnet,
849
                                    network6=subnet6,
850
                                    gateway=gateway,
851
                                    gateway6=gateway6,
852
                                    mac_prefix=mac_prefix,
853
                                    conflicts_check=conflicts_check,
854
                                    tags=tags)
855

    
856

    
857
def connect_network(network, backend, depends=[], group=None):
858
    """Connect a network to nodegroups."""
859
    log.debug("Connecting network %s to backend %s", network, backend)
860

    
861
    if network.public:
862
        conflicts_check = True
863
    else:
864
        conflicts_check = False
865

    
866
    depends = create_job_dependencies(depends)
867
    with pooled_rapi_client(backend) as client:
868
        groups = [group] if group is not None else client.GetGroups()
869
        job_ids = []
870
        for group in groups:
871
            job_id = client.ConnectNetwork(network.backend_id, group,
872
                                           network.mode, network.link,
873
                                           conflicts_check,
874
                                           depends=depends)
875
            job_ids.append(job_id)
876
    return job_ids
877

    
878

    
879
def delete_network(network, backend, disconnect=True):
880
    log.debug("Deleting network %s from backend %s", network, backend)
881

    
882
    depends = []
883
    if disconnect:
884
        depends = disconnect_network(network, backend)
885
    _delete_network(network, backend, depends=depends)
886

    
887

    
888
def _delete_network(network, backend, depends=[]):
889
    depends = create_job_dependencies(depends)
890
    with pooled_rapi_client(backend) as client:
891
        return client.DeleteNetwork(network.backend_id, depends)
892

    
893

    
894
def disconnect_network(network, backend, group=None):
895
    log.debug("Disconnecting network %s to backend %s", network, backend)
896

    
897
    with pooled_rapi_client(backend) as client:
898
        groups = [group] if group is not None else client.GetGroups()
899
        job_ids = []
900
        for group in groups:
901
            job_id = client.DisconnectNetwork(network.backend_id, group)
902
            job_ids.append(job_id)
903
    return job_ids
904

    
905

    
906
def connect_to_network(vm, nic):
907
    network = nic.network
908
    backend = vm.backend
909
    bnet, depend_jobs = ensure_network_is_active(backend, network.id)
910

    
911
    depends = create_job_dependencies(depend_jobs)
912

    
913
    nic = {'name': nic.backend_uuid,
914
           'network': network.backend_id,
915
           'ip': nic.ipv4_address}
916

    
917
    log.debug("Adding NIC %s to VM %s", nic, vm)
918

    
919
    kwargs = {
920
        "instance": vm.backend_vm_id,
921
        "nics": [("add", "-1", nic)],
922
        "depends": depends,
923
    }
924
    if vm.backend.use_hotplug():
925
        kwargs["hotplug"] = True
926
    if settings.TEST:
927
        kwargs["dry_run"] = True
928

    
929
    with pooled_rapi_client(vm) as client:
930
        return client.ModifyInstance(**kwargs)
931

    
932

    
933
def disconnect_from_network(vm, nic):
934
    log.debug("Removing NIC %s of VM %s", nic, vm)
935

    
936
    kwargs = {
937
        "instance": vm.backend_vm_id,
938
        "nics": [("remove", nic.backend_uuid, {})],
939
    }
940
    if vm.backend.use_hotplug():
941
        kwargs["hotplug"] = True
942
    if settings.TEST:
943
        kwargs["dry_run"] = True
944

    
945
    with pooled_rapi_client(vm) as client:
946
        jobID = client.ModifyInstance(**kwargs)
947
        firewall_profile = nic.firewall_profile
948
        if firewall_profile and firewall_profile != "DISABLED":
949
            tag = _firewall_tags[firewall_profile] % nic.backend_uuid
950
            client.DeleteInstanceTags(vm.backend_vm_id, [tag],
951
                                      dry_run=settings.TEST)
952

    
953
        return jobID
954

    
955

    
956
def set_firewall_profile(vm, profile, nic):
957
    uuid = nic.backend_uuid
958
    try:
959
        tag = _firewall_tags[profile] % uuid
960
    except KeyError:
961
        raise ValueError("Unsopported Firewall Profile: %s" % profile)
962

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

    
965
    with pooled_rapi_client(vm) as client:
966
        # Delete previous firewall tags
967
        old_tags = client.GetInstanceTags(vm.backend_vm_id)
968
        delete_tags = [(t % uuid) for t in _firewall_tags.values()
969
                       if (t % uuid) in old_tags]
970
        if delete_tags:
971
            client.DeleteInstanceTags(vm.backend_vm_id, delete_tags,
972
                                      dry_run=settings.TEST)
973

    
974
        if profile != "DISABLED":
975
            client.AddInstanceTags(vm.backend_vm_id, [tag],
976
                                   dry_run=settings.TEST)
977

    
978
        # XXX NOP ModifyInstance call to force process_net_status to run
979
        # on the dispatcher
980
        os_name = settings.GANETI_CREATEINSTANCE_KWARGS['os']
981
        client.ModifyInstance(vm.backend_vm_id,
982
                              os_name=os_name)
983
    return None
984

    
985

    
986
def get_instances(backend, bulk=True):
987
    with pooled_rapi_client(backend) as c:
988
        return c.GetInstances(bulk=bulk)
989

    
990

    
991
def get_nodes(backend, bulk=True):
992
    with pooled_rapi_client(backend) as c:
993
        return c.GetNodes(bulk=bulk)
994

    
995

    
996
def get_jobs(backend, bulk=True):
997
    with pooled_rapi_client(backend) as c:
998
        return c.GetJobs(bulk=bulk)
999

    
1000

    
1001
def get_physical_resources(backend):
1002
    """ Get the physical resources of a backend.
1003

1004
    Get the resources of a backend as reported by the backend (not the db).
1005

1006
    """
1007
    nodes = get_nodes(backend, bulk=True)
1008
    attr = ['mfree', 'mtotal', 'dfree', 'dtotal', 'pinst_cnt', 'ctotal']
1009
    res = {}
1010
    for a in attr:
1011
        res[a] = 0
1012
    for n in nodes:
1013
        # Filter out drained, offline and not vm_capable nodes since they will
1014
        # not take part in the vm allocation process
1015
        can_host_vms = n['vm_capable'] and not (n['drained'] or n['offline'])
1016
        if can_host_vms and n['cnodes']:
1017
            for a in attr:
1018
                res[a] += int(n[a] or 0)
1019
    return res
1020

    
1021

    
1022
def update_backend_resources(backend, resources=None):
1023
    """ Update the state of the backend resources in db.
1024

1025
    """
1026

    
1027
    if not resources:
1028
        resources = get_physical_resources(backend)
1029

    
1030
    backend.mfree = resources['mfree']
1031
    backend.mtotal = resources['mtotal']
1032
    backend.dfree = resources['dfree']
1033
    backend.dtotal = resources['dtotal']
1034
    backend.pinst_cnt = resources['pinst_cnt']
1035
    backend.ctotal = resources['ctotal']
1036
    backend.updated = datetime.now()
1037
    backend.save()
1038

    
1039

    
1040
def get_memory_from_instances(backend):
1041
    """ Get the memory that is used from instances.
1042

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

1046
    """
1047
    with pooled_rapi_client(backend) as client:
1048
        instances = client.GetInstances(bulk=True)
1049
    mem = 0
1050
    for i in instances:
1051
        mem += i['oper_ram']
1052
    return mem
1053

    
1054

    
1055
def get_available_disk_templates(backend):
1056
    """Get the list of available disk templates of a Ganeti backend.
1057

1058
    The list contains the disk templates that are enabled in the Ganeti backend
1059
    and also included in ipolicy-disk-templates.
1060

1061
    """
1062
    with pooled_rapi_client(backend) as c:
1063
        info = c.GetInfo()
1064
    ipolicy_disk_templates = info["ipolicy"]["disk-templates"]
1065
    try:
1066
        enabled_disk_templates = info["enabled_disk_templates"]
1067
        return [dp for dp in enabled_disk_templates
1068
                if dp in ipolicy_disk_templates]
1069
    except KeyError:
1070
        # Ganeti < 2.8 does not have 'enabled_disk_templates'
1071
        return ipolicy_disk_templates
1072

    
1073

    
1074
def update_backend_disk_templates(backend):
1075
    disk_templates = get_available_disk_templates(backend)
1076
    backend.disk_templates = disk_templates
1077
    backend.save()
1078

    
1079

    
1080
##
1081
## Synchronized operations for reconciliation
1082
##
1083

    
1084

    
1085
def create_network_synced(network, backend):
1086
    result = _create_network_synced(network, backend)
1087
    if result[0] != rapi.JOB_STATUS_SUCCESS:
1088
        return result
1089
    result = connect_network_synced(network, backend)
1090
    return result
1091

    
1092

    
1093
def _create_network_synced(network, backend):
1094
    with pooled_rapi_client(backend) as client:
1095
        job = _create_network(network, backend)
1096
        result = wait_for_job(client, job)
1097
    return result
1098

    
1099

    
1100
def connect_network_synced(network, backend):
1101
    with pooled_rapi_client(backend) as client:
1102
        for group in client.GetGroups():
1103
            job = client.ConnectNetwork(network.backend_id, group,
1104
                                        network.mode, network.link)
1105
            result = wait_for_job(client, job)
1106
            if result[0] != rapi.JOB_STATUS_SUCCESS:
1107
                return result
1108

    
1109
    return result
1110

    
1111

    
1112
def wait_for_job(client, jobid):
1113
    result = client.WaitForJobChange(jobid, ['status', 'opresult'], None, None)
1114
    status = result['job_info'][0]
1115
    while status not in rapi.JOB_STATUS_FINALIZED:
1116
        result = client.WaitForJobChange(jobid, ['status', 'opresult'],
1117
                                         [result], None)
1118
        status = result['job_info'][0]
1119

    
1120
    if status == rapi.JOB_STATUS_SUCCESS:
1121
        return (status, None)
1122
    else:
1123
        error = result['job_info'][1]
1124
        return (status, error)
1125

    
1126

    
1127
def create_job_dependencies(job_ids=[], job_states=None):
1128
    """Transform a list of job IDs to Ganeti 'depends' attribute."""
1129
    if job_states is None:
1130
        job_states = list(rapi.JOB_STATUS_FINALIZED)
1131
    assert(type(job_states) == list)
1132
    return [[job_id, job_states] for job_id in job_ids]