Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / logic / servers.py @ ae2da8a2

History | View | Annotate | Download (19.9 kB)

1
import logging
2

    
3
from socket import getfqdn
4
from functools import wraps
5
from django import dispatch
6
from django.db import transaction
7
from django.utils import simplejson as json
8

    
9
from snf_django.lib.api import faults
10
from django.conf import settings
11
from synnefo import quotas
12
from synnefo.api import util
13
from synnefo.logic import backend
14
from synnefo.logic.backend_allocator import BackendAllocator
15
from synnefo.db import pools
16
from synnefo.db.models import (NetworkInterface, VirtualMachine,
17
                               VirtualMachineMetadata, IPAddressLog)
18
from vncauthproxy.client import request_forwarding as request_vnc_forwarding
19

    
20
log = logging.getLogger(__name__)
21

    
22
# server creation signal
23
server_created = dispatch.Signal(providing_args=["created_vm_params"])
24

    
25

    
26
def validate_server_action(vm, action):
27
    if vm.deleted:
28
        raise faults.BadRequest("Server '%s' has been deleted." % vm.id)
29

    
30
    # Destroyin a server should always be permitted
31
    if action == "DESTROY":
32
        return
33

    
34
    # Check that there is no pending action
35
    pending_action = vm.task
36
    if pending_action:
37
        if pending_action == "BUILD":
38
            raise faults.BuildInProgress("Server '%s' is being build." % vm.id)
39
        raise faults.BadRequest("Can not perform '%s' action while there is a"
40
                                " pending '%s'." % (action, pending_action))
41

    
42
    # Check if action can be performed to VM's operstate
43
    operstate = vm.operstate
44
    if operstate == "BUILD" and action != "BUILD":
45
        raise faults.BuildInProgress("Server '%s' is being build." % vm.id)
46
    elif (action == "START" and operstate != "STOPPED") or\
47
         (action == "STOP" and operstate != "STARTED") or\
48
         (action == "RESIZE" and operstate != "STOPPED") or\
49
         (action in ["CONNECT", "DISCONNECT"] and operstate != "STOPPED"
50
          and not settings.GANETI_USE_HOTPLUG):
51
        raise faults.BadRequest("Can not perform '%s' action while server is"
52
                                " in '%s' state." % (action, operstate))
53
    return
54

    
55

    
56
def server_command(action):
57
    """Handle execution of a server action.
58

59
    Helper function to validate and execute a server action, handle quota
60
    commission and update the 'task' of the VM in the DB.
61

62
    1) Check if action can be performed. If it can, then there must be no
63
       pending task (with the exception of DESTROY).
64
    2) Handle previous commission if unresolved:
65
       * If it is not pending and it to accept, then accept
66
       * If it is not pending and to reject or is pending then reject it. Since
67
       the action can be performed only if there is no pending task, then there
68
       can be no pending commission. The exception is DESTROY, but in this case
69
       the commission can safely be rejected, and the dispatcher will generate
70
       the correct ones!
71
    3) Issue new commission and associate it with the VM. Also clear the task.
72
    4) Send job to ganeti
73
    5) Update task and commit
74
    """
75
    def decorator(func):
76
        @wraps(func)
77
        @transaction.commit_on_success
78
        def wrapper(vm, *args, **kwargs):
79
            user_id = vm.userid
80
            validate_server_action(vm, action)
81
            vm.action = action
82

    
83
            commission_name = "client: api, resource: %s" % vm
84
            quotas.handle_resource_commission(vm, action=action,
85
                                              commission_name=commission_name)
86
            vm.save()
87

    
88
            # XXX: Special case for server creation!
89
            if action == "BUILD":
90
                # Perform a commit, because the VirtualMachine must be saved to
91
                # DB before the OP_INSTANCE_CREATE job in enqueued in Ganeti.
92
                # Otherwise, messages will arrive from snf-dispatcher about
93
                # this instance, before the VM is stored in DB.
94
                transaction.commit()
95
                # After committing the locks are released. Refetch the instance
96
                # to guarantee x-lock.
97
                vm = VirtualMachine.objects.select_for_update().get(id=vm.id)
98

    
99
            # Send the job to Ganeti and get the associated jobID
100
            try:
101
                job_id = func(vm, *args, **kwargs)
102
            except Exception as e:
103
                if vm.serial is not None:
104
                    # Since the job never reached Ganeti, reject the commission
105
                    log.debug("Rejecting commission: '%s', could not perform"
106
                              " action '%s': %s" % (vm.serial,  action, e))
107
                    transaction.rollback()
108
                    quotas.reject_serial(vm.serial)
109
                    transaction.commit()
110
                raise
111

    
112
            if action == "BUILD" and vm.serial is not None:
113
                # XXX: Special case for server creation: we must accept the
114
                # commission because the VM has been stored in DB. Also, if
115
                # communication with Ganeti fails, the job will never reach
116
                # Ganeti, and the commission will never be resolved.
117
                quotas.accept_serial(vm.serial)
118

    
119
            log.info("user: %s, vm: %s, action: %s, job_id: %s, serial: %s",
120
                     user_id, vm.id, action, job_id, vm.serial)
121

    
122
            # store the new task in the VM
123
            if job_id is not None:
124
                vm.task = action
125
                vm.task_job_id = job_id
126
            vm.save()
127

    
128
            return vm
129
        return wrapper
130
    return decorator
131

    
132

    
133
@transaction.commit_on_success
134
def create(userid, name, password, flavor, image, metadata={},
135
           personality=[], private_networks=None, floating_ips=None,
136
           use_backend=None):
137
    if use_backend is None:
138
        # Allocate server to a Ganeti backend
139
        use_backend = allocate_new_server(userid, flavor)
140

    
141
    if private_networks is None:
142
        private_networks = []
143
    if floating_ips is None:
144
        floating_ips = []
145

    
146
    # Fix flavor for archipelago
147
    disk_template, provider = util.get_flavor_provider(flavor)
148
    if provider:
149
        flavor.disk_template = disk_template
150
        flavor.disk_provider = provider
151
        flavor.disk_origin = None
152
        if provider == 'vlmc':
153
            flavor.disk_origin = image['checksum']
154
            image['backend_id'] = 'null'
155
    else:
156
        flavor.disk_provider = None
157

    
158
    # We must save the VM instance now, so that it gets a valid
159
    # vm.backend_vm_id.
160
    vm = VirtualMachine.objects.create(name=name,
161
                                       backend=use_backend,
162
                                       userid=userid,
163
                                       imageid=image["id"],
164
                                       flavor=flavor,
165
                                       operstate="BUILD")
166
    log.info("Created entry in DB for VM '%s'", vm)
167

    
168
    nics = create_instance_nics(vm, userid, private_networks, floating_ips)
169

    
170
    for key, val in metadata.items():
171
        VirtualMachineMetadata.objects.create(
172
            meta_key=key,
173
            meta_value=val,
174
            vm=vm)
175

    
176
    # Create the server in Ganeti.
177
    vm = create_server(vm, nics, flavor, image, personality, password)
178

    
179
    return vm
180

    
181

    
182
@transaction.commit_on_success
183
def allocate_new_server(userid, flavor):
184
    """Allocate a new server to a Ganeti backend.
185

186
    Allocation is performed based on the owner of the server and the specified
187
    flavor. Also, backends that do not have a public IPv4 address are excluded
188
    from server allocation.
189

190
    This function runs inside a transaction, because after allocating the
191
    instance a commit must be performed in order to release all locks.
192

193
    """
194
    backend_allocator = BackendAllocator()
195
    use_backend = backend_allocator.allocate(userid, flavor)
196
    if use_backend is None:
197
        log.error("No available backend for VM with flavor %s", flavor)
198
        raise faults.ServiceUnavailable("No available backends")
199
    return use_backend
200

    
201

    
202
@server_command("BUILD")
203
def create_server(vm, nics, flavor, image, personality, password):
204
    # dispatch server created signal needed to trigger the 'vmapi', which
205
    # enriches the vm object with the 'config_url' attribute which must be
206
    # passed to the Ganeti job.
207
    server_created.send(sender=vm, created_vm_params={
208
        'img_id': image['backend_id'],
209
        'img_passwd': password,
210
        'img_format': str(image['format']),
211
        'img_personality': json.dumps(personality),
212
        'img_properties': json.dumps(image['metadata']),
213
    })
214
    # send job to Ganeti
215
    try:
216
        jobID = backend.create_instance(vm, nics, flavor, image)
217
    except:
218
        log.exception("Failed create instance '%s'", vm)
219
        jobID = None
220
        vm.operstate = "ERROR"
221
        vm.backendlogmsg = "Failed to send job to Ganeti."
222
        vm.save()
223
        vm.nics.all().update(state="ERROR")
224

    
225
    # At this point the job is enqueued in the Ganeti backend
226
    vm.backendjobid = jobID
227
    vm.save()
228
    log.info("User %s created VM %s, NICs %s, Backend %s, JobID %s",
229
             vm.userid, vm, nics, backend, str(jobID))
230

    
231
    return jobID
232

    
233

    
234
def create_instance_nics(vm, userid, private_networks=[], floating_ips=[]):
235
    """Create NICs for VirtualMachine.
236

237
    Helper function for allocating IP addresses and creating NICs in the DB
238
    for a VirtualMachine. Created NICs are the combination of the default
239
    network policy (defined by administration settings) and the private
240
    networks defined by the user.
241

242
    """
243
    nics = []
244
    for network_id in settings.DEFAULT_INSTANCE_NETWORKS:
245
        if network_id == "SNF:ANY_PUBLIC":
246
            ipaddress = util.allocate_public_ip(userid=userid,
247
                                                backend=vm.backend)
248
            nic = _create_port(userid, network=ipaddress.network,
249
                               use_ipaddress=ipaddress)
250
        else:
251
            try:
252
                network = util.get_network(network_id, userid,
253
                                           non_deleted=True)
254
            except faults.ItemNotFound:
255
                msg = "Invalid configuration. Setting"\
256
                      " 'DEFAULT_INSTANCE_NETWORKS' contains invalid"\
257
                      " network '%s'" % network_id
258
                log.error(msg)
259
                raise faults.InternalServerError(msg)
260
            nic = _create_port(userid, network)
261
        nics.append(nic)
262
    for address in floating_ips:
263
        floating_ip = util.get_floating_ip_by_address(vm.userid, address,
264
                                                      for_update=True)
265
        nic = _create_port(userid, network=floating_ip.network,
266
                           use_ipaddress=floating_ip)
267
        nics.append(nic)
268
    for network_id in private_networks:
269
        network = util.get_network(network_id, userid, non_deleted=True)
270
        if network.public:
271
            raise faults.Forbidden("Can not connect to public network")
272
        nic = _create_port(userid, network)
273
        nics.append(nic)
274
    for index, nic in enumerate(nics):
275
        associate_port_with_machine(nic, vm)
276
        nic.index = index
277
        nic.save()
278
    return nics
279

    
280

    
281
@server_command("DESTROY")
282
def destroy(vm):
283
    log.info("Deleting VM %s", vm)
284
    return backend.delete_instance(vm)
285

    
286

    
287
@server_command("START")
288
def start(vm):
289
    log.info("Starting VM %s", vm)
290
    return backend.startup_instance(vm)
291

    
292

    
293
@server_command("STOP")
294
def stop(vm):
295
    log.info("Stopping VM %s", vm)
296
    return backend.shutdown_instance(vm)
297

    
298

    
299
@server_command("REBOOT")
300
def reboot(vm, reboot_type):
301
    if reboot_type not in ("SOFT", "HARD"):
302
        raise faults.BadRequest("Malformed request. Invalid reboot"
303
                                " type %s" % reboot_type)
304
    log.info("Rebooting VM %s. Type %s", vm, reboot_type)
305

    
306
    return backend.reboot_instance(vm, reboot_type.lower())
307

    
308

    
309
@server_command("RESIZE")
310
def resize(vm, flavor):
311
    old_flavor = vm.flavor
312
    # User requested the same flavor
313
    if old_flavor.id == flavor.id:
314
        raise faults.BadRequest("Server '%s' flavor is already '%s'."
315
                                % (vm, flavor))
316
        return None
317
    # Check that resize can be performed
318
    if old_flavor.disk != flavor.disk:
319
        raise faults.BadRequest("Can not resize instance disk.")
320
    if old_flavor.disk_template != flavor.disk_template:
321
        raise faults.BadRequest("Can not change instance disk template.")
322

    
323
    log.info("Resizing VM from flavor '%s' to '%s", old_flavor, flavor)
324
    commission_info = {"cyclades.cpu": flavor.cpu - old_flavor.cpu,
325
                       "cyclades.ram": 1048576 * (flavor.ram - old_flavor.ram)}
326
    # Save serial to VM, since it is needed by server_command decorator
327
    vm.serial = quotas.issue_commission(user=vm.userid,
328
                                        source=quotas.DEFAULT_SOURCE,
329
                                        provisions=commission_info,
330
                                        name="resource: %s. resize" % vm)
331
    return backend.resize_instance(vm, vcpus=flavor.cpu, memory=flavor.ram)
332

    
333

    
334
@server_command("SET_FIREWALL_PROFILE")
335
def set_firewall_profile(vm, profile, nic):
336
    log.info("Setting VM %s, NIC %s, firewall %s", vm, nic, profile)
337

    
338
    if profile not in [x[0] for x in NetworkInterface.FIREWALL_PROFILES]:
339
        raise faults.BadRequest("Unsupported firewall profile")
340
    backend.set_firewall_profile(vm, profile=profile, nic=nic)
341
    return None
342

    
343

    
344
@server_command("CONNECT")
345
def connect(vm, network, port=None):
346
    if port is None:
347
        port = _create_port(vm.userid, network)
348
    associate_port_with_machine(port, vm)
349

    
350
    log.info("Creating NIC %s with IPv4 Address %s", port, port.ipv4_address)
351

    
352
    return backend.connect_to_network(vm, port)
353

    
354

    
355
@server_command("DISCONNECT")
356
def disconnect(vm, nic):
357
    log.info("Removing NIC %s from VM %s", nic, vm)
358
    return backend.disconnect_from_network(vm, nic)
359

    
360

    
361
def console(vm, console_type):
362
    """Arrange for an OOB console of the specified type
363

364
    This method arranges for an OOB console of the specified type.
365
    Only consoles of type "vnc" are supported for now.
366

367
    It uses a running instance of vncauthproxy to setup proper
368
    VNC forwarding with a random password, then returns the necessary
369
    VNC connection info to the caller.
370

371
    """
372
    log.info("Get console  VM %s, type %s", vm, console_type)
373

    
374
    # Use RAPI to get VNC console information for this instance
375
    if vm.operstate != "STARTED":
376
        raise faults.BadRequest('Server not in ACTIVE state.')
377

    
378
    if settings.TEST:
379
        console_data = {'kind': 'vnc', 'host': 'ganeti_node', 'port': 1000}
380
    else:
381
        console_data = backend.get_instance_console(vm)
382

    
383
    if console_data['kind'] != 'vnc':
384
        message = 'got console of kind %s, not "vnc"' % console_data['kind']
385
        raise faults.ServiceUnavailable(message)
386

    
387
    # Let vncauthproxy decide on the source port.
388
    # The alternative: static allocation, e.g.
389
    # sport = console_data['port'] - 1000
390
    sport = 0
391
    daddr = console_data['host']
392
    dport = console_data['port']
393
    password = util.random_password()
394

    
395
    if settings.TEST:
396
        fwd = {'source_port': 1234, 'status': 'OK'}
397
    else:
398
        fwd = request_vnc_forwarding(sport, daddr, dport, password)
399

    
400
    if fwd['status'] != "OK":
401
        raise faults.ServiceUnavailable('vncauthproxy returned error status')
402

    
403
    # Verify that the VNC server settings haven't changed
404
    if not settings.TEST:
405
        if console_data != backend.get_instance_console(vm):
406
            raise faults.ServiceUnavailable('VNC Server settings changed.')
407

    
408
    console = {
409
        'type': 'vnc',
410
        'host': getfqdn(),
411
        'port': fwd['source_port'],
412
        'password': password}
413

    
414
    return console
415

    
416

    
417
def rename(server, new_name):
418
    """Rename a VirtualMachine."""
419
    old_name = server.name
420
    server.name = new_name
421
    server.save()
422
    log.info("Renamed server '%s' from '%s' to '%s'", server, old_name,
423
             new_name)
424
    return server
425

    
426

    
427
@transaction.commit_on_success
428
def create_port(*args, **kwargs):
429
    return _create_port(*args, **kwargs)
430

    
431

    
432
def _create_port(userid, network, machine=None, use_ipaddress=None,
433
                 address=None, name="", security_groups=None,
434
                 device_owner=None):
435
    """Create a new port on the specified network.
436

437
    Create a new Port(NetworkInterface model) on the specified Network. If
438
    'machine' is specified, the machine will be connected to the network using
439
    this port. If 'use_ipaddress' argument is specified, the port will be
440
    assigned this IPAddress. Otherwise, an IPv4 address from the IPv4 subnet
441
    will be allocated.
442

443
    """
444
    if network.state != "ACTIVE":
445
        raise faults.BuildInProgress("Can not create port while network is in"
446
                                     " state %s" % network.state)
447
    ipaddress = None
448
    if use_ipaddress is not None:
449
        # Use an existing IPAddress object.
450
        ipaddress = use_ipaddress
451
        if ipaddress and (ipaddress.network_id != network.id):
452
            msg = "IP Address %s does not belong to network %s"
453
            raise faults.Conflict(msg % (ipaddress.address, network.id))
454
    else:
455
        # If network has IPv4 subnets, try to allocate the address that the
456
        # the user specified or a random one.
457
        if network.subnets.filter(ipversion=4).exists():
458
            try:
459
                ipaddress = util.allocate_ip(network, userid=userid,
460
                                             address=address)
461
            except pools.ValueNotAvailable:
462
                msg = "Address %s is already in use." % address
463
                raise faults.Conflict(msg)
464
        elif address is not None:
465
            raise faults.BadRequest("Address %s is not a valid IP for the"
466
                                    " defined network subnets" % address)
467

    
468
    if ipaddress is not None and ipaddress.nic is not None:
469
        raise faults.Conflict("IP address '%s' is already in use" %
470
                              ipaddress.address)
471

    
472
    port = NetworkInterface.objects.create(network=network,
473
                                           state="DOWN",
474
                                           userid=userid,
475
                                           device_owner=None,
476
                                           name=name)
477

    
478
    # add the security groups if any
479
    if security_groups:
480
        port.security_groups.add(*security_groups)
481

    
482
    if ipaddress is not None:
483
        # Associate IPAddress with the Port
484
        ipaddress.nic = port
485
        ipaddress.save()
486

    
487
    if machine is not None:
488
        # Connect port to the instance.
489
        machine = connect(machine, network, port)
490
        jobID = machine.task_job_id
491
        log.info("Created Port %s with IP %s. Ganeti Job: %s",
492
                 port, ipaddress, jobID)
493
    else:
494
        log.info("Created Port %s with IP %s not attached to any instance",
495
                 port, ipaddress)
496

    
497
    return port
498

    
499

    
500
def associate_port_with_machine(port, machine):
501
    """Associate a Port with a VirtualMachine.
502

503
    Associate the port with the VirtualMachine and add an entry to the
504
    IPAddressLog if the port has a public IPv4 address from a public network.
505

506
    """
507
    if port.machine is not None:
508
        raise faults.Conflict("Port %s is already in use." % port.id)
509
    if port.network.public:
510
        ipv4_address = port.ipv4_address
511
        if ipv4_address is not None:
512
            ip_log = IPAddressLog.objects.create(server_id=machine.id,
513
                                                 network_id=port.network_id,
514
                                                 address=ipv4_address,
515
                                                 active=True)
516
            log.debug("Created IP log entry %s", ip_log)
517
    port.machine = machine
518
    port.state = "BUILD"
519
    port.device_owner = "vm"
520
    port.save()
521
    return port
522

    
523

    
524
@transaction.commit_on_success
525
def delete_port(port):
526
    """Delete a port by removing the NIC card from the instance.
527

528
    Send a Job to remove the NIC card from the instance. The port
529
    will be deleted and the associated IPv4 addressess will be released
530
    when the job completes successfully.
531

532
    """
533

    
534
    if port.machine is not None:
535
        vm = disconnect(port.machine, port)
536
        log.info("Removing port %s, Job: %s", port, vm.task_job_id)
537
    else:
538
        backend.remove_nic_ips(port)
539
        port.delete()
540
        log.info("Removed port %s", port)
541

    
542
    return port