diff options
| author | Todd Willey <todd@ansolabs.com> | 2011-04-04 21:12:57 -0400 |
|---|---|---|
| committer | Todd Willey <todd@ansolabs.com> | 2011-04-04 21:12:57 -0400 |
| commit | da346dac037a46582d569180915010f4c4e2cd50 (patch) | |
| tree | 94cbeab7422f22b443868cb21d02406deb35826b /nova/virt | |
| parent | 2ee9070c3824d296bada49fc6637c09f8e18a5eb (diff) | |
| parent | 08417c48c223ad1b698ab1d00686a967b6a2dc0a (diff) | |
Merge trunk.
Diffstat (limited to 'nova/virt')
29 files changed, 6362 insertions, 814 deletions
diff --git a/nova/virt/connection.py b/nova/virt/connection.py index f5a978997..875d558a0 100644 --- a/nova/virt/connection.py +++ b/nova/virt/connection.py @@ -23,10 +23,13 @@ import sys from nova import flags from nova import log as logging +from nova import utils +from nova.virt import driver from nova.virt import fake +from nova.virt import hyperv from nova.virt import libvirt_conn +from nova.virt import vmwareapi_conn from nova.virt import xenapi_conn -from nova.virt import hyperv LOG = logging.getLogger("nova.virt.connection") @@ -67,10 +70,12 @@ def get_connection(read_only=False): conn = xenapi_conn.get_connection(read_only) elif t == 'hyperv': conn = hyperv.get_connection(read_only) + elif t == 'vmwareapi': + conn = vmwareapi_conn.get_connection(read_only) else: raise Exception('Unknown connection type "%s"' % t) if conn is None: LOG.error(_('Failed to open connection to the hypervisor')) sys.exit(1) - return conn + return utils.check_isinstance(conn, driver.ComputeDriver) diff --git a/nova/virt/cpuinfo.xml.template b/nova/virt/cpuinfo.xml.template new file mode 100644 index 000000000..48842b29d --- /dev/null +++ b/nova/virt/cpuinfo.xml.template @@ -0,0 +1,9 @@ +<cpu> + <arch>$arch</arch> + <model>$model</model> + <vendor>$vendor</vendor> + <topology sockets="$topology.sockets" cores="$topology.cores" threads="$topology.threads"/> +#for $var in $features + <features name="$var" /> +#end for +</cpu> diff --git a/nova/virt/disk.py b/nova/virt/disk.py index c5565abfa..ddea1a1f7 100644 --- a/nova/virt/disk.py +++ b/nova/virt/disk.py @@ -26,6 +26,8 @@ import os import tempfile import time +from nova import context +from nova import db from nova import exception from nova import flags from nova import log as logging @@ -38,6 +40,13 @@ flags.DEFINE_integer('minimum_root_size', 1024 * 1024 * 1024 * 10, 'minimum size in bytes of root partition') flags.DEFINE_integer('block_size', 1024 * 1024 * 256, 'block_size to use for dd') +flags.DEFINE_string('injected_network_template', + utils.abspath('virt/interfaces.template'), + 'Template file for injected network') +flags.DEFINE_integer('timeout_nbd', 10, + 'time to wait for a NBD device coming up') +flags.DEFINE_integer('max_nbd_devices', 16, + 'maximum number of possible nbd devices') def extend(image, size): @@ -45,10 +54,10 @@ def extend(image, size): file_size = os.path.getsize(image) if file_size >= size: return - utils.execute('truncate -s %s %s' % (size, image)) + utils.execute('truncate', '-s', size, image) # NOTE(vish): attempts to resize filesystem - utils.execute('e2fsck -fp %s' % image, check_exit_code=False) - utils.execute('resize2fs %s' % image, check_exit_code=False) + utils.execute('e2fsck', '-fp', image, check_exit_code=False) + utils.execute('resize2fs', image, check_exit_code=False) def inject_data(image, key=None, net=None, partition=None, nbd=False): @@ -64,7 +73,7 @@ def inject_data(image, key=None, net=None, partition=None, nbd=False): try: if not partition is None: # create partition - out, err = utils.execute('sudo kpartx -a %s' % device) + out, err = utils.execute('sudo', 'kpartx', '-a', device) if err: raise exception.Error(_('Failed to load partition: %s') % err) mapped_device = '/dev/mapper/%sp%s' % (device.split('/')[-1], @@ -80,50 +89,82 @@ def inject_data(image, key=None, net=None, partition=None, nbd=False): mapped_device) # Configure ext2fs so that it doesn't auto-check every N boots - out, err = utils.execute('sudo tune2fs -c 0 -i 0 %s' % mapped_device) + out, err = utils.execute('sudo', 'tune2fs', + '-c', 0, '-i', 0, mapped_device) tmpdir = tempfile.mkdtemp() try: # mount loopback to dir out, err = utils.execute( - 'sudo mount %s %s' % (mapped_device, tmpdir)) + 'sudo', 'mount', mapped_device, tmpdir) if err: raise exception.Error(_('Failed to mount filesystem: %s') % err) try: - if key: - # inject key file - _inject_key_into_fs(key, tmpdir) - if net: - _inject_net_into_fs(net, tmpdir) + inject_data_into_fs(tmpdir, key, net, utils.execute) finally: # unmount device - utils.execute('sudo umount %s' % mapped_device) + utils.execute('sudo', 'umount', mapped_device) finally: # remove temporary directory - utils.execute('rmdir %s' % tmpdir) + utils.execute('rmdir', tmpdir) if not partition is None: # remove partitions - utils.execute('sudo kpartx -d %s' % device) + utils.execute('sudo', 'kpartx', '-d', device) finally: _unlink_device(device, nbd) +def setup_container(image, container_dir=None, nbd=False): + """Setup the LXC container. + + It will mount the loopback image to the container directory in order + to create the root filesystem for the container. + + LXC does not support qcow2 images yet. + """ + try: + device = _link_device(image, nbd) + utils.execute('sudo', 'mount', device, container_dir) + except Exception, exn: + LOG.exception(_('Failed to mount filesystem: %s'), exn) + _unlink_device(device, nbd) + + +def destroy_container(target, instance, nbd=False): + """Destroy the container once it terminates. + + It will umount the container that is mounted, try to find the loopback + device associated with the container and delete it. + + LXC does not support qcow2 images yet. + """ + try: + container_dir = '%s/rootfs' % target + utils.execute('sudo', 'umount', container_dir) + finally: + out, err = utils.execute('sudo', 'losetup', '-a') + for loop in out.splitlines(): + if instance['name'] in loop: + device = loop.split(loop, ':') + _unlink_device(device, nbd) + + def _link_device(image, nbd): """Link image to device using loopback or nbd""" if nbd: device = _allocate_device() - utils.execute('sudo qemu-nbd -c %s %s' % (device, image)) + utils.execute('sudo', 'qemu-nbd', '-c', device, image) # NOTE(vish): this forks into another process, so give it a chance # to set up before continuuing - for i in xrange(10): + for i in xrange(FLAGS.timeout_nbd): if os.path.exists("/sys/block/%s/pid" % os.path.basename(device)): return device time.sleep(1) raise exception.Error(_('nbd device %s did not show up') % device) else: - out, err = utils.execute('sudo losetup --find --show %s' % image) + out, err = utils.execute('sudo', 'losetup', '--find', '--show', image) if err: raise exception.Error(_('Could not attach image to loopback: %s') % err) @@ -133,13 +174,13 @@ def _link_device(image, nbd): def _unlink_device(device, nbd): """Unlink image from device using loopback or nbd""" if nbd: - utils.execute('sudo qemu-nbd -d %s' % device) + utils.execute('sudo', 'qemu-nbd', '-d', device) _free_device(device) else: - utils.execute('sudo losetup --detach %s' % device) + utils.execute('sudo', 'losetup', '--detach', device) -_DEVICES = ['/dev/nbd%s' % i for i in xrange(16)] +_DEVICES = ['/dev/nbd%s' % i for i in xrange(FLAGS.max_nbd_devices)] def _allocate_device(): @@ -159,28 +200,40 @@ def _free_device(device): _DEVICES.append(device) -def _inject_key_into_fs(key, fs): +def inject_data_into_fs(fs, key, net, execute): + """Injects data into a filesystem already mounted by the caller. + Virt connections can call this directly if they mount their fs + in a different way to inject_data + """ + if key: + _inject_key_into_fs(key, fs, execute=execute) + if net: + _inject_net_into_fs(net, fs, execute=execute) + + +def _inject_key_into_fs(key, fs, execute=None): """Add the given public ssh key to root's authorized_keys. key is an ssh key string. fs is the path to the base of the filesystem into which to inject the key. """ sshdir = os.path.join(fs, 'root', '.ssh') - utils.execute('sudo mkdir -p %s' % sshdir) # existing dir doesn't matter - utils.execute('sudo chown root %s' % sshdir) - utils.execute('sudo chmod 700 %s' % sshdir) + utils.execute('sudo', 'mkdir', '-p', sshdir) # existing dir doesn't matter + utils.execute('sudo', 'chown', 'root', sshdir) + utils.execute('sudo', 'chmod', '700', sshdir) keyfile = os.path.join(sshdir, 'authorized_keys') - utils.execute('sudo tee -a %s' % keyfile, '\n' + key.strip() + '\n') + utils.execute('sudo', 'tee', '-a', keyfile, + process_input='\n' + key.strip() + '\n') -def _inject_net_into_fs(net, fs): +def _inject_net_into_fs(net, fs, execute=None): """Inject /etc/network/interfaces into the filesystem rooted at fs. net is the contents of /etc/network/interfaces. """ netdir = os.path.join(os.path.join(fs, 'etc'), 'network') - utils.execute('sudo mkdir -p %s' % netdir) # existing dir doesn't matter - utils.execute('sudo chown root:root %s' % netdir) - utils.execute('sudo chmod 755 %s' % netdir) + utils.execute('sudo', 'mkdir', '-p', netdir) # existing dir doesn't matter + utils.execute('sudo', 'chown', 'root:root', netdir) + utils.execute('sudo', 'chmod', 755, netdir) netfile = os.path.join(netdir, 'interfaces') - utils.execute('sudo tee %s' % netfile, net) + utils.execute('sudo', 'tee', netfile, process_input=net) diff --git a/nova/virt/driver.py b/nova/virt/driver.py new file mode 100644 index 000000000..eb9626d08 --- /dev/null +++ b/nova/virt/driver.py @@ -0,0 +1,243 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 Justin Santa Barbara +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +""" +Driver base-classes: + + (Beginning of) the contract that compute drivers must follow, and shared + types that support that contract +""" + +from nova.compute import power_state + + +class InstanceInfo(object): + def __init__(self, name, state): + self.name = name + assert state in power_state.valid_states(), "Bad state: %s" % state + self.state = state + + +class ComputeDriver(object): + """Base class for compute drivers. + + Lots of documentation is currently on fake.py. + """ + + def init_host(self, host): + """Adopt existing VM's running here""" + raise NotImplementedError() + + def get_info(self, instance_name): + """Get the current status of an instance, by name (not ID!) + + Returns a dict containing: + :state: the running state, one of the power_state codes + :max_mem: (int) the maximum memory in KBytes allowed + :mem: (int) the memory in KBytes used by the domain + :num_cpu: (int) the number of virtual CPUs for the domain + :cpu_time: (int) the CPU time used in nanoseconds + """ + raise NotImplementedError() + + def list_instances(self): + raise NotImplementedError() + + def list_instances_detail(self): + """Return a list of InstanceInfo for all registered VMs""" + raise NotImplementedError() + + def spawn(self, instance, network_info=None): + """Launch a VM for the specified instance""" + raise NotImplementedError() + + def destroy(self, instance, cleanup=True): + """Destroy (shutdown and delete) the specified instance. + + The given parameter is an instance of nova.compute.service.Instance, + and so the instance is being specified as instance.name. + + The work will be done asynchronously. This function returns a + task that allows the caller to detect when it is complete. + + If the instance is not found (for example if networking failed), this + function should still succeed. It's probably a good idea to log a + warning in that case. + + """ + raise NotImplementedError() + + def reboot(self, instance): + """Reboot specified VM""" + raise NotImplementedError() + + def snapshot_instance(self, context, instance_id, image_id): + raise NotImplementedError() + + def get_console_pool_info(self, console_type): + raise NotImplementedError() + + def get_console_output(self, instance): + raise NotImplementedError() + + def get_ajax_console(self, instance): + raise NotImplementedError() + + def get_diagnostics(self, instance): + """Return data about VM diagnostics""" + raise NotImplementedError() + + def get_host_ip_addr(self): + raise NotImplementedError() + + def attach_volume(self, context, instance_id, volume_id, mountpoint): + raise NotImplementedError() + + def detach_volume(self, context, instance_id, volume_id): + raise NotImplementedError() + + def compare_cpu(self, context, cpu_info): + raise NotImplementedError() + + def migrate_disk_and_power_off(self, instance, dest): + """Transfers the VHD of a running instance to another host, then shuts + off the instance copies over the COW disk""" + raise NotImplementedError() + + def snapshot(self, instance, image_id): + """Create snapshot from a running VM instance.""" + raise NotImplementedError() + + def finish_resize(self, instance, disk_info): + """Completes a resize, turning on the migrated instance""" + raise NotImplementedError() + + def revert_resize(self, instance): + """Reverts a resize, powering back on the instance""" + raise NotImplementedError() + + def pause(self, instance, callback): + """Pause VM instance""" + raise NotImplementedError() + + def unpause(self, instance, callback): + """Unpause paused VM instance""" + raise NotImplementedError() + + def suspend(self, instance, callback): + """suspend the specified instance""" + raise NotImplementedError() + + def resume(self, instance, callback): + """resume the specified instance""" + raise NotImplementedError() + + def rescue(self, instance, callback): + """Rescue the specified instance""" + raise NotImplementedError() + + def unrescue(self, instance, callback): + """Unrescue the specified instance""" + raise NotImplementedError() + + def update_available_resource(self, ctxt, host): + """Updates compute manager resource info on ComputeNode table. + + This method is called when nova-compute launches, and + whenever admin executes "nova-manage service update_resource". + + :param ctxt: security context + :param host: hostname that compute manager is currently running + + """ + raise NotImplementedError() + + def live_migration(self, ctxt, instance_ref, dest, + post_method, recover_method): + """Spawning live_migration operation for distributing high-load. + + :params ctxt: security context + :params instance_ref: + nova.db.sqlalchemy.models.Instance object + instance object that is migrated. + :params dest: destination host + :params post_method: + post operation method. + expected nova.compute.manager.post_live_migration. + :params recover_method: + recovery method when any exception occurs. + expected nova.compute.manager.recover_live_migration. + + """ + raise NotImplementedError() + + def refresh_security_group_rules(self, security_group_id): + raise NotImplementedError() + + def refresh_security_group_members(self, security_group_id): + raise NotImplementedError() + + def reset_network(self, instance): + """reset networking for specified instance""" + raise NotImplementedError() + + def ensure_filtering_rules_for_instance(self, instance_ref): + """Setting up filtering rules and waiting for its completion. + + To migrate an instance, filtering rules to hypervisors + and firewalls are inevitable on destination host. + ( Waiting only for filtering rules to hypervisor, + since filtering rules to firewall rules can be set faster). + + Concretely, the below method must be called. + - setup_basic_filtering (for nova-basic, etc.) + - prepare_instance_filter(for nova-instance-instance-xxx, etc.) + + to_xml may have to be called since it defines PROJNET, PROJMASK. + but libvirt migrates those value through migrateToURI(), + so , no need to be called. + + Don't use thread for this method since migration should + not be started when setting-up filtering rules operations + are not completed. + + :params instance_ref: nova.db.sqlalchemy.models.Instance object + + """ + raise NotImplementedError() + + def unfilter_instance(self, instance): + """Stop filtering instance""" + raise NotImplementedError() + + def set_admin_password(self, context, instance_id, new_pass=None): + """Set the root/admin password for an instance on this server.""" + raise NotImplementedError() + + def inject_file(self, instance, b64_path, b64_contents): + """Create a file on the VM instance. The file path and contents + should be base64-encoded. + """ + raise NotImplementedError() + + def inject_network_info(self, instance): + """inject network info for specified instance""" + raise NotImplementedError() + + def poll_rescued_instances(self, timeout): + """Poll for rescued instances""" + raise NotImplementedError() diff --git a/nova/virt/fake.py b/nova/virt/fake.py index b16d53634..3f030833e 100644 --- a/nova/virt/fake.py +++ b/nova/virt/fake.py @@ -26,7 +26,13 @@ semantics of real hypervisor connections. """ from nova import exception +from nova import log as logging +from nova import utils from nova.compute import power_state +from nova.virt import driver + + +LOG = logging.getLogger('nova.compute.disk') def get_connection(_): @@ -34,7 +40,14 @@ def get_connection(_): return FakeConnection.instance() -class FakeConnection(object): +class FakeInstance(object): + + def __init__(self, name, state): + self.name = name + self.state = state + + +class FakeConnection(driver.ComputeDriver): """ The interface to this class talks in terms of 'instances' (Amazon EC2 and internal Nova terminology), by which we mean 'running virtual machine' @@ -90,6 +103,17 @@ class FakeConnection(object): """ return self.instances.keys() + def _map_to_instance_info(self, instance): + instance = utils.check_isinstance(instance, FakeInstance) + info = driver.InstanceInfo(instance.name, instance.state) + return info + + def list_instances_detail(self): + info_list = [] + for instance in self.instances.values(): + info_list.append(self._map_to_instance_info(instance)) + return info_list + def spawn(self, instance): """ Create a new instance/VM/domain on the virtualization platform. @@ -109,9 +133,10 @@ class FakeConnection(object): that it was before this call began. """ - fake_instance = FakeInstance() - self.instances[instance.name] = fake_instance - fake_instance._state = power_state.RUNNING + name = instance.name + state = power_state.RUNNING + fake_instance = FakeInstance(name, state) + self.instances[name] = fake_instance def snapshot(self, instance, name): """ @@ -139,6 +164,24 @@ class FakeConnection(object): """ pass + def get_host_ip_addr(self): + """ + Retrieves the IP address of the dom0 + """ + pass + + def resize(self, instance, flavor): + """ + Resizes/Migrates the specified instance. + + The flavor parameter determines whether or not the instance RAM and + disk space are modified, and if so, to what size. + + The work will be done asynchronously. This function returns a task + that allows the caller to detect when it is complete. + """ + pass + def set_admin_password(self, instance, new_pass): """ Set the root password on the specified instance. @@ -152,6 +195,21 @@ class FakeConnection(object): """ pass + def inject_file(self, instance, b64_path, b64_contents): + """ + Writes a file on the specified instance. + + The first parameter is an instance of nova.compute.service.Instance, + and so the instance is being specified as instance.name. The second + parameter is the base64-encoded path to which the file is to be + written on the instance; the third is the contents of the file, also + base64-encoded. + + The work will be done asynchronously. This function returns a + task that allows the caller to detect when it is complete. + """ + pass + def rescue(self, instance): """ Rescue the specified instance. @@ -164,6 +222,19 @@ class FakeConnection(object): """ pass + def migrate_disk_and_power_off(self, instance, dest): + """ + Transfers the disk of a running instance in multiple phases, turning + off the instance before the end. + """ + pass + + def attach_disk(self, instance, disk_info): + """ + Attaches the disk to an instance given the metadata disk_info + """ + pass + def pause(self, instance, callback): """ Pause the specified instance. @@ -189,16 +260,12 @@ class FakeConnection(object): pass def destroy(self, instance): - """ - Destroy (shutdown and delete) the specified instance. - - The given parameter is an instance of nova.compute.service.Instance, - and so the instance is being specified as instance.name. - - The work will be done asynchronously. This function returns a - task that allows the caller to detect when it is complete. - """ - del self.instances[instance.name] + key = instance.name + if key in self.instances: + del self.instances[key] + else: + LOG.warning("Key '%s' not in instances '%s'" % + (key, self.instances)) def attach_volume(self, instance_name, device_path, mountpoint): """Attach the disk at device_path to the instance at mountpoint""" @@ -224,7 +291,7 @@ class FakeConnection(object): raise exception.NotFound(_("Instance %s Not Found") % instance_name) i = self.instances[instance_name] - return {'state': i._state, + return {'state': i.state, 'max_mem': 0, 'mem': 0, 'num_cpu': 2, @@ -277,7 +344,7 @@ class FakeConnection(object): Note that this function takes an instance ID, not a compute.service.Instance, so that it can be called by compute.monitor. """ - return [0L, 0L, 0L, 0L, null] + return [0L, 0L, 0L, 0L, None] def interface_stats(self, instance_name, iface_id): """ @@ -304,7 +371,14 @@ class FakeConnection(object): return 'FAKE CONSOLE OUTPUT' def get_ajax_console(self, instance): - return 'http://fakeajaxconsole.com/?token=FAKETOKEN' + return {'token': 'FAKETOKEN', + 'host': 'fakeajaxconsole.com', + 'port': 6969} + + def get_vnc_console(self, instance): + return {'token': 'FAKETOKEN', + 'host': 'fakevncconsole.com', + 'port': 6969} def get_console_pool_info(self, console_type): return {'address': '127.0.0.1', @@ -375,8 +449,27 @@ class FakeConnection(object): """ pass + def update_available_resource(self, ctxt, host): + """This method is supported only by libvirt.""" + return -class FakeInstance(object): + def compare_cpu(self, xml): + """This method is supported only by libvirt.""" + raise NotImplementedError('This method is supported only by libvirt.') - def __init__(self): - self._state = power_state.NOSTATE + def ensure_filtering_rules_for_instance(self, instance_ref): + """This method is supported only by libvirt.""" + raise NotImplementedError('This method is supported only by libvirt.') + + def live_migration(self, context, instance_ref, dest, + post_method, recover_method): + """This method is supported only by libvirt.""" + return + + def unfilter_instance(self, instance_ref): + """This method is supported only by libvirt.""" + raise NotImplementedError('This method is supported only by libvirt.') + + def test_remove_vm(self, instance_name): + """ Removes the named VM, as if it crashed. For testing""" + self.instances.pop(instance_name) diff --git a/nova/virt/hyperv.py b/nova/virt/hyperv.py index 29d18dac5..13f403a66 100644 --- a/nova/virt/hyperv.py +++ b/nova/virt/hyperv.py @@ -68,6 +68,7 @@ from nova import flags from nova import log as logging from nova.auth import manager from nova.compute import power_state +from nova.virt import driver from nova.virt import images wmi = None @@ -108,8 +109,9 @@ def get_connection(_): return HyperVConnection() -class HyperVConnection(object): +class HyperVConnection(driver.ComputeDriver): def __init__(self): + super(HyperVConnection, self).__init__() self._conn = wmi.WMI(moniker='//./root/virtualization') self._cim_conn = wmi.WMI(moniker='//./root/cimv2') @@ -124,6 +126,19 @@ class HyperVConnection(object): for v in self._conn.Msvm_ComputerSystem(['ElementName'])] return vms + def list_instances_detail(self): + # TODO(justinsb): This is a terrible implementation (1+N) + instance_infos = [] + for instance_name in self.list_instances(): + info = self.get_info(instance_name) + + state = info['state'] + + instance_info = driver.InstanceInfo(instance_name, state) + instance_infos.append(instance_info) + + return instance_infos + def spawn(self, instance): """ Create a new VM and start it.""" vm = self._lookup(instance.name) @@ -345,7 +360,7 @@ class HyperVConnection(object): newinst = cl.new() #Copy the properties from the original. for prop in wmi_obj._properties: - newinst.Properties_.Item(prop).Value =\ + newinst.Properties_.Item(prop).Value = \ wmi_obj.Properties_.Item(prop).Value return newinst @@ -467,3 +482,10 @@ class HyperVConnection(object): if vm is None: raise exception.NotFound('Cannot detach volume from missing %s ' % instance_name) + + def poll_rescued_instances(self, timeout): + pass + + def update_available_resource(self, ctxt, host): + """This method is supported only by libvirt.""" + return diff --git a/nova/virt/images.py b/nova/virt/images.py index 7a6fef330..2e3f2ee4d 100644 --- a/nova/virt/images.py +++ b/nova/virt/images.py @@ -28,29 +28,32 @@ import time import urllib2 import urlparse +from nova import context from nova import flags from nova import log as logging from nova import utils from nova.auth import manager from nova.auth import signer -from nova.objectstore import image FLAGS = flags.FLAGS -flags.DEFINE_bool('use_s3', True, - 'whether to get images from s3 or use local copy') - LOG = logging.getLogger('nova.virt.images') -def fetch(image, path, user, project): - if FLAGS.use_s3: - f = _fetch_s3_image - else: - f = _fetch_local_image - return f(image, path, user, project) +def fetch(image_id, path, _user, _project): + # TODO(vish): Improve context handling and add owner and auth data + # when it is added to glance. Right now there is no + # auth checking in glance, so we assume that access was + # checked before we got here. + image_service = utils.import_object(FLAGS.image_service) + with open(path, "wb") as image_file: + elevated = context.get_admin_context() + metadata = image_service.get(elevated, image_id, image_file) + return metadata +# NOTE(vish): The methods below should be unnecessary, but I'm leaving +# them in case the glance client does not work on windows. def _fetch_image_no_curl(url, path, headers): request = urllib2.Request(url) for (k, v) in headers.iteritems(): @@ -94,8 +97,7 @@ def _fetch_s3_image(image, path, user, project): cmd += ['-H', '\'%s: %s\'' % (k, v)] cmd += ['-o', path] - cmd_out = ' '.join(cmd) - return utils.execute(cmd_out) + return utils.execute(*cmd) def _fetch_local_image(image, path, user, project): @@ -103,13 +105,15 @@ def _fetch_local_image(image, path, user, project): if sys.platform.startswith('win'): return shutil.copy(source, path) else: - return utils.execute('cp %s %s' % (source, path)) + return utils.execute('cp', source, path) def _image_path(path): return os.path.join(FLAGS.images_path, path) +# TODO(vish): xenapi should use the glance client code directly instead +# of retrieving the image using this method. def image_url(image): if FLAGS.image_service == "nova.image.glance.GlanceImageService": return "http://%s:%s/images/%s" % (FLAGS.glance_host, diff --git a/nova/virt/interfaces.template b/nova/virt/interfaces.template index 87b92b84a..e527cf35c 100644 --- a/nova/virt/interfaces.template +++ b/nova/virt/interfaces.template @@ -5,13 +5,20 @@ auto lo iface lo inet loopback -# The primary network interface -auto eth0 -iface eth0 inet static - address %(address)s - netmask %(netmask)s - broadcast %(broadcast)s - gateway %(gateway)s - dns-nameservers %(dns)s +#for $ifc in $interfaces +auto ${ifc.name} +iface ${ifc.name} inet static + address ${ifc.address} + netmask ${ifc.netmask} + broadcast ${ifc.broadcast} + gateway ${ifc.gateway} + dns-nameservers ${ifc.dns} +#if $use_ipv6 +iface ${ifc.name} inet6 static + address ${ifc.address_v6} + netmask ${ifc.netmask_v6} + gateway ${ifc.gateway_v6} +#end if +#end for diff --git a/nova/virt/libvirt.xml.template b/nova/virt/libvirt.xml.template index 88bfbc668..de2497a76 100644 --- a/nova/virt/libvirt.xml.template +++ b/nova/virt/libvirt.xml.template @@ -2,7 +2,12 @@ <name>${name}</name> <memory>${memory_kb}</memory> <os> -#if $type == 'uml' +#if $type == 'lxc' + #set $disk_prefix = '' + #set $disk_bus = '' + <type>exe</type> + <init>/sbin/init</init> +#else if $type == 'uml' #set $disk_prefix = 'ubd' #set $disk_bus = 'uml' <type>uml</type> @@ -44,7 +49,13 @@ </features> <vcpu>${vcpus}</vcpu> <devices> -#if $getVar('rescue', False) +#if $type == 'lxc' + <filesystem type='mount'> + <source dir='${basepath}/rootfs'/> + <target dir='/'/> + </filesystem> +#else + #if $getVar('rescue', False) <disk type='file'> <driver type='${driver_type}'/> <source file='${basepath}/disk.rescue'/> @@ -55,36 +66,39 @@ <source file='${basepath}/disk'/> <target dev='${disk_prefix}b' bus='${disk_bus}'/> </disk> -#else + #else <disk type='file'> <driver type='${driver_type}'/> <source file='${basepath}/disk'/> <target dev='${disk_prefix}a' bus='${disk_bus}'/> </disk> - #if $getVar('local', False) - <disk type='file'> - <driver type='${driver_type}'/> - <source file='${basepath}/disk.local'/> - <target dev='${disk_prefix}b' bus='${disk_bus}'/> - </disk> + #if $getVar('local', False) + <disk type='file'> + <driver type='${driver_type}'/> + <source file='${basepath}/disk.local'/> + <target dev='${disk_prefix}b' bus='${disk_bus}'/> + </disk> + #end if #end if #end if + +#for $nic in $nics <interface type='bridge'> - <source bridge='${bridge_name}'/> - <mac address='${mac_address}'/> + <source bridge='${nic.bridge_name}'/> + <mac address='${nic.mac_address}'/> <!-- <model type='virtio'/> CANT RUN virtio network right now --> - <filterref filter="nova-instance-${name}"> - <parameter name="IP" value="${ip_address}" /> - <parameter name="DHCPSERVER" value="${dhcp_server}" /> -#if $getVar('extra_params', False) - ${extra_params} + <filterref filter="nova-instance-${name}-${nic.id}"> + <parameter name="IP" value="${nic.ip_address}" /> + <parameter name="DHCPSERVER" value="${nic.dhcp_server}" /> +#if $getVar('nic.extra_params', False) + ${nic.extra_params} #end if -#if $getVar('ra_server', False) - <parameter name="RASERVER" value="${ra_server}" /> +#if $getVar('nic.gateway_v6', False) + <parameter name="RASERVER" value="${nic.gateway_v6}" /> #end if </filterref> </interface> - +#end for <!-- The order is significant here. File must be defined first --> <serial type="file"> <source path='${basepath}/console.log'/> @@ -101,5 +115,8 @@ <target port='0'/> </serial> +#if $getVar('vncserver_host', False) + <graphics type='vnc' port='-1' autoport='yes' keymap='en-us' listen='${vncserver_host}'/> +#end if </devices> </domain> diff --git a/nova/virt/libvirt_conn.py b/nova/virt/libvirt_conn.py index ec6572d3f..f3d73103b 100644 --- a/nova/virt/libvirt_conn.py +++ b/nova/virt/libvirt_conn.py @@ -20,7 +20,7 @@ """ A connection to a hypervisor through libvirt. -Supports KVM, QEMU, UML, and XEN. +Supports KVM, LXC, QEMU, UML, and XEN. **Related Flags** @@ -36,16 +36,19 @@ Supports KVM, QEMU, UML, and XEN. """ +import multiprocessing import os -import shutil import random +import shutil import subprocess +import sys +import tempfile +import time import uuid from xml.dom import minidom - +from xml.etree import ElementTree from eventlet import greenthread -from eventlet import event from eventlet import tpool import IPy @@ -55,12 +58,14 @@ from nova import db from nova import exception from nova import flags from nova import log as logging +#from nova import test from nova import utils -#from nova.api import context +from nova import vnc from nova.auth import manager from nova.compute import instance_types from nova.compute import power_state from nova.virt import disk +from nova.virt import driver from nova.virt import images libvirt = None @@ -70,20 +75,19 @@ Template = None LOG = logging.getLogger('nova.virt.libvirt_conn') FLAGS = flags.FLAGS +flags.DECLARE('live_migration_retry_count', 'nova.compute.manager') # TODO(vish): These flags should probably go into a shared location flags.DEFINE_string('rescue_image_id', 'ami-rescue', 'Rescue ami image') flags.DEFINE_string('rescue_kernel_id', 'aki-rescue', 'Rescue aki image') flags.DEFINE_string('rescue_ramdisk_id', 'ari-rescue', 'Rescue ari image') -flags.DEFINE_string('injected_network_template', - utils.abspath('virt/interfaces.template'), - 'Template file for injected network') + flags.DEFINE_string('libvirt_xml_template', utils.abspath('virt/libvirt.xml.template'), 'Libvirt XML Template') flags.DEFINE_string('libvirt_type', 'kvm', 'Libvirt domain type (valid options are: ' - 'kvm, qemu, uml, xen)') + 'kvm, lxc, qemu, uml, xen)') flags.DEFINE_string('libvirt_uri', '', 'Override the default libvirt URI (which is dependent' @@ -100,6 +104,19 @@ flags.DEFINE_string('ajaxterm_portrange', flags.DEFINE_string('firewall_driver', 'nova.virt.libvirt_conn.IptablesFirewallDriver', 'Firewall driver (defaults to iptables)') +flags.DEFINE_string('cpuinfo_xml_template', + utils.abspath('virt/cpuinfo.xml.template'), + 'CpuInfo XML Template (Used only live migration now)') +flags.DEFINE_string('live_migration_uri', + "qemu+tcp://%s/system", + 'Define protocol used by live_migration feature') +flags.DEFINE_string('live_migration_flag', + "VIR_MIGRATE_UNDEFINE_SOURCE, VIR_MIGRATE_PEER2PEER", + 'Define live migration behavior.') +flags.DEFINE_integer('live_migration_bandwidth', 0, + 'Define live migration behavior') +flags.DEFINE_string('qemu_img', 'qemu-img', + 'binary to use for qemu-img commands') def get_connection(read_only): @@ -120,8 +137,8 @@ def get_connection(read_only): def _late_load_cheetah(): global Template if Template is None: - t = __import__('Cheetah.Template', globals(), locals(), ['Template'], - -1) + t = __import__('Cheetah.Template', globals(), locals(), + ['Template'], -1) Template = t.Template @@ -140,12 +157,60 @@ def _get_ip_version(cidr): return int(net.version()) -class LibvirtConnection(object): +def _get_network_info(instance): + # TODO(adiantum) If we will keep this function + # we should cache network_info + admin_context = context.get_admin_context() + + ip_addresses = db.fixed_ip_get_all_by_instance(admin_context, + instance['id']) + + networks = db.network_get_all_by_instance(admin_context, + instance['id']) + network_info = [] + + def ip_dict(ip): + return { + "ip": ip.address, + "netmask": network["netmask"], + "enabled": "1"} + + def ip6_dict(ip6): + prefix = ip6.network.cidr_v6 + mac = instance.mac_address + return { + "ip": utils.to_global_ipv6(prefix, mac), + "netmask": ip6.network.netmask_v6, + "gateway": ip6.network.gateway_v6, + "enabled": "1"} + + for network in networks: + network_ips = [ip for ip in ip_addresses + if ip.network_id == network.id] + + mapping = { + 'label': network['label'], + 'gateway': network['gateway'], + 'mac': instance.mac_address, + 'dns': [network['dns']], + 'ips': [ip_dict(ip) for ip in network_ips]} + + if FLAGS.use_ipv6: + mapping['ip6s'] = [ip6_dict(ip) for ip in network_ips] + + network_info.append((network, mapping)) + return network_info + + +class LibvirtConnection(driver.ComputeDriver): def __init__(self, read_only): + super(LibvirtConnection, self).__init__() self.libvirt_uri = self.get_uri() self.libvirt_xml = open(FLAGS.libvirt_xml_template).read() + self.interfaces_xml = open(FLAGS.injected_network_template).read() + self.cpuinfo_xml = open(FLAGS.cpuinfo_xml_template).read() self._wrapped_conn = None self.read_only = read_only @@ -202,6 +267,8 @@ class LibvirtConnection(object): uri = FLAGS.libvirt_uri or 'uml:///system' elif FLAGS.libvirt_type == 'xen': uri = FLAGS.libvirt_uri or 'xen:///' + elif FLAGS.libvirt_type == 'lxc': + uri = FLAGS.libvirt_uri or 'lxc:///' else: uri = FLAGS.libvirt_uri or 'qemu:///system' return uri @@ -220,6 +287,29 @@ class LibvirtConnection(object): return [self._conn.lookupByID(x).name() for x in self._conn.listDomainsID()] + def _map_to_instance_info(self, domain): + """Gets info from a virsh domain object into an InstanceInfo""" + + # domain.info() returns a list of: + # state: one of the state values (virDomainState) + # maxMemory: the maximum memory used by the domain + # memory: the current amount of memory used by the domain + # nbVirtCPU: the number of virtual CPU + # puTime: the time used by the domain in nanoseconds + + (state, _max_mem, _mem, _num_cpu, _cpu_time) = domain.info() + name = domain.name() + + return driver.InstanceInfo(name, state) + + def list_instances_detail(self): + infos = [] + for domain_id in self._conn.listDomainsID(): + domain = self._conn.lookupByID(domain_id) + info = self._map_to_instance_info(domain) + infos.append(info) + return infos + def destroy(self, instance, cleanup=True): try: virt_dom = self._conn.lookupByName(instance['name']) @@ -257,6 +347,8 @@ class LibvirtConnection(object): instance_name = instance['name'] LOG.info(_('instance %(instance_name)s: deleting instance files' ' %(target)s') % locals()) + if FLAGS.libvirt_type == 'lxc': + disk.destroy_container(target, instance, nbd=FLAGS.use_cow_images) if os.path.exists(target): shutil.rmtree(target) @@ -315,16 +407,77 @@ class LibvirtConnection(object): @exception.wrap_exception def snapshot(self, instance, image_id): - """ Create snapshot from a running VM instance """ - raise NotImplementedError( - _("Instance snapshotting is not supported for libvirt" - "at this time")) + """Create snapshot from a running VM instance. + + This command only works with qemu 0.14+, the qemu_img flag is + provided so that a locally compiled binary of qemu-img can be used + to support this command. + + """ + image_service = utils.import_object(FLAGS.image_service) + virt_dom = self._conn.lookupByName(instance['name']) + elevated = context.get_admin_context() + + base = image_service.show(elevated, instance['image_id']) + + metadata = {'disk_format': base['disk_format'], + 'container_format': base['container_format'], + 'is_public': False, + 'properties': {'architecture': base['architecture'], + 'type': base['type'], + 'name': '%s.%s' % (base['name'], image_id), + 'kernel_id': instance['kernel_id'], + 'image_location': 'snapshot', + 'image_state': 'available', + 'owner_id': instance['project_id'], + 'ramdisk_id': instance['ramdisk_id'], + } + } + + # Make the snapshot + snapshot_name = uuid.uuid4().hex + snapshot_xml = """ + <domainsnapshot> + <name>%s</name> + </domainsnapshot> + """ % snapshot_name + snapshot_ptr = virt_dom.snapshotCreateXML(snapshot_xml, 0) + + # Find the disk + xml_desc = virt_dom.XMLDesc(0) + domain = ElementTree.fromstring(xml_desc) + source = domain.find('devices/disk/source') + disk_path = source.get('file') + + # Export the snapshot to a raw image + temp_dir = tempfile.mkdtemp() + out_path = os.path.join(temp_dir, snapshot_name) + qemu_img_cmd = '%s convert -f qcow2 -O raw -s %s %s %s' % ( + FLAGS.qemu_img, + snapshot_name, + disk_path, + out_path) + utils.execute(qemu_img_cmd) + + # Upload that image to the image service + with open(out_path) as image_file: + image_service.update(elevated, + image_id, + metadata, + image_file) + + # Clean up + shutil.rmtree(temp_dir) @exception.wrap_exception def reboot(self, instance): self.destroy(instance, False) xml = self.to_xml(instance) + self.firewall_driver.setup_basic_filtering(instance) + self.firewall_driver.prepare_instance_filter(instance) self._conn.createXML(xml, 0) + self.firewall_driver.apply_instance_filter(instance) + timer = utils.LoopingCall(f=None) def _wait_for_reboot(): @@ -347,22 +500,22 @@ class LibvirtConnection(object): @exception.wrap_exception def pause(self, instance, callback): - raise exception.APIError("pause not supported for libvirt.") + raise exception.ApiError("pause not supported for libvirt.") @exception.wrap_exception def unpause(self, instance, callback): - raise exception.APIError("unpause not supported for libvirt.") + raise exception.ApiError("unpause not supported for libvirt.") @exception.wrap_exception def suspend(self, instance, callback): - raise exception.APIError("suspend not supported for libvirt") + raise exception.ApiError("suspend not supported for libvirt") @exception.wrap_exception def resume(self, instance, callback): - raise exception.APIError("resume not supported for libvirt") + raise exception.ApiError("resume not supported for libvirt") @exception.wrap_exception - def rescue(self, instance): + def rescue(self, instance, callback=None): self.destroy(instance, False) xml = self.to_xml(instance, rescue=True) @@ -392,21 +545,27 @@ class LibvirtConnection(object): return timer.start(interval=0.5, now=True) @exception.wrap_exception - def unrescue(self, instance): + def unrescue(self, instance, callback=None): # NOTE(vish): Because reboot destroys and recreates an instance using # the normal xml file, we can just call reboot here self.reboot(instance) @exception.wrap_exception - def spawn(self, instance): - xml = self.to_xml(instance) + def poll_rescued_instances(self, timeout): + pass + + # NOTE(ilyaalekseyev): Implementation like in multinics + # for xenapi(tr3buchet) + @exception.wrap_exception + def spawn(self, instance, network_info=None): + xml = self.to_xml(instance, network_info) db.instance_set_state(context.get_admin_context(), instance['id'], power_state.NOSTATE, 'launching') - self.firewall_driver.setup_basic_filtering(instance) - self.firewall_driver.prepare_instance_filter(instance) - self._create_image(instance, xml) + self.firewall_driver.setup_basic_filtering(instance, network_info) + self.firewall_driver.prepare_instance_filter(instance, network_info) + self._create_image(instance, xml, network_info) self._conn.createXML(xml, 0) LOG.debug(_("instance %s: is running"), instance['name']) self.firewall_driver.apply_instance_filter(instance) @@ -438,8 +597,10 @@ class LibvirtConnection(object): if virsh_output.startswith('/dev/'): LOG.info(_("cool, it's a device")) - out, err = utils.execute("sudo dd if=%s iflag=nonblock" % - virsh_output, check_exit_code=False) + out, err = utils.execute('sudo', 'dd', + "if=%s" % virsh_output, + 'iflag=nonblock', + check_exit_code=False) return out else: return '' @@ -461,14 +622,17 @@ class LibvirtConnection(object): console_log = os.path.join(FLAGS.instances_path, instance['name'], 'console.log') - utils.execute('sudo chown %d %s' % (os.getuid(), console_log)) + utils.execute('sudo', 'chown', os.getuid(), console_log) if FLAGS.libvirt_type == 'xen': # Xen is special - virsh_output = utils.execute("virsh ttyconsole %s" % + virsh_output = utils.execute('virsh', 'ttyconsole', instance['name']) data = self._flush_xen_console(virsh_output) fpath = self._append_to_file(data, console_log) + elif FLAGS.libvirt_type == 'lxc': + # LXC is also special + LOG.info(_("Unable to read LXC console")) else: fpath = console_log @@ -482,9 +646,10 @@ class LibvirtConnection(object): port = random.randint(int(start_port), int(end_port)) # netcat will exit with 0 only if the port is in use, # so a nonzero return value implies it is unused - cmd = 'netcat 0.0.0.0 %s -w 1 </dev/null || echo free' % (port) - stdout, stderr = utils.execute(cmd) - if stdout.strip() == 'free': + cmd = 'netcat', '0.0.0.0', port, '-w', '1' + try: + stdout, stderr = utils.execute(*cmd, process_input='') + except exception.ProcessExecutionError: return port raise Exception(_('Unable to find an open port')) @@ -511,7 +676,26 @@ class LibvirtConnection(object): subprocess.Popen(cmd, shell=True) return {'token': token, 'host': host, 'port': port} - def _cache_image(self, fn, target, fname, cow=False, *args, **kwargs): + @exception.wrap_exception + def get_vnc_console(self, instance): + def get_vnc_port_for_instance(instance_name): + virt_dom = self._conn.lookupByName(instance_name) + xml = virt_dom.XMLDesc(0) + # TODO: use etree instead of minidom + dom = minidom.parseString(xml) + + for graphic in dom.getElementsByTagName('graphics'): + if graphic.getAttribute('type') == 'vnc': + return graphic.getAttribute('port') + + port = get_vnc_port_for_instance(instance['name']) + token = str(uuid.uuid4()) + host = instance['host'] + + return {'token': token, 'host': host, 'port': port} + + @staticmethod + def _cache_image(fn, target, fname, cow=False, *args, **kwargs): """Wrapper for a method that creates an image that caches the image. This wrapper will save the image into a common store and create a @@ -530,14 +714,20 @@ class LibvirtConnection(object): if not os.path.exists(base_dir): os.mkdir(base_dir) base = os.path.join(base_dir, fname) - if not os.path.exists(base): - fn(target=base, *args, **kwargs) + + @utils.synchronized(fname) + def call_if_not_exists(base, fn, *args, **kwargs): + if not os.path.exists(base): + fn(target=base, *args, **kwargs) + + call_if_not_exists(base, fn, *args, **kwargs) + if cow: - utils.execute('qemu-img create -f qcow2 -o ' - 'cluster_size=2M,backing_file=%s %s' - % (base, target)) + utils.execute('qemu-img', 'create', '-f', 'qcow2', '-o', + 'cluster_size=2M,backing_file=%s' % base, + target) else: - utils.execute('cp %s %s' % (base, target)) + utils.execute('cp', base, target) def _fetch_image(self, target, image_id, user, project, size=None): """Grab image and optionally attempt to resize it""" @@ -547,10 +737,17 @@ class LibvirtConnection(object): def _create_local(self, target, local_gb): """Create a blank image of specified size""" - utils.execute('truncate %s -s %dG' % (target, local_gb)) + utils.execute('truncate', target, '-s', "%dG" % local_gb) # TODO(vish): should we format disk by default? - def _create_image(self, inst, libvirt_xml, suffix='', disk_images=None): + def _create_image(self, inst, libvirt_xml, suffix='', disk_images=None, + network_info=None): + if not network_info: + network_info = _get_network_info(inst) + + if not suffix: + suffix = '' + # syntactic nicety def basepath(fname='', suffix=suffix): return os.path.join(FLAGS.instances_path, @@ -558,13 +755,17 @@ class LibvirtConnection(object): fname + suffix) # ensure directories exist and are writable - utils.execute('mkdir -p %s' % basepath(suffix='')) + utils.execute('mkdir', '-p', basepath(suffix='')) LOG.info(_('instance %s: Creating image'), inst['name']) f = open(basepath('libvirt.xml'), 'w') f.write(libvirt_xml) f.close() + if FLAGS.libvirt_type == 'lxc': + container_dir = '%s/rootfs' % basepath(suffix='') + utils.execute('mkdir', '-p', container_dir) + # NOTE(vish): No need add the suffix to console.log os.close(os.open(basepath('console.log', ''), os.O_CREAT | os.O_WRONLY, 0660)) @@ -578,21 +779,23 @@ class LibvirtConnection(object): 'ramdisk_id': inst['ramdisk_id']} if disk_images['kernel_id']: + fname = '%08x' % int(disk_images['kernel_id']) self._cache_image(fn=self._fetch_image, target=basepath('kernel'), - fname=disk_images['kernel_id'], + fname=fname, image_id=disk_images['kernel_id'], user=user, project=project) if disk_images['ramdisk_id']: + fname = '%08x' % int(disk_images['ramdisk_id']) self._cache_image(fn=self._fetch_image, target=basepath('ramdisk'), - fname=disk_images['ramdisk_id'], + fname=fname, image_id=disk_images['ramdisk_id'], user=user, project=project) - root_fname = disk_images['image_id'] + root_fname = '%08x' % int(disk_images['image_id']) size = FLAGS.minimum_root_size if inst['instance_type'] == 'm1.tiny' or suffix == '.rescue': size = None @@ -606,7 +809,7 @@ class LibvirtConnection(object): user=user, project=project, size=size) - type_data = instance_types.INSTANCE_TYPES[inst['instance_type']] + type_data = instance_types.get_instance_type(inst['instance_type']) if type_data['local_gb']: self._cache_image(fn=self._create_local, @@ -622,23 +825,47 @@ class LibvirtConnection(object): if not inst['kernel_id']: target_partition = "1" - key = str(inst['key_data']) + if FLAGS.libvirt_type == 'lxc': + target_partition = None + + if inst['key_data']: + key = str(inst['key_data']) + else: + key = None net = None - network_ref = db.network_get_by_instance(context.get_admin_context(), - inst['id']) - if network_ref['injected']: - admin_context = context.get_admin_context() - address = db.instance_get_fixed_address(admin_context, inst['id']) - ra_server = network_ref['ra_server'] - if not ra_server: - ra_server = "fd00::" - with open(FLAGS.injected_network_template) as f: - net = f.read() % {'address': address, - 'netmask': network_ref['netmask'], - 'gateway': network_ref['gateway'], - 'broadcast': network_ref['broadcast'], - 'dns': network_ref['dns'], - 'ra_server': ra_server} + + nets = [] + ifc_template = open(FLAGS.injected_network_template).read() + ifc_num = -1 + have_injected_networks = False + admin_context = context.get_admin_context() + for (network_ref, mapping) in network_info: + ifc_num += 1 + + if not network_ref['injected']: + continue + + have_injected_networks = True + address = mapping['ips'][0]['ip'] + address_v6 = None + if FLAGS.use_ipv6: + address_v6 = mapping['ip6s'][0]['ip'] + net_info = {'name': 'eth%d' % ifc_num, + 'address': address, + 'netmask': network_ref['netmask'], + 'gateway': network_ref['gateway'], + 'broadcast': network_ref['broadcast'], + 'dns': network_ref['dns'], + 'address_v6': address_v6, + 'gateway_v6': network_ref['gateway_v6'], + 'netmask_v6': network_ref['netmask_v6']} + nets.append(net_info) + + if have_injected_networks: + net = str(Template(ifc_template, + searchList=[{'interfaces': nets, + 'use_ipv6': FLAGS.use_ipv6}])) + if key or net: inst_name = inst['name'] img_id = inst.image_id @@ -652,27 +879,24 @@ class LibvirtConnection(object): disk.inject_data(basepath('disk'), key, net, partition=target_partition, nbd=FLAGS.use_cow_images) + + if FLAGS.libvirt_type == 'lxc': + disk.setup_container(basepath('disk'), + container_dir=container_dir, + nbd=FLAGS.use_cow_images) except Exception as e: # This could be a windows image, or a vmdk format disk LOG.warn(_('instance %(inst_name)s: ignoring error injecting' ' data into image %(img_id)s (%(e)s)') % locals()) if FLAGS.libvirt_type == 'uml': - utils.execute('sudo chown root %s' % basepath('disk')) + utils.execute('sudo', 'chown', 'root', basepath('disk')) - def to_xml(self, instance, rescue=False): - # TODO(termie): cache? - LOG.debug(_('instance %s: starting toXML method'), instance['name']) - network = db.network_get_by_instance(context.get_admin_context(), - instance['id']) - # FIXME(vish): stick this in db - instance_type = instance['instance_type'] - instance_type = instance_types.INSTANCE_TYPES[instance_type] - ip_address = db.instance_get_fixed_address(context.get_admin_context(), - instance['id']) + def _get_nic_for_xml(self, network, mapping): # Assume that the gateway also acts as the dhcp server. dhcp_server = network['gateway'] - ra_server = network['ra_server'] + gateway_v6 = network['gateway_v6'] + mac_id = mapping['mac'].replace(':', '') if FLAGS.allow_project_net_traffic: if FLAGS.use_ipv6: @@ -697,6 +921,38 @@ class LibvirtConnection(object): (net, mask) else: extra_params = "\n" + + result = { + 'id': mac_id, + 'bridge_name': network['bridge'], + 'mac_address': mapping['mac'], + 'ip_address': mapping['ips'][0]['ip'], + 'dhcp_server': dhcp_server, + 'extra_params': extra_params, + } + + if gateway_v6: + result['gateway_v6'] = gateway_v6 + "/128" + + return result + + def to_xml(self, instance, rescue=False, network_info=None): + # TODO(termie): cache? + LOG.debug(_('instance %s: starting toXML method'), instance['name']) + + # TODO(adiantum) remove network_info creation code + # when multinics will be completed + if not network_info: + network_info = _get_network_info(instance) + + nics = [] + for (network, mapping) in network_info: + nics.append(self._get_nic_for_xml(network, + mapping)) + # FIXME(vish): stick this in db + instance_type_name = instance['instance_type'] + instance_type = instance_types.get_instance_type(instance_type_name) + if FLAGS.use_cow_images: driver_type = 'qcow2' else: @@ -708,17 +964,13 @@ class LibvirtConnection(object): instance['name']), 'memory_kb': instance_type['memory_mb'] * 1024, 'vcpus': instance_type['vcpus'], - 'bridge_name': network['bridge'], - 'mac_address': instance['mac_address'], - 'ip_address': ip_address, - 'dhcp_server': dhcp_server, - 'extra_params': extra_params, 'rescue': rescue, 'local': instance_type['local_gb'], - 'driver_type': driver_type} + 'driver_type': driver_type, + 'nics': nics} - if ra_server: - xml_info['ra_server'] = ra_server + "/128" + if FLAGS.vnc_enabled: + xml_info['vncserver_host'] = FLAGS.vncserver_host if not rescue: if instance['kernel_id']: xml_info['kernel'] = xml_info['basepath'] + "/kernel" @@ -731,7 +983,6 @@ class LibvirtConnection(object): xml = str(Template(self.libvirt_xml, searchList=[xml_info])) LOG.debug(_('instance %s: finished toXML method'), instance['name']) - return xml def get_info(self, instance_name): @@ -748,7 +999,7 @@ class LibvirtConnection(object): 'cpu_time': cpu_time} def get_diagnostics(self, instance_name): - raise exception.APIError(_("diagnostics are not supported " + raise exception.ApiError(_("diagnostics are not supported " "for libvirt")) def get_disks(self, instance_name): @@ -835,6 +1086,181 @@ class LibvirtConnection(object): return interfaces + def get_vcpu_total(self): + """Get vcpu number of physical computer. + + :returns: the number of cpu core. + + """ + + # On certain platforms, this will raise a NotImplementedError. + try: + return multiprocessing.cpu_count() + except NotImplementedError: + LOG.warn(_("Cannot get the number of cpu, because this " + "function is not implemented for this platform. " + "This error can be safely ignored for now.")) + return 0 + + def get_memory_mb_total(self): + """Get the total memory size(MB) of physical computer. + + :returns: the total amount of memory(MB). + + """ + + if sys.platform.upper() != 'LINUX2': + return 0 + + meminfo = open('/proc/meminfo').read().split() + idx = meminfo.index('MemTotal:') + # transforming kb to mb. + return int(meminfo[idx + 1]) / 1024 + + def get_local_gb_total(self): + """Get the total hdd size(GB) of physical computer. + + :returns: + The total amount of HDD(GB). + Note that this value shows a partition where + NOVA-INST-DIR/instances mounts. + + """ + + hddinfo = os.statvfs(FLAGS.instances_path) + return hddinfo.f_frsize * hddinfo.f_blocks / 1024 / 1024 / 1024 + + def get_vcpu_used(self): + """ Get vcpu usage number of physical computer. + + :returns: The total number of vcpu that currently used. + + """ + + total = 0 + for dom_id in self._conn.listDomainsID(): + dom = self._conn.lookupByID(dom_id) + total += len(dom.vcpus()[1]) + return total + + def get_memory_mb_used(self): + """Get the free memory size(MB) of physical computer. + + :returns: the total usage of memory(MB). + + """ + + if sys.platform.upper() != 'LINUX2': + return 0 + + m = open('/proc/meminfo').read().split() + idx1 = m.index('MemFree:') + idx2 = m.index('Buffers:') + idx3 = m.index('Cached:') + avail = (int(m[idx1 + 1]) + int(m[idx2 + 1]) + int(m[idx3 + 1])) / 1024 + return self.get_memory_mb_total() - avail + + def get_local_gb_used(self): + """Get the free hdd size(GB) of physical computer. + + :returns: + The total usage of HDD(GB). + Note that this value shows a partition where + NOVA-INST-DIR/instances mounts. + + """ + + hddinfo = os.statvfs(FLAGS.instances_path) + avail = hddinfo.f_frsize * hddinfo.f_bavail / 1024 / 1024 / 1024 + return self.get_local_gb_total() - avail + + def get_hypervisor_type(self): + """Get hypervisor type. + + :returns: hypervisor type (ex. qemu) + + """ + + return self._conn.getType() + + def get_hypervisor_version(self): + """Get hypervisor version. + + :returns: hypervisor version (ex. 12003) + + """ + + # NOTE(justinsb): getVersion moved between libvirt versions + # Trying to do be compatible with older versions is a lost cause + # But ... we can at least give the user a nice message + method = getattr(self._conn, 'getVersion', None) + if method is None: + raise exception.Error(_("libvirt version is too old" + " (does not support getVersion)")) + # NOTE(justinsb): If we wanted to get the version, we could: + # method = getattr(libvirt, 'getVersion', None) + # NOTE(justinsb): This would then rely on a proper version check + + return method() + + def get_cpu_info(self): + """Get cpuinfo information. + + Obtains cpu feature from virConnect.getCapabilities, + and returns as a json string. + + :return: see above description + + """ + + xml = self._conn.getCapabilities() + xml = libxml2.parseDoc(xml) + nodes = xml.xpathEval('//host/cpu') + if len(nodes) != 1: + raise exception.Invalid(_("Invalid xml. '<cpu>' must be 1," + "but %d\n") % len(nodes) + + xml.serialize()) + + cpu_info = dict() + + arch_nodes = xml.xpathEval('//host/cpu/arch') + if arch_nodes: + cpu_info['arch'] = arch_nodes[0].getContent() + + model_nodes = xml.xpathEval('//host/cpu/model') + if model_nodes: + cpu_info['model'] = model_nodes[0].getContent() + + vendor_nodes = xml.xpathEval('//host/cpu/vendor') + if vendor_nodes: + cpu_info['vendor'] = vendor_nodes[0].getContent() + + topology_nodes = xml.xpathEval('//host/cpu/topology') + topology = dict() + if topology_nodes: + topology_node = topology_nodes[0].get_properties() + while topology_node: + name = topology_node.get_name() + topology[name] = topology_node.getContent() + topology_node = topology_node.get_next() + + keys = ['cores', 'sockets', 'threads'] + tkeys = topology.keys() + if set(tkeys) != set(keys): + ks = ', '.join(keys) + raise exception.Invalid(_("Invalid xml: topology" + "(%(topology)s) must have " + "%(ks)s") % locals()) + + feature_nodes = xml.xpathEval('//host/cpu/feature') + features = list() + for nodes in feature_nodes: + features.append(nodes.get_properties().getContent()) + + cpu_info['topology'] = topology + cpu_info['features'] = features + return utils.dumps(cpu_info) + def block_stats(self, instance_name, disk): """ Note that this function takes an instance name, not an Instance, so @@ -868,9 +1294,214 @@ class LibvirtConnection(object): def refresh_provider_fw_rules(self): self.firewall_driver.refresh_provider_fw_rules() + def update_available_resource(self, ctxt, host): + """Updates compute manager resource info on ComputeNode table. + + This method is called when nova-coompute launches, and + whenever admin executes "nova-manage service update_resource". + + :param ctxt: security context + :param host: hostname that compute manager is currently running + + """ + + try: + service_ref = db.service_get_all_compute_by_host(ctxt, host)[0] + except exception.NotFound: + raise exception.Invalid(_("Cannot update compute manager " + "specific info, because no service " + "record was found.")) + + # Updating host information + dic = {'vcpus': self.get_vcpu_total(), + 'memory_mb': self.get_memory_mb_total(), + 'local_gb': self.get_local_gb_total(), + 'vcpus_used': self.get_vcpu_used(), + 'memory_mb_used': self.get_memory_mb_used(), + 'local_gb_used': self.get_local_gb_used(), + 'hypervisor_type': self.get_hypervisor_type(), + 'hypervisor_version': self.get_hypervisor_version(), + 'cpu_info': self.get_cpu_info()} + + compute_node_ref = service_ref['compute_node'] + if not compute_node_ref: + LOG.info(_('Compute_service record created for %s ') % host) + dic['service_id'] = service_ref['id'] + db.compute_node_create(ctxt, dic) + else: + LOG.info(_('Compute_service record updated for %s ') % host) + db.compute_node_update(ctxt, compute_node_ref[0]['id'], dic) + + def compare_cpu(self, cpu_info): + """Checks the host cpu is compatible to a cpu given by xml. + + "xml" must be a part of libvirt.openReadonly().getCapabilities(). + return values follows by virCPUCompareResult. + if 0 > return value, do live migration. + 'http://libvirt.org/html/libvirt-libvirt.html#virCPUCompareResult' + + :param cpu_info: json string that shows cpu feature(see get_cpu_info()) + :returns: + None. if given cpu info is not compatible to this server, + raise exception. + + """ + + LOG.info(_('Instance launched has CPU info:\n%s') % cpu_info) + dic = utils.loads(cpu_info) + xml = str(Template(self.cpuinfo_xml, searchList=dic)) + LOG.info(_('to xml...\n:%s ' % xml)) + + u = "http://libvirt.org/html/libvirt-libvirt.html#virCPUCompareResult" + m = _("CPU doesn't have compatibility.\n\n%(ret)s\n\nRefer to %(u)s") + # unknown character exists in xml, then libvirt complains + try: + ret = self._conn.compareCPU(xml, 0) + except libvirt.libvirtError, e: + ret = e.message + LOG.error(m % locals()) + raise + + if ret <= 0: + raise exception.Invalid(m % locals()) + + return + + def ensure_filtering_rules_for_instance(self, instance_ref, + time=None): + """Setting up filtering rules and waiting for its completion. + + To migrate an instance, filtering rules to hypervisors + and firewalls are inevitable on destination host. + ( Waiting only for filterling rules to hypervisor, + since filtering rules to firewall rules can be set faster). + + Concretely, the below method must be called. + - setup_basic_filtering (for nova-basic, etc.) + - prepare_instance_filter(for nova-instance-instance-xxx, etc.) + + to_xml may have to be called since it defines PROJNET, PROJMASK. + but libvirt migrates those value through migrateToURI(), + so , no need to be called. + + Don't use thread for this method since migration should + not be started when setting-up filtering rules operations + are not completed. + + :params instance_ref: nova.db.sqlalchemy.models.Instance object + + """ + + if not time: + time = greenthread + + # If any instances never launch at destination host, + # basic-filtering must be set here. + self.firewall_driver.setup_basic_filtering(instance_ref) + # setting up n)ova-instance-instance-xx mainly. + self.firewall_driver.prepare_instance_filter(instance_ref) + + # wait for completion + timeout_count = range(FLAGS.live_migration_retry_count) + while timeout_count: + try: + filter_name = 'nova-instance-%s' % instance_ref.name + self._conn.nwfilterLookupByName(filter_name) + break + except libvirt.libvirtError: + timeout_count.pop() + if len(timeout_count) == 0: + ec2_id = instance_ref['hostname'] + iname = instance_ref.name + msg = _('Timeout migrating for %(ec2_id)s(%(iname)s)') + raise exception.Error(msg % locals()) + time.sleep(1) + + def live_migration(self, ctxt, instance_ref, dest, + post_method, recover_method): + """Spawning live_migration operation for distributing high-load. + + :params ctxt: security context + :params instance_ref: + nova.db.sqlalchemy.models.Instance object + instance object that is migrated. + :params dest: destination host + :params post_method: + post operation method. + expected nova.compute.manager.post_live_migration. + :params recover_method: + recovery method when any exception occurs. + expected nova.compute.manager.recover_live_migration. + + """ + + greenthread.spawn(self._live_migration, ctxt, instance_ref, dest, + post_method, recover_method) + + def _live_migration(self, ctxt, instance_ref, dest, + post_method, recover_method): + """Do live migration. + + :params ctxt: security context + :params instance_ref: + nova.db.sqlalchemy.models.Instance object + instance object that is migrated. + :params dest: destination host + :params post_method: + post operation method. + expected nova.compute.manager.post_live_migration. + :params recover_method: + recovery method when any exception occurs. + expected nova.compute.manager.recover_live_migration. + + """ + + # Do live migration. + try: + flaglist = FLAGS.live_migration_flag.split(',') + flagvals = [getattr(libvirt, x.strip()) for x in flaglist] + logical_sum = reduce(lambda x, y: x | y, flagvals) + + if self.read_only: + tmpconn = self._connect(self.libvirt_uri, False) + dom = tmpconn.lookupByName(instance_ref.name) + dom.migrateToURI(FLAGS.live_migration_uri % dest, + logical_sum, + None, + FLAGS.live_migration_bandwidth) + tmpconn.close() + else: + dom = self._conn.lookupByName(instance_ref.name) + dom.migrateToURI(FLAGS.live_migration_uri % dest, + logical_sum, + None, + FLAGS.live_migration_bandwidth) + + except Exception: + recover_method(ctxt, instance_ref) + raise + + # Waiting for completion of live_migration. + timer = utils.LoopingCall(f=None) + + def wait_for_live_migration(): + """waiting for live migration completion""" + try: + self.get_info(instance_ref.name)['state'] + except exception.NotFound: + timer.stop() + post_method(ctxt, instance_ref, dest) + + timer.f = wait_for_live_migration + timer.start(interval=0.5, now=True) + + def unfilter_instance(self, instance_ref): + """See comments of same method in firewall_driver.""" + self.firewall_driver.unfilter_instance(instance_ref) + class FirewallDriver(object): - def prepare_instance_filter(self, instance): + def prepare_instance_filter(self, instance, network_info=None): """Prepare filters for the instance. At this point, the instance isn't running yet.""" @@ -911,7 +1542,7 @@ class FirewallDriver(object): the list of rules (via admin api).""" raise NotImplementedError() - def setup_basic_filtering(self, instance): + def setup_basic_filtering(self, instance, network_info=None): """Create rules to block spoofing and allow dhcp. This gets called when spawning an instance, before @@ -920,11 +1551,6 @@ class FirewallDriver(object): """ raise NotImplementedError() - def _ra_server_for_instance(self, instance): - network = db.network_get_by_instance(context.get_admin_context(), - instance['id']) - return network['ra_server'] - class NWFilterFirewall(FirewallDriver): """ @@ -1017,10 +1643,13 @@ class NWFilterFirewall(FirewallDriver): </rule> </filter>''' - def setup_basic_filtering(self, instance): + def setup_basic_filtering(self, instance, network_info=None): """Set up basic filtering (MAC, IP, and ARP spoofing protection)""" logging.info('called setup_basic_filtering in nwfilter') + if not network_info: + network_info = _get_network_info(instance) + if self.handle_security_groups: # No point in setting up a filter set that we'll be overriding # anyway. @@ -1032,9 +1661,11 @@ class NWFilterFirewall(FirewallDriver): logging.info('ensuring intermediate filters') self._ensure_intermediate_filters() - instance_filter_name = self._instance_filter_name(instance) - self._define_filter(self._filter_container(instance_filter_name, - ['nova-base'])) + for (network, mapping) in network_info: + nic_id = mapping['mac'].replace(':', '') + instance_filter_name = self._instance_filter_name(instance, nic_id) + self._define_filter(self._filter_container(instance_filter_name, + ['nova-base'])) def _ensure_static_filters(self): """Static filters are filters that have no need to be IP aware. @@ -1075,9 +1706,7 @@ class NWFilterFirewall(FirewallDriver): if self.intermediate_filters_configured: return - self.refresh_provider_fw_rules() - self.intermediate_filters_configured = True def _filter_container(self, name, filters): @@ -1148,49 +1777,54 @@ class NWFilterFirewall(FirewallDriver): # Nothing to do pass - def prepare_instance_filter(self, instance): + def prepare_instance_filter(self, instance, network_info=None): """ Creates an NWFilter for the given instance. In the process, it makes sure the filters for the security groups as well as the base filter are all in place. """ + if not network_info: + network_info = _get_network_info(instance) if instance['image_id'] == FLAGS.vpn_image_id: base_filter = 'nova-vpn' else: base_filter = 'nova-base' - instance_filter_name = self._instance_filter_name(instance) - instance_secgroup_filter_name = '%s-secgroup' % (instance_filter_name,) - instance_filter_children = [base_filter, 'nova-provider-rules', - instance_secgroup_filter_name] instance_secgroup_filter_children = ['nova-base-ipv4', 'nova-base-ipv6', 'nova-allow-dhcp-server'] - if FLAGS.use_ipv6: - ra_server = self._ra_server_for_instance(instance) - if ra_server: - instance_secgroup_filter_children += ['nova-allow-ra-server'] - ctxt = context.get_admin_context() - - if FLAGS.allow_project_net_traffic: - instance_filter_children += ['nova-project'] - if FLAGS.use_ipv6: - instance_filter_children += ['nova-project-v6'] - - for security_group in db.security_group_get_by_instance(ctxt, - instance['id']): + for security_group in \ + db.security_group_get_by_instance(ctxt, instance['id']): self.refresh_security_group_rules(security_group['id']) instance_secgroup_filter_children += [('nova-secgroup-%s' % - security_group['id'])] + security_group['id'])] - self._define_filter( + self._define_filter( self._filter_container(instance_secgroup_filter_name, instance_secgroup_filter_children)) - self._define_filter( + for (network, mapping) in network_info: + nic_id = mapping['mac'].replace(':', '') + instance_filter_name = self._instance_filter_name(instance, nic_id) + instance_filter_children = [base_filter, 'nova-provider-rules', + instance_secgroup_filter_name] + + if FLAGS.use_ipv6: + gateway_v6 = network['gateway_v6'] + + if gateway_v6: + instance_secgroup_filter_children += \ + ['nova-allow-ra-server'] + + if FLAGS.allow_project_net_traffic: + instance_filter_children += ['nova-project'] + if FLAGS.use_ipv6: + instance_filter_children += ['nova-project-v6'] + + self._define_filter( self._filter_container(instance_filter_name, instance_filter_children)) @@ -1287,89 +1921,159 @@ class NWFilterFirewall(FirewallDriver): xml += "chain='ipv4'>%s</filter>" % rule_xml return xml - def _instance_filter_name(self, instance): - return 'nova-instance-%s' % instance['name'] + def _instance_filter_name(self, instance, nic_id=None): + if not nic_id: + return 'nova-instance-%s' % (instance['name']) + return 'nova-instance-%s-%s' % (instance['name'], nic_id) class IptablesFirewallDriver(FirewallDriver): def __init__(self, execute=None, **kwargs): - self.execute = execute or utils.execute + from nova.network import linux_net + self.iptables = linux_net.iptables_manager self.instances = {} self.nwfilter = NWFilterFirewall(kwargs['get_connection']) - def setup_basic_filtering(self, instance): + self.iptables.ipv4['filter'].add_chain('sg-fallback') + self.iptables.ipv4['filter'].add_rule('sg-fallback', '-j DROP') + self.iptables.ipv6['filter'].add_chain('sg-fallback') + self.iptables.ipv6['filter'].add_rule('sg-fallback', '-j DROP') + + def setup_basic_filtering(self, instance, network_info=None): """Use NWFilter from libvirt for this.""" - return self.nwfilter.setup_basic_filtering(instance) + if not network_info: + network_info = _get_network_info(instance) + return self.nwfilter.setup_basic_filtering(instance, network_info) def apply_instance_filter(self, instance): """No-op. Everything is done in prepare_instance_filter""" pass - def remove_instance(self, instance): - if instance['id'] in self.instances: - del self.instances[instance['id']] + def unfilter_instance(self, instance): + if self.instances.pop(instance['id'], None): + self.remove_filters_for_instance(instance) + self.iptables.apply() else: LOG.info(_('Attempted to unfilter instance %s which is not ' - 'filtered'), instance['id']) + 'filtered'), instance['id']) - def add_instance(self, instance): + def prepare_instance_filter(self, instance, network_info=None): + if not network_info: + network_info = _get_network_info(instance) self.instances[instance['id']] = instance + self.add_filters_for_instance(instance, network_info) + self.iptables.apply() - def unfilter_instance(self, instance): - self.remove_instance(instance) - self.apply_ruleset() + def add_filters_for_instance(self, instance, network_info=None): + if not network_info: + network_info = _get_network_info(instance) + chain_name = self._instance_chain_name(instance) - def prepare_instance_filter(self, instance): - self.add_instance(instance) - self.apply_ruleset() + self.iptables.ipv4['filter'].add_chain(chain_name) - def apply_ruleset(self): - current_filter, _ = self.execute('sudo iptables-save -t filter') - current_lines = current_filter.split('\n') - new_filter = self.modify_rules(current_lines, 4) - self.execute('sudo iptables-restore', - process_input='\n'.join(new_filter)) - if(FLAGS.use_ipv6): - current_filter, _ = self.execute('sudo ip6tables-save -t filter') - current_lines = current_filter.split('\n') - new_filter = self.modify_rules(current_lines, 6) - self.execute('sudo ip6tables-restore', - process_input='\n'.join(new_filter)) + ips_v4 = [ip['ip'] for (_, mapping) in network_info + for ip in mapping['ips']] + + for ipv4_address in ips_v4: + self.iptables.ipv4['filter'].add_rule('local', + '-d %s -j $%s' % + (ipv4_address, chain_name)) + + if FLAGS.use_ipv6: + self.iptables.ipv6['filter'].add_chain(chain_name) + ips_v6 = [ip['ip'] for (_, mapping) in network_info + for ip in mapping['ip6s']] + + for ipv6_address in ips_v6: + self.iptables.ipv6['filter'].add_rule('local', + '-d %s -j $%s' % + (ipv6_address, + chain_name)) - def modify_rules(self, current_lines, ip_version=4): + ipv4_rules, ipv6_rules = self.instance_rules(instance, network_info) + + for rule in ipv4_rules: + self.iptables.ipv4['filter'].add_rule(chain_name, rule) + + if FLAGS.use_ipv6: + for rule in ipv6_rules: + self.iptables.ipv6['filter'].add_rule(chain_name, rule) + + def remove_filters_for_instance(self, instance): + chain_name = self._instance_chain_name(instance) + + self.iptables.ipv4['filter'].remove_chain(chain_name) + if FLAGS.use_ipv6: + self.iptables.ipv6['filter'].remove_chain(chain_name) + + def instance_rules(self, instance, network_info=None): + if not network_info: + network_info = _get_network_info(instance) ctxt = context.get_admin_context() - # Remove any trace of nova rules. - new_filter = filter(lambda l: 'nova-' not in l, current_lines) - - seen_chains = False - for rules_index in range(len(new_filter)): - if not seen_chains: - if new_filter[rules_index].startswith(':'): - seen_chains = True - elif seen_chains == 1: - if not new_filter[rules_index].startswith(':'): - break - our_chains = [':nova-fallback - [0:0]'] - our_rules = ['-A nova-fallback -j DROP'] + ipv4_rules = [] + ipv6_rules = [] + + # Always drop invalid packets + ipv4_rules += ['-m state --state ' 'INVALID -j DROP'] + ipv6_rules += ['-m state --state ' 'INVALID -j DROP'] - our_chains += [':nova-local - [0:0]'] - our_rules += ['-A OUTPUT -j nova-local'] + # Allow established connections + ipv4_rules += ['-m state --state ESTABLISHED,RELATED -j ACCEPT'] + ipv6_rules += ['-m state --state ESTABLISHED,RELATED -j ACCEPT'] - our_chains += [':nova-provider - [0:0]'] - our_rules += ['-A FORWARD -j nova-provider'] + dhcp_servers = [network['gateway'] for (network, _m) in network_info] - # Build all the provider-level drops, then jump to local + for dhcp_server in dhcp_servers: + ipv4_rules.append('-s %s -p udp --sport 67 --dport 68 ' + '-j ACCEPT' % (dhcp_server,)) + + #Allow project network traffic + if FLAGS.allow_project_net_traffic: + cidrs = [network['cidr'] for (network, _m) in network_info] + for cidr in cidrs: + ipv4_rules.append('-s %s -j ACCEPT' % (cidr,)) + + # We wrap these in FLAGS.use_ipv6 because they might cause + # a DB lookup. The other ones are just list operations, so + # they're not worth the clutter. + if FLAGS.use_ipv6: + # Allow RA responses + gateways_v6 = [network['gateway_v6'] for (network, _) in + network_info] + for gateway_v6 in gateways_v6: + ipv6_rules.append( + '-s %s/128 -p icmpv6 -j ACCEPT' % (gateway_v6,)) + + #Allow project network traffic + if FLAGS.allow_project_net_traffic: + cidrv6s = [network['cidr_v6'] for (network, _m) + in network_info] + + for cidrv6 in cidrv6s: + ipv6_rules.append('-s %s -j ACCEPT' % (cidrv6,)) + + # block provider-identified sources rules = db.provider_fw_rule_get_all(ctxt) for rule in rules: logging.info('%r', rule) - version = _get_ip_version(rule.cidr) - if version != ip_version: + + if not rule.cidr: + # Eventually, a mechanism to grant access for security + # groups will turn up here. It'll use ipsets. continue + + version = _get_ip_version(rule.cidr) + if version == 4: + fw_rules = ipv4_rules + else: + fw_rules = ipv6_rules + protocol = rule.protocol if version == 6 and rule.protocol == 'icmp': protocol = 'icmpv6' - args = ['-A nova-provider -p', protocol, '-s', rule.cidr] + + args = ['-A', 'nova-provider', '-p', protocol, '-s', rule.cidr] if rule.protocol in ['udp', 'tcp']: if rule.from_port == rule.to_port: @@ -1390,89 +2094,26 @@ class IptablesFirewallDriver(FirewallDriver): icmp_type_arg += '/%s' % icmp_code if icmp_type_arg: - if(ip_version == 4): + if version == 4: args += ['-m', 'icmp', '--icmp-type', icmp_type_arg] - elif(ip_version == 6): + elif version == 6: args += ['-m', 'icmp6', '--icmpv6-type', icmp_type_arg] args += ['-j DROP'] - our_rules += [' '.join(args)] - - our_rules += ['-A nova-provider -j nova-local'] - - security_groups = {} - # Add our chains - # First, we add instance chains and rules - for instance_id in self.instances: - instance = self.instances[instance_id] - chain_name = self._instance_chain_name(instance) - if(ip_version == 4): - ip_address = self._ip_for_instance(instance) - elif(ip_version == 6): - ip_address = self._ip_for_instance_v6(instance) - - our_chains += [':%s - [0:0]' % chain_name] - - # Jump to the per-instance chain - our_rules += ['-A nova-local -d %s -j %s' % (ip_address, - chain_name)] - - # Always drop invalid packets - our_rules += ['-A %s -m state --state ' - 'INVALID -j DROP' % (chain_name,)] - - # Allow established connections - our_rules += ['-A %s -m state --state ' - 'ESTABLISHED,RELATED -j ACCEPT' % (chain_name,)] - - # Jump to each security group chain in turn - for security_group in \ - db.security_group_get_by_instance(ctxt, - instance['id']): - security_groups[security_group['id']] = security_group - - sg_chain_name = self._security_group_chain_name( - security_group['id']) + fw_rules += [' '.join(args)] - our_rules += ['-A %s -j %s' % (chain_name, sg_chain_name)] - - if(ip_version == 4): - # Allow DHCP responses - dhcp_server = self._dhcp_server_for_instance(instance) - our_rules += ['-A %s -s %s -p udp --sport 67 --dport 68 ' - '-j ACCEPT ' % (chain_name, dhcp_server)] - #Allow project network traffic - if (FLAGS.allow_project_net_traffic): - cidr = self._project_cidr_for_instance(instance) - our_rules += ['-A %s -s %s -j ACCEPT' % (chain_name, cidr)] - elif(ip_version == 6): - # Allow RA responses - ra_server = self._ra_server_for_instance(instance) - if ra_server: - our_rules += ['-A %s -s %s -p icmpv6 -j ACCEPT' % - (chain_name, ra_server + "/128")] - #Allow project network traffic - if (FLAGS.allow_project_net_traffic): - cidrv6 = self._project_cidrv6_for_instance(instance) - our_rules += ['-A %s -s %s -j ACCEPT' % - (chain_name, cidrv6)] - - # If nothing matches, jump to the fallback chain - our_rules += ['-A %s -j nova-fallback' % (chain_name,)] + security_groups = db.security_group_get_by_instance(ctxt, + instance['id']) # then, security group chains and rules - for security_group_id in security_groups: - chain_name = self._security_group_chain_name(security_group_id) - our_chains += [':%s - [0:0]' % chain_name] - - rules = \ - db.security_group_rule_get_by_security_group(ctxt, - security_group_id) + for security_group in security_groups: + rules = db.security_group_rule_get_by_security_group(ctxt, + security_group['id']) for rule in rules: - logging.info('%r', rule) + LOG.debug('%r', rule) if not rule.cidr: # Eventually, a mechanism to grant access for security @@ -1480,14 +2121,16 @@ class IptablesFirewallDriver(FirewallDriver): continue version = _get_ip_version(rule.cidr) - if version != ip_version: - continue + if version == 4: + fw_rules = ipv4_rules + else: + fw_rules = ipv6_rules protocol = rule.protocol if version == 6 and rule.protocol == 'icmp': protocol = 'icmpv6' - args = ['-A', chain_name, '-p', protocol, '-s', rule.cidr] + args = ['-p', protocol, '-s', rule.cidr] if rule.protocol in ['udp', 'tcp']: if rule.from_port == rule.to_port: @@ -1508,26 +2151,33 @@ class IptablesFirewallDriver(FirewallDriver): icmp_type_arg += '/%s' % icmp_code if icmp_type_arg: - if(ip_version == 4): + if version == 4: args += ['-m', 'icmp', '--icmp-type', icmp_type_arg] - elif(ip_version == 6): + elif version == 6: args += ['-m', 'icmp6', '--icmpv6-type', icmp_type_arg] args += ['-j ACCEPT'] - our_rules += [' '.join(args)] + fw_rules += [' '.join(args)] - new_filter[rules_index:rules_index] = our_rules - new_filter[rules_index:rules_index] = our_chains - logging.info('new_filter: %s', '\n'.join(new_filter)) - return new_filter + ipv4_rules += ['-j $sg-fallback'] + ipv6_rules += ['-j $sg-fallback'] + + return ipv4_rules, ipv6_rules def refresh_security_group_members(self, security_group): pass def refresh_security_group_rules(self, security_group): - self.apply_ruleset() + self.do_refresh_security_group_rules(security_group) + self.iptables.apply() + + @utils.synchronized('iptables', external=True) + def do_refresh_security_group_rules(self, security_group): + for instance in self.instances.values(): + self.remove_filters_for_instance(instance) + self.add_filters_for_instance(instance) def refresh_provider_fw_rules(self): self.apply_ruleset() @@ -1536,32 +2186,4 @@ class IptablesFirewallDriver(FirewallDriver): return 'nova-sg-%s' % (security_group_id,) def _instance_chain_name(self, instance): - return 'nova-inst-%s' % (instance['id'],) - - def _ip_for_instance(self, instance): - return db.instance_get_fixed_address(context.get_admin_context(), - instance['id']) - - def _ip_for_instance_v6(self, instance): - return db.instance_get_fixed_address_v6(context.get_admin_context(), - instance['id']) - - def _dhcp_server_for_instance(self, instance): - network = db.network_get_by_instance(context.get_admin_context(), - instance['id']) - return network['gateway'] - - def _ra_server_for_instance(self, instance): - network = db.network_get_by_instance(context.get_admin_context(), - instance['id']) - return network['ra_server'] - - def _project_cidr_for_instance(self, instance): - network = db.network_get_by_instance(context.get_admin_context(), - instance['id']) - return network['cidr'] - - def _project_cidrv6_for_instance(self, instance): - network = db.network_get_by_instance(context.get_admin_context(), - instance['id']) - return network['cidr_v6'] + return 'inst-%s' % (instance['id'],) diff --git a/nova/virt/vmwareapi/__init__.py b/nova/virt/vmwareapi/__init__.py new file mode 100644 index 000000000..d9b27de08 --- /dev/null +++ b/nova/virt/vmwareapi/__init__.py @@ -0,0 +1,19 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright (c) 2011 Citrix Systems, Inc. +# Copyright 2011 OpenStack LLC. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +""" +:mod:`vmwareapi` -- Nova support for VMware ESX/ESXi Server through VMware API. +""" diff --git a/nova/virt/vmwareapi/error_util.py b/nova/virt/vmwareapi/error_util.py new file mode 100644 index 000000000..53fa8f24d --- /dev/null +++ b/nova/virt/vmwareapi/error_util.py @@ -0,0 +1,96 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""
+Exception classes and SOAP response error checking module.
+"""
+
+FAULT_NOT_AUTHENTICATED = "NotAuthenticated"
+FAULT_ALREADY_EXISTS = "AlreadyExists"
+
+
+class VimException(Exception):
+ """The VIM Exception class."""
+
+ def __init__(self, exception_summary, excep):
+ Exception.__init__(self)
+ self.exception_summary = exception_summary
+ self.exception_obj = excep
+
+ def __str__(self):
+ return self.exception_summary + str(self.exception_obj)
+
+
+class SessionOverLoadException(VimException):
+ """Session Overload Exception."""
+ pass
+
+
+class VimAttributeError(VimException):
+ """VI Attribute Error."""
+ pass
+
+
+class VimFaultException(Exception):
+ """The VIM Fault exception class."""
+
+ def __init__(self, fault_list, excep):
+ Exception.__init__(self)
+ self.fault_list = fault_list
+ self.exception_obj = excep
+
+ def __str__(self):
+ return str(self.exception_obj)
+
+
+class FaultCheckers(object):
+ """
+ Methods for fault checking of SOAP response. Per Method error handlers
+ for which we desire error checking are defined. SOAP faults are
+ embedded in the SOAP messages as properties and not as SOAP faults.
+ """
+
+ @staticmethod
+ def retrieveproperties_fault_checker(resp_obj):
+ """
+ Checks the RetrieveProperties response for errors. Certain faults
+ are sent as part of the SOAP body as property of missingSet.
+ For example NotAuthenticated fault.
+ """
+ fault_list = []
+ if not resp_obj:
+ # This is the case when the session has timed out. ESX SOAP server
+ # sends an empty RetrievePropertiesResponse. Normally missingSet in
+ # the returnval field has the specifics about the error, but that's
+ # not the case with a timed out idle session. It is as bad as a
+ # terminated session for we cannot use the session. So setting
+ # fault to NotAuthenticated fault.
+ fault_list = ["NotAuthenticated"]
+ else:
+ for obj_cont in resp_obj:
+ if hasattr(obj_cont, "missingSet"):
+ for missing_elem in obj_cont.missingSet:
+ fault_type = \
+ missing_elem.fault.fault.__class__.__name__
+ # Fault needs to be added to the type of fault for
+ # uniformity in error checking as SOAP faults define
+ fault_list.append(fault_type)
+ if fault_list:
+ exc_msg_list = ', '.join(fault_list)
+ raise VimFaultException(fault_list, Exception(_("Error(s) %s "
+ "occurred in the call to RetrieveProperties") %
+ exc_msg_list))
diff --git a/nova/virt/vmwareapi/fake.py b/nova/virt/vmwareapi/fake.py new file mode 100644 index 000000000..4bb467fa9 --- /dev/null +++ b/nova/virt/vmwareapi/fake.py @@ -0,0 +1,711 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""
+A fake VMWare VI API implementation.
+"""
+
+from pprint import pformat
+import uuid
+
+from nova import exception
+from nova import log as logging
+from nova.virt.vmwareapi import vim
+from nova.virt.vmwareapi import error_util
+
+_CLASSES = ['Datacenter', 'Datastore', 'ResourcePool', 'VirtualMachine',
+ 'Network', 'HostSystem', 'HostNetworkSystem', 'Task', 'session',
+ 'files']
+
+_FAKE_FILE_SIZE = 1024
+
+_db_content = {}
+
+LOG = logging.getLogger("nova.virt.vmwareapi.fake")
+
+
+def log_db_contents(msg=None):
+ """Log DB Contents."""
+ text = msg or ""
+ content = pformat(_db_content)
+ LOG.debug(_("%(text)s: _db_content => %(content)s") % locals())
+
+
+def reset():
+ """Resets the db contents."""
+ for c in _CLASSES:
+ # We fake the datastore by keeping the file references as a list of
+ # names in the db
+ if c == 'files':
+ _db_content[c] = []
+ else:
+ _db_content[c] = {}
+ create_network()
+ create_host_network_system()
+ create_host()
+ create_datacenter()
+ create_datastore()
+ create_res_pool()
+
+
+def cleanup():
+ """Clear the db contents."""
+ for c in _CLASSES:
+ _db_content[c] = {}
+
+
+def _create_object(table, table_obj):
+ """Create an object in the db."""
+ _db_content[table][table_obj.obj] = table_obj
+
+
+def _get_objects(obj_type):
+ """Get objects of the type."""
+ lst_objs = []
+ for key in _db_content[obj_type]:
+ lst_objs.append(_db_content[obj_type][key])
+ return lst_objs
+
+
+class Prop(object):
+ """Property Object base class."""
+
+ def __init__(self):
+ self.name = None
+ self.val = None
+
+
+class ManagedObject(object):
+ """Managed Data Object base class."""
+
+ def __init__(self, name="ManagedObject", obj_ref=None):
+ """Sets the obj property which acts as a reference to the object."""
+ super(ManagedObject, self).__setattr__('objName', name)
+ if obj_ref is None:
+ obj_ref = str(uuid.uuid4())
+ object.__setattr__(self, 'obj', obj_ref)
+ object.__setattr__(self, 'propSet', [])
+
+ def set(self, attr, val):
+ """
+ Sets an attribute value. Not using the __setattr__ directly for we
+ want to set attributes of the type 'a.b.c' and using this function
+ class we set the same.
+ """
+ self.__setattr__(attr, val)
+
+ def get(self, attr):
+ """
+ Gets an attribute. Used as an intermediary to get nested
+ property like 'a.b.c' value.
+ """
+ return self.__getattr__(attr)
+
+ def __setattr__(self, attr, val):
+ for prop in self.propSet:
+ if prop.name == attr:
+ prop.val = val
+ return
+ elem = Prop()
+ elem.name = attr
+ elem.val = val
+ self.propSet.append(elem)
+
+ def __getattr__(self, attr):
+ for elem in self.propSet:
+ if elem.name == attr:
+ return elem.val
+ raise exception.Error(_("Property %(attr)s not set for the managed "
+ "object %(objName)s") %
+ {'attr': attr,
+ 'objName': self.objName})
+
+
+class DataObject(object):
+ """Data object base class."""
+ pass
+
+
+class VirtualDisk(DataObject):
+ """
+ Virtual Disk class. Does nothing special except setting
+ __class__.__name__ to 'VirtualDisk'. Refer place where __class__.__name__
+ is used in the code.
+ """
+ pass
+
+
+class VirtualDiskFlatVer2BackingInfo(DataObject):
+ """VirtualDiskFlatVer2BackingInfo class."""
+ pass
+
+
+class VirtualLsiLogicController(DataObject):
+ """VirtualLsiLogicController class."""
+ pass
+
+
+class VirtualMachine(ManagedObject):
+ """Virtual Machine class."""
+
+ def __init__(self, **kwargs):
+ super(VirtualMachine, self).__init__("VirtualMachine")
+ self.set("name", kwargs.get("name"))
+ self.set("runtime.connectionState",
+ kwargs.get("conn_state", "connected"))
+ self.set("summary.config.guestId", kwargs.get("guest", "otherGuest"))
+ ds_do = DataObject()
+ ds_do.ManagedObjectReference = [kwargs.get("ds").obj]
+ self.set("datastore", ds_do)
+ self.set("summary.guest.toolsStatus", kwargs.get("toolsstatus",
+ "toolsOk"))
+ self.set("summary.guest.toolsRunningStatus", kwargs.get(
+ "toolsrunningstate", "guestToolsRunning"))
+ self.set("runtime.powerState", kwargs.get("powerstate", "poweredOn"))
+ self.set("config.files.vmPathName", kwargs.get("vmPathName"))
+ self.set("summary.config.numCpu", kwargs.get("numCpu", 1))
+ self.set("summary.config.memorySizeMB", kwargs.get("mem", 1))
+ self.set("config.hardware.device", kwargs.get("virtual_disk", None))
+ self.set("config.extraConfig", kwargs.get("extra_config", None))
+
+ def reconfig(self, factory, val):
+ """
+ Called to reconfigure the VM. Actually customizes the property
+ setting of the Virtual Machine object.
+ """
+ try:
+ # Case of Reconfig of VM to attach disk
+ controller_key = val.deviceChange[1].device.controllerKey
+ filename = val.deviceChange[1].device.backing.fileName
+
+ disk = VirtualDisk()
+ disk.controllerKey = controller_key
+
+ disk_backing = VirtualDiskFlatVer2BackingInfo()
+ disk_backing.fileName = filename
+ disk_backing.key = -101
+ disk.backing = disk_backing
+
+ controller = VirtualLsiLogicController()
+ controller.key = controller_key
+
+ self.set("config.hardware.device", [disk, controller])
+ except AttributeError:
+ # Case of Reconfig of VM to set extra params
+ self.set("config.extraConfig", val.extraConfig)
+
+
+class Network(ManagedObject):
+ """Network class."""
+
+ def __init__(self):
+ super(Network, self).__init__("Network")
+ self.set("summary.name", "vmnet0")
+
+
+class ResourcePool(ManagedObject):
+ """Resource Pool class."""
+
+ def __init__(self):
+ super(ResourcePool, self).__init__("ResourcePool")
+ self.set("name", "ResPool")
+
+
+class Datastore(ManagedObject):
+ """Datastore class."""
+
+ def __init__(self):
+ super(Datastore, self).__init__("Datastore")
+ self.set("summary.type", "VMFS")
+ self.set("summary.name", "fake-ds")
+
+
+class HostNetworkSystem(ManagedObject):
+ """HostNetworkSystem class."""
+
+ def __init__(self):
+ super(HostNetworkSystem, self).__init__("HostNetworkSystem")
+ self.set("name", "networkSystem")
+
+ pnic_do = DataObject()
+ pnic_do.device = "vmnic0"
+
+ net_info_pnic = DataObject()
+ net_info_pnic.PhysicalNic = [pnic_do]
+
+ self.set("networkInfo.pnic", net_info_pnic)
+
+
+class HostSystem(ManagedObject):
+ """Host System class."""
+
+ def __init__(self):
+ super(HostSystem, self).__init__("HostSystem")
+ self.set("name", "ha-host")
+ if _db_content.get("HostNetworkSystem", None) is None:
+ create_host_network_system()
+ host_net_key = _db_content["HostNetworkSystem"].keys()[0]
+ host_net_sys = _db_content["HostNetworkSystem"][host_net_key].obj
+ self.set("configManager.networkSystem", host_net_sys)
+
+ if _db_content.get("Network", None) is None:
+ create_network()
+ net_ref = _db_content["Network"][_db_content["Network"].keys()[0]].obj
+ network_do = DataObject()
+ network_do.ManagedObjectReference = [net_ref]
+ self.set("network", network_do)
+
+ vswitch_do = DataObject()
+ vswitch_do.pnic = ["vmnic0"]
+ vswitch_do.name = "vSwitch0"
+ vswitch_do.portgroup = ["PortGroup-vmnet0"]
+
+ net_swicth = DataObject()
+ net_swicth.HostVirtualSwitch = [vswitch_do]
+ self.set("config.network.vswitch", net_swicth)
+
+ host_pg_do = DataObject()
+ host_pg_do.key = "PortGroup-vmnet0"
+
+ pg_spec = DataObject()
+ pg_spec.vlanId = 0
+ pg_spec.name = "vmnet0"
+
+ host_pg_do.spec = pg_spec
+
+ host_pg = DataObject()
+ host_pg.HostPortGroup = [host_pg_do]
+ self.set("config.network.portgroup", host_pg)
+
+ def _add_port_group(self, spec):
+ """Adds a port group to the host system object in the db."""
+ pg_name = spec.name
+ vswitch_name = spec.vswitchName
+ vlanid = spec.vlanId
+
+ vswitch_do = DataObject()
+ vswitch_do.pnic = ["vmnic0"]
+ vswitch_do.name = vswitch_name
+ vswitch_do.portgroup = ["PortGroup-%s" % pg_name]
+
+ vswitches = self.get("config.network.vswitch").HostVirtualSwitch
+ vswitches.append(vswitch_do)
+
+ host_pg_do = DataObject()
+ host_pg_do.key = "PortGroup-%s" % pg_name
+
+ pg_spec = DataObject()
+ pg_spec.vlanId = vlanid
+ pg_spec.name = pg_name
+
+ host_pg_do.spec = pg_spec
+ host_pgrps = self.get("config.network.portgroup").HostPortGroup
+ host_pgrps.append(host_pg_do)
+
+
+class Datacenter(ManagedObject):
+ """Datacenter class."""
+
+ def __init__(self):
+ super(Datacenter, self).__init__("Datacenter")
+ self.set("name", "ha-datacenter")
+ self.set("vmFolder", "vm_folder_ref")
+ if _db_content.get("Network", None) is None:
+ create_network()
+ net_ref = _db_content["Network"][_db_content["Network"].keys()[0]].obj
+ network_do = DataObject()
+ network_do.ManagedObjectReference = [net_ref]
+ self.set("network", network_do)
+
+
+class Task(ManagedObject):
+ """Task class."""
+
+ def __init__(self, task_name, state="running"):
+ super(Task, self).__init__("Task")
+ info = DataObject
+ info.name = task_name
+ info.state = state
+ self.set("info", info)
+
+
+def create_host_network_system():
+ host_net_system = HostNetworkSystem()
+ _create_object("HostNetworkSystem", host_net_system)
+
+
+def create_host():
+ host_system = HostSystem()
+ _create_object('HostSystem', host_system)
+
+
+def create_datacenter():
+ data_center = Datacenter()
+ _create_object('Datacenter', data_center)
+
+
+def create_datastore():
+ data_store = Datastore()
+ _create_object('Datastore', data_store)
+
+
+def create_res_pool():
+ res_pool = ResourcePool()
+ _create_object('ResourcePool', res_pool)
+
+
+def create_network():
+ network = Network()
+ _create_object('Network', network)
+
+
+def create_task(task_name, state="running"):
+ task = Task(task_name, state)
+ _create_object("Task", task)
+ return task
+
+
+def _add_file(file_path):
+ """Adds a file reference to the db."""
+ _db_content["files"].append(file_path)
+
+
+def _remove_file(file_path):
+ """Removes a file reference from the db."""
+ if _db_content.get("files") is None:
+ raise exception.NotFound(_("No files have been added yet"))
+ # Check if the remove is for a single file object or for a folder
+ if file_path.find(".vmdk") != -1:
+ if file_path not in _db_content.get("files"):
+ raise exception.NotFound(_("File- '%s' is not there in the "
+ "datastore") % file_path)
+ _db_content.get("files").remove(file_path)
+ else:
+ # Removes the files in the folder and the folder too from the db
+ for file in _db_content.get("files"):
+ if file.find(file_path) != -1:
+ lst_files = _db_content.get("files")
+ if lst_files and lst_files.count(file):
+ lst_files.remove(file)
+
+
+def fake_fetch_image(image, instance, **kwargs):
+ """Fakes fetch image call. Just adds a reference to the db for the file."""
+ ds_name = kwargs.get("datastore_name")
+ file_path = kwargs.get("file_path")
+ ds_file_path = "[" + ds_name + "] " + file_path
+ _add_file(ds_file_path)
+
+
+def fake_upload_image(image, instance, **kwargs):
+ """Fakes the upload of an image."""
+ pass
+
+
+def fake_get_vmdk_size_and_properties(image_id, instance):
+ """Fakes the file size and properties fetch for the image file."""
+ props = {"vmware_ostype": "otherGuest",
+ "vmware_adaptertype": "lsiLogic"}
+ return _FAKE_FILE_SIZE, props
+
+
+def _get_vm_mdo(vm_ref):
+ """Gets the Virtual Machine with the ref from the db."""
+ if _db_content.get("VirtualMachine", None) is None:
+ raise exception.NotFound(_("There is no VM registered"))
+ if vm_ref not in _db_content.get("VirtualMachine"):
+ raise exception.NotFound(_("Virtual Machine with ref %s is not "
+ "there") % vm_ref)
+ return _db_content.get("VirtualMachine")[vm_ref]
+
+
+class FakeFactory(object):
+ """Fake factory class for the suds client."""
+
+ def create(self, obj_name):
+ """Creates a namespace object."""
+ return DataObject()
+
+
+class FakeVim(object):
+ """Fake VIM Class."""
+
+ def __init__(self, protocol="https", host="localhost", trace=None):
+ """
+ Initializes the suds client object, sets the service content
+ contents and the cookies for the session.
+ """
+ self._session = None
+ self.client = DataObject()
+ self.client.factory = FakeFactory()
+
+ transport = DataObject()
+ transport.cookiejar = "Fake-CookieJar"
+ options = DataObject()
+ options.transport = transport
+
+ self.client.options = options
+
+ service_content = self.client.factory.create('ns0:ServiceContent')
+ service_content.propertyCollector = "PropCollector"
+ service_content.virtualDiskManager = "VirtualDiskManager"
+ service_content.fileManager = "FileManager"
+ service_content.rootFolder = "RootFolder"
+ service_content.sessionManager = "SessionManager"
+ self._service_content = service_content
+
+ def get_service_content(self):
+ return self._service_content
+
+ def __repr__(self):
+ return "Fake VIM Object"
+
+ def __str__(self):
+ return "Fake VIM Object"
+
+ def _login(self):
+ """Logs in and sets the session object in the db."""
+ self._session = str(uuid.uuid4())
+ session = DataObject()
+ session.key = self._session
+ _db_content['session'][self._session] = session
+ return session
+
+ def _logout(self):
+ """Logs out and remove the session object ref from the db."""
+ s = self._session
+ self._session = None
+ if s not in _db_content['session']:
+ raise exception.Error(
+ _("Logging out a session that is invalid or already logged "
+ "out: %s") % s)
+ del _db_content['session'][s]
+
+ def _terminate_session(self, *args, **kwargs):
+ """Terminates a session."""
+ s = kwargs.get("sessionId")[0]
+ if s not in _db_content['session']:
+ return
+ del _db_content['session'][s]
+
+ def _check_session(self):
+ """Checks if the session is active."""
+ if (self._session is None or self._session not in
+ _db_content['session']):
+ LOG.debug(_("Session is faulty"))
+ raise error_util.VimFaultException(
+ [error_util.FAULT_NOT_AUTHENTICATED],
+ _("Session Invalid"))
+
+ def _create_vm(self, method, *args, **kwargs):
+ """Creates and registers a VM object with the Host System."""
+ config_spec = kwargs.get("config")
+ ds = _db_content["Datastore"][_db_content["Datastore"].keys()[0]]
+ vm_dict = {"name": config_spec.name,
+ "ds": ds,
+ "powerstate": "poweredOff",
+ "vmPathName": config_spec.files.vmPathName,
+ "numCpu": config_spec.numCPUs,
+ "mem": config_spec.memoryMB}
+ virtual_machine = VirtualMachine(**vm_dict)
+ _create_object("VirtualMachine", virtual_machine)
+ task_mdo = create_task(method, "success")
+ return task_mdo.obj
+
+ def _reconfig_vm(self, method, *args, **kwargs):
+ """Reconfigures a VM and sets the properties supplied."""
+ vm_ref = args[0]
+ vm_mdo = _get_vm_mdo(vm_ref)
+ vm_mdo.reconfig(self.client.factory, kwargs.get("spec"))
+ task_mdo = create_task(method, "success")
+ return task_mdo.obj
+
+ def _create_copy_disk(self, method, vmdk_file_path):
+ """Creates/copies a vmdk file object in the datastore."""
+ # We need to add/create both .vmdk and .-flat.vmdk files
+ flat_vmdk_file_path = \
+ vmdk_file_path.replace(".vmdk", "-flat.vmdk")
+ _add_file(vmdk_file_path)
+ _add_file(flat_vmdk_file_path)
+ task_mdo = create_task(method, "success")
+ return task_mdo.obj
+
+ def _snapshot_vm(self, method):
+ """Snapshots a VM. Here we do nothing for faking sake."""
+ task_mdo = create_task(method, "success")
+ return task_mdo.obj
+
+ def _delete_disk(self, method, *args, **kwargs):
+ """Deletes .vmdk and -flat.vmdk files corresponding to the VM."""
+ vmdk_file_path = kwargs.get("name")
+ flat_vmdk_file_path = \
+ vmdk_file_path.replace(".vmdk", "-flat.vmdk")
+ _remove_file(vmdk_file_path)
+ _remove_file(flat_vmdk_file_path)
+ task_mdo = create_task(method, "success")
+ return task_mdo.obj
+
+ def _delete_file(self, method, *args, **kwargs):
+ """Deletes a file from the datastore."""
+ _remove_file(kwargs.get("name"))
+ task_mdo = create_task(method, "success")
+ return task_mdo.obj
+
+ def _just_return(self):
+ """Fakes a return."""
+ return
+
+ def _unregister_vm(self, method, *args, **kwargs):
+ """Unregisters a VM from the Host System."""
+ vm_ref = args[0]
+ _get_vm_mdo(vm_ref)
+ del _db_content["VirtualMachine"][vm_ref]
+
+ def _search_ds(self, method, *args, **kwargs):
+ """Searches the datastore for a file."""
+ ds_path = kwargs.get("datastorePath")
+ if _db_content.get("files", None) is None:
+ raise exception.NotFound(_("No files have been added yet"))
+ for file in _db_content.get("files"):
+ if file.find(ds_path) != -1:
+ task_mdo = create_task(method, "success")
+ return task_mdo.obj
+ task_mdo = create_task(method, "error")
+ return task_mdo.obj
+
+ def _make_dir(self, method, *args, **kwargs):
+ """Creates a directory in the datastore."""
+ ds_path = kwargs.get("name")
+ if _db_content.get("files", None) is None:
+ raise exception.NotFound(_("No files have been added yet"))
+ _db_content["files"].append(ds_path)
+
+ def _set_power_state(self, method, vm_ref, pwr_state="poweredOn"):
+ """Sets power state for the VM."""
+ if _db_content.get("VirtualMachine", None) is None:
+ raise exception.NotFound(_(" No Virtual Machine has been "
+ "registered yet"))
+ if vm_ref not in _db_content.get("VirtualMachine"):
+ raise exception.NotFound(_("Virtual Machine with ref %s is not "
+ "there") % vm_ref)
+ vm_mdo = _db_content.get("VirtualMachine").get(vm_ref)
+ vm_mdo.set("runtime.powerState", pwr_state)
+ task_mdo = create_task(method, "success")
+ return task_mdo.obj
+
+ def _retrieve_properties(self, method, *args, **kwargs):
+ """Retrieves properties based on the type."""
+ spec_set = kwargs.get("specSet")[0]
+ type = spec_set.propSet[0].type
+ properties = spec_set.propSet[0].pathSet
+ objs = spec_set.objectSet
+ lst_ret_objs = []
+ for obj in objs:
+ try:
+ obj_ref = obj.obj
+ # This means that we are doing a search for the managed
+ # dataobjects of the type in the inventory
+ if obj_ref == "RootFolder":
+ for mdo_ref in _db_content[type]:
+ mdo = _db_content[type][mdo_ref]
+ # Create a temp Managed object which has the same ref
+ # as the parent object and copies just the properties
+ # asked for. We need .obj along with the propSet of
+ # just the properties asked for
+ temp_mdo = ManagedObject(mdo.objName, mdo.obj)
+ for prop in properties:
+ temp_mdo.set(prop, mdo.get(prop))
+ lst_ret_objs.append(temp_mdo)
+ else:
+ if obj_ref in _db_content[type]:
+ mdo = _db_content[type][obj_ref]
+ temp_mdo = ManagedObject(mdo.objName, obj_ref)
+ for prop in properties:
+ temp_mdo.set(prop, mdo.get(prop))
+ lst_ret_objs.append(temp_mdo)
+ except Exception, exc:
+ LOG.exception(exc)
+ continue
+ return lst_ret_objs
+
+ def _add_port_group(self, method, *args, **kwargs):
+ """Adds a port group to the host system."""
+ host_mdo = \
+ _db_content["HostSystem"][_db_content["HostSystem"].keys()[0]]
+ host_mdo._add_port_group(kwargs.get("portgrp"))
+
+ def __getattr__(self, attr_name):
+ if attr_name != "Login":
+ self._check_session()
+ if attr_name == "Login":
+ return lambda *args, **kwargs: self._login()
+ elif attr_name == "Logout":
+ self._logout()
+ elif attr_name == "TerminateSession":
+ return lambda *args, **kwargs: self._terminate_session(
+ *args, **kwargs)
+ elif attr_name == "CreateVM_Task":
+ return lambda *args, **kwargs: self._create_vm(attr_name,
+ *args, **kwargs)
+ elif attr_name == "ReconfigVM_Task":
+ return lambda *args, **kwargs: self._reconfig_vm(attr_name,
+ *args, **kwargs)
+ elif attr_name == "CreateVirtualDisk_Task":
+ return lambda *args, **kwargs: self._create_copy_disk(attr_name,
+ kwargs.get("name"))
+ elif attr_name == "DeleteDatastoreFile_Task":
+ return lambda *args, **kwargs: self._delete_file(attr_name,
+ *args, **kwargs)
+ elif attr_name == "PowerOnVM_Task":
+ return lambda *args, **kwargs: self._set_power_state(attr_name,
+ args[0], "poweredOn")
+ elif attr_name == "PowerOffVM_Task":
+ return lambda *args, **kwargs: self._set_power_state(attr_name,
+ args[0], "poweredOff")
+ elif attr_name == "RebootGuest":
+ return lambda *args, **kwargs: self._just_return()
+ elif attr_name == "ResetVM_Task":
+ return lambda *args, **kwargs: self._set_power_state(attr_name,
+ args[0], "poweredOn")
+ elif attr_name == "SuspendVM_Task":
+ return lambda *args, **kwargs: self._set_power_state(attr_name,
+ args[0], "suspended")
+ elif attr_name == "CreateSnapshot_Task":
+ return lambda *args, **kwargs: self._snapshot_vm(attr_name)
+ elif attr_name == "CopyVirtualDisk_Task":
+ return lambda *args, **kwargs: self._create_copy_disk(attr_name,
+ kwargs.get("destName"))
+ elif attr_name == "DeleteVirtualDisk_Task":
+ return lambda *args, **kwargs: self._delete_disk(attr_name,
+ *args, **kwargs)
+ elif attr_name == "UnregisterVM":
+ return lambda *args, **kwargs: self._unregister_vm(attr_name,
+ *args, **kwargs)
+ elif attr_name == "SearchDatastore_Task":
+ return lambda *args, **kwargs: self._search_ds(attr_name,
+ *args, **kwargs)
+ elif attr_name == "MakeDirectory":
+ return lambda *args, **kwargs: self._make_dir(attr_name,
+ *args, **kwargs)
+ elif attr_name == "RetrieveProperties":
+ return lambda *args, **kwargs: self._retrieve_properties(
+ attr_name, *args, **kwargs)
+ elif attr_name == "AcquireCloneTicket":
+ return lambda *args, **kwargs: self._just_return()
+ elif attr_name == "AddPortGroup":
+ return lambda *args, **kwargs: self._add_port_group(attr_name,
+ *args, **kwargs)
diff --git a/nova/virt/vmwareapi/io_util.py b/nova/virt/vmwareapi/io_util.py new file mode 100644 index 000000000..2ec773b7b --- /dev/null +++ b/nova/virt/vmwareapi/io_util.py @@ -0,0 +1,168 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""
+Utility classes for defining the time saving transfer of data from the reader
+to the write using a LightQueue as a Pipe between the reader and the writer.
+"""
+
+from eventlet import event
+from eventlet import greenthread
+from eventlet.queue import LightQueue
+
+from glance import client
+
+from nova import exception
+from nova import log as logging
+
+LOG = logging.getLogger("nova.virt.vmwareapi.io_util")
+
+IO_THREAD_SLEEP_TIME = .01
+GLANCE_POLL_INTERVAL = 5
+
+
+class ThreadSafePipe(LightQueue):
+ """The pipe to hold the data which the reader writes to and the writer
+ reads from."""
+
+ def __init__(self, maxsize, transfer_size):
+ LightQueue.__init__(self, maxsize)
+ self.transfer_size = transfer_size
+ self.transferred = 0
+
+ def read(self, chunk_size):
+ """Read data from the pipe. Chunksize if ignored for we have ensured
+ that the data chunks written to the pipe by readers is the same as the
+ chunks asked for by the Writer."""
+ if self.transferred < self.transfer_size:
+ data_item = self.get()
+ self.transferred += len(data_item)
+ return data_item
+ else:
+ return ""
+
+ def write(self, data):
+ """Put a data item in the pipe."""
+ self.put(data)
+
+ def close(self):
+ """A place-holder to maintain consistency."""
+ pass
+
+
+class GlanceWriteThread(object):
+ """Ensures that image data is written to in the glance client and that
+ it is in correct ('active')state."""
+
+ def __init__(self, input, glance_client, image_id, image_meta={}):
+ self.input = input
+ self.glance_client = glance_client
+ self.image_id = image_id
+ self.image_meta = image_meta
+ self._running = False
+
+ def start(self):
+ self.done = event.Event()
+
+ def _inner():
+ """Function to do the image data transfer through an update
+ and thereon checks if the state is 'active'."""
+ self.glance_client.update_image(self.image_id,
+ image_meta=self.image_meta,
+ image_data=self.input)
+ self._running = True
+ while self._running:
+ try:
+ image_status = \
+ self.glance_client.get_image_meta(self.image_id).get(
+ "status")
+ if image_status == "active":
+ self.stop()
+ self.done.send(True)
+ # If the state is killed, then raise an exception.
+ elif image_status == "killed":
+ self.stop()
+ exc_msg = _("Glance image %s is in killed state") %\
+ self.image_id
+ LOG.exception(exc_msg)
+ self.done.send_exception(exception.Error(exc_msg))
+ elif image_status in ["saving", "queued"]:
+ greenthread.sleep(GLANCE_POLL_INTERVAL)
+ else:
+ self.stop()
+ exc_msg = _("Glance image "
+ "%(image_id)s is in unknown state "
+ "- %(state)s") % {
+ "image_id": self.image_id,
+ "state": image_status}
+ LOG.exception(exc_msg)
+ self.done.send_exception(exception.Error(exc_msg))
+ except Exception, exc:
+ self.stop()
+ self.done.send_exception(exc)
+
+ greenthread.spawn(_inner)
+ return self.done
+
+ def stop(self):
+ self._running = False
+
+ def wait(self):
+ return self.done.wait()
+
+ def close(self):
+ pass
+
+
+class IOThread(object):
+ """Class that reads chunks from the input file and writes them to the
+ output file till the transfer is completely done."""
+
+ def __init__(self, input, output):
+ self.input = input
+ self.output = output
+ self._running = False
+ self.got_exception = False
+
+ def start(self):
+ self.done = event.Event()
+
+ def _inner():
+ """Read data from the input and write the same to the output
+ until the transfer completes."""
+ self._running = True
+ while self._running:
+ try:
+ data = self.input.read(None)
+ if not data:
+ self.stop()
+ self.done.send(True)
+ self.output.write(data)
+ greenthread.sleep(IO_THREAD_SLEEP_TIME)
+ except Exception, exc:
+ self.stop()
+ LOG.exception(exc)
+ self.done.send_exception(exc)
+
+ greenthread.spawn(_inner)
+ return self.done
+
+ def stop(self):
+ self._running = False
+
+ def wait(self):
+ return self.done.wait()
diff --git a/nova/virt/vmwareapi/network_utils.py b/nova/virt/vmwareapi/network_utils.py new file mode 100644 index 000000000..e77842535 --- /dev/null +++ b/nova/virt/vmwareapi/network_utils.py @@ -0,0 +1,149 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""
+Utility functions for ESX Networking.
+"""
+
+from nova import exception
+from nova import log as logging
+from nova.virt.vmwareapi import error_util
+from nova.virt.vmwareapi import vim_util
+from nova.virt.vmwareapi import vm_util
+
+LOG = logging.getLogger("nova.virt.vmwareapi.network_utils")
+
+
+def get_network_with_the_name(session, network_name="vmnet0"):
+ """
+ Gets reference to the network whose name is passed as the
+ argument.
+ """
+ hostsystems = session._call_method(vim_util, "get_objects",
+ "HostSystem", ["network"])
+ vm_networks_ret = hostsystems[0].propSet[0].val
+ # Meaning there are no networks on the host. suds responds with a ""
+ # in the parent property field rather than a [] in the
+ # ManagedObjectRefernce property field of the parent
+ if not vm_networks_ret:
+ return None
+ vm_networks = vm_networks_ret.ManagedObjectReference
+ networks = session._call_method(vim_util,
+ "get_properties_for_a_collection_of_objects",
+ "Network", vm_networks, ["summary.name"])
+ for network in networks:
+ if network.propSet[0].val == network_name:
+ return network.obj
+ return None
+
+
+def get_vswitch_for_vlan_interface(session, vlan_interface):
+ """
+ Gets the vswitch associated with the physical network adapter
+ with the name supplied.
+ """
+ # Get the list of vSwicthes on the Host System
+ host_mor = session._call_method(vim_util, "get_objects",
+ "HostSystem")[0].obj
+ vswitches_ret = session._call_method(vim_util,
+ "get_dynamic_property", host_mor,
+ "HostSystem", "config.network.vswitch")
+ # Meaning there are no vSwitches on the host. Shouldn't be the case,
+ # but just doing code check
+ if not vswitches_ret:
+ return
+ vswitches = vswitches_ret.HostVirtualSwitch
+ # Get the vSwitch associated with the network adapter
+ for elem in vswitches:
+ try:
+ for nic_elem in elem.pnic:
+ if str(nic_elem).split('-')[-1].find(vlan_interface) != -1:
+ return elem.name
+ # Catching Attribute error as a vSwitch may not be associated with a
+ # physical NIC.
+ except AttributeError:
+ pass
+
+
+def check_if_vlan_interface_exists(session, vlan_interface):
+ """Checks if the vlan_inteface exists on the esx host."""
+ host_net_system_mor = session._call_method(vim_util, "get_objects",
+ "HostSystem", ["configManager.networkSystem"])[0].propSet[0].val
+ physical_nics_ret = session._call_method(vim_util,
+ "get_dynamic_property", host_net_system_mor,
+ "HostNetworkSystem", "networkInfo.pnic")
+ # Meaning there are no physical nics on the host
+ if not physical_nics_ret:
+ return False
+ physical_nics = physical_nics_ret.PhysicalNic
+ for pnic in physical_nics:
+ if vlan_interface == pnic.device:
+ return True
+ return False
+
+
+def get_vlanid_and_vswitch_for_portgroup(session, pg_name):
+ """Get the vlan id and vswicth associated with the port group."""
+ host_mor = session._call_method(vim_util, "get_objects",
+ "HostSystem")[0].obj
+ port_grps_on_host_ret = session._call_method(vim_util,
+ "get_dynamic_property", host_mor,
+ "HostSystem", "config.network.portgroup")
+ if not port_grps_on_host_ret:
+ excep = ("ESX SOAP server returned an empty port group "
+ "for the host system in its response")
+ LOG.exception(excep)
+ raise exception.Error(_(excep))
+ port_grps_on_host = port_grps_on_host_ret.HostPortGroup
+ for p_gp in port_grps_on_host:
+ if p_gp.spec.name == pg_name:
+ p_grp_vswitch_name = p_gp.vswitch.split("-")[-1]
+ return p_gp.spec.vlanId, p_grp_vswitch_name
+
+
+def create_port_group(session, pg_name, vswitch_name, vlan_id=0):
+ """
+ Creates a port group on the host system with the vlan tags
+ supplied. VLAN id 0 means no vlan id association.
+ """
+ client_factory = session._get_vim().client.factory
+ add_prt_grp_spec = vm_util.get_add_vswitch_port_group_spec(
+ client_factory,
+ vswitch_name,
+ pg_name,
+ vlan_id)
+ host_mor = session._call_method(vim_util, "get_objects",
+ "HostSystem")[0].obj
+ network_system_mor = session._call_method(vim_util,
+ "get_dynamic_property", host_mor,
+ "HostSystem", "configManager.networkSystem")
+ LOG.debug(_("Creating Port Group with name %s on "
+ "the ESX host") % pg_name)
+ try:
+ session._call_method(session._get_vim(),
+ "AddPortGroup", network_system_mor,
+ portgrp=add_prt_grp_spec)
+ except error_util.VimFaultException, exc:
+ # There can be a race condition when two instances try
+ # adding port groups at the same time. One succeeds, then
+ # the other one will get an exception. Since we are
+ # concerned with the port group being created, which is done
+ # by the other call, we can ignore the exception.
+ if error_util.FAULT_ALREADY_EXISTS not in exc.fault_list:
+ raise exception.Error(exc)
+ LOG.debug(_("Created Port Group with name %s on "
+ "the ESX host") % pg_name)
diff --git a/nova/virt/vmwareapi/read_write_util.py b/nova/virt/vmwareapi/read_write_util.py new file mode 100644 index 000000000..84f4942eb --- /dev/null +++ b/nova/virt/vmwareapi/read_write_util.py @@ -0,0 +1,182 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""Classes to handle image files
+
+Collection of classes to handle image upload/download to/from Image service
+(like Glance image storage and retrieval service) from/to ESX/ESXi server.
+
+"""
+
+import httplib
+import urllib
+import urllib2
+import urlparse
+
+from eventlet import event
+from eventlet import greenthread
+
+from glance import client
+
+from nova import flags
+from nova import log as logging
+
+LOG = logging.getLogger("nova.virt.vmwareapi.read_write_util")
+
+FLAGS = flags.FLAGS
+
+USER_AGENT = "OpenStack-ESX-Adapter"
+
+try:
+ READ_CHUNKSIZE = client.BaseClient.CHUNKSIZE
+except AttributeError:
+ READ_CHUNKSIZE = 65536
+
+
+class GlanceFileRead(object):
+ """Glance file read handler class."""
+
+ def __init__(self, glance_read_iter):
+ self.glance_read_iter = glance_read_iter
+ self.iter = self.get_next()
+
+ def read(self, chunk_size):
+ """Read an item from the queue. The chunk size is ignored for the
+ Client ImageBodyIterator uses its own CHUNKSIZE."""
+ try:
+ return self.iter.next()
+ except StopIteration:
+ return ""
+
+ def get_next(self):
+ """Get the next item from the image iterator."""
+ for data in self.glance_read_iter:
+ yield data
+
+ def close(self):
+ """A dummy close just to maintain consistency."""
+ pass
+
+
+class VMwareHTTPFile(object):
+ """Base class for HTTP file."""
+
+ def __init__(self, file_handle):
+ self.eof = False
+ self.file_handle = file_handle
+
+ def set_eof(self, eof):
+ """Set the end of file marker."""
+ self.eof = eof
+
+ def get_eof(self):
+ """Check if the end of file has been reached."""
+ return self.eof
+
+ def close(self):
+ """Close the file handle."""
+ try:
+ self.file_handle.close()
+ except Exception, exc:
+ LOG.exception(exc)
+
+ def __del__(self):
+ """Close the file handle on garbage collection."""
+ self.close()
+
+ def _build_vim_cookie_headers(self, vim_cookies):
+ """Build ESX host session cookie headers."""
+ cookie_header = ""
+ for vim_cookie in vim_cookies:
+ cookie_header = vim_cookie.name + "=" + vim_cookie.value
+ break
+ return cookie_header
+
+ def write(self, data):
+ """Write data to the file."""
+ raise NotImplementedError
+
+ def read(self, chunk_size):
+ """Read a chunk of data."""
+ raise NotImplementedError
+
+ def get_size(self):
+ """Get size of the file to be read."""
+ raise NotImplementedError
+
+
+class VMWareHTTPWriteFile(VMwareHTTPFile):
+ """VMWare file write handler class."""
+
+ def __init__(self, host, data_center_name, datastore_name, cookies,
+ file_path, file_size, scheme="https"):
+ base_url = "%s://%s/folder/%s" % (scheme, host, file_path)
+ param_list = {"dcPath": data_center_name, "dsName": datastore_name}
+ base_url = base_url + "?" + urllib.urlencode(param_list)
+ (scheme, netloc, path, params, query, fragment) = \
+ urlparse.urlparse(base_url)
+ if scheme == "http":
+ conn = httplib.HTTPConnection(netloc)
+ elif scheme == "https":
+ conn = httplib.HTTPSConnection(netloc)
+ conn.putrequest("PUT", path + "?" + query)
+ conn.putheader("User-Agent", USER_AGENT)
+ conn.putheader("Content-Length", file_size)
+ conn.putheader("Cookie", self._build_vim_cookie_headers(cookies))
+ conn.endheaders()
+ self.conn = conn
+ VMwareHTTPFile.__init__(self, conn)
+
+ def write(self, data):
+ """Write to the file."""
+ self.file_handle.send(data)
+
+ def close(self):
+ """Get the response and close the connection."""
+ try:
+ self.conn.getresponse()
+ except Exception, excep:
+ LOG.debug(_("Exception during HTTP connection close in "
+ "VMWareHTTpWrite. Exception is %s") % excep)
+ super(VMWareHTTPWriteFile, self).close()
+
+
+class VmWareHTTPReadFile(VMwareHTTPFile):
+ """VMWare file read handler class."""
+
+ def __init__(self, host, data_center_name, datastore_name, cookies,
+ file_path, scheme="https"):
+ base_url = "%s://%s/folder/%s" % (scheme, host,
+ urllib.pathname2url(file_path))
+ param_list = {"dcPath": data_center_name, "dsName": datastore_name}
+ base_url = base_url + "?" + urllib.urlencode(param_list)
+ headers = {'User-Agent': USER_AGENT,
+ 'Cookie': self._build_vim_cookie_headers(cookies)}
+ request = urllib2.Request(base_url, None, headers)
+ conn = urllib2.urlopen(request)
+ VMwareHTTPFile.__init__(self, conn)
+
+ def read(self, chunk_size):
+ """Read a chunk of data."""
+ # We are ignoring the chunk size passed for we want the pipe to hold
+ # data items of the chunk-size that Glance Client uses for read
+ # while writing.
+ return self.file_handle.read(READ_CHUNKSIZE)
+
+ def get_size(self):
+ """Get size of the file to be read."""
+ return self.file_handle.headers.get("Content-Length", -1)
diff --git a/nova/virt/vmwareapi/vim.py b/nova/virt/vmwareapi/vim.py new file mode 100644 index 000000000..159e16a80 --- /dev/null +++ b/nova/virt/vmwareapi/vim.py @@ -0,0 +1,180 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""
+Classes for making VMware VI SOAP calls.
+"""
+
+import httplib
+
+try:
+ import suds
+except ImportError:
+ suds = None
+
+from nova import flags
+from nova.virt.vmwareapi import error_util
+
+RESP_NOT_XML_ERROR = 'Response is "text/html", not "text/xml"'
+CONN_ABORT_ERROR = 'Software caused connection abort'
+ADDRESS_IN_USE_ERROR = 'Address already in use'
+
+FLAGS = flags.FLAGS
+flags.DEFINE_string('vmwareapi_wsdl_loc',
+ None,
+ 'VIM Service WSDL Location'
+ 'e.g http://<server>/vimService.wsdl'
+ 'Due to a bug in vSphere ESX 4.1 default wsdl'
+ 'Refer readme-vmware to setup')
+
+
+if suds:
+ class VIMMessagePlugin(suds.plugin.MessagePlugin):
+
+ def addAttributeForValue(self, node):
+ # suds does not handle AnyType properly.
+ # VI SDK requires type attribute to be set when AnyType is used
+ if node.name == 'value':
+ node.set('xsi:type', 'xsd:string')
+
+ def marshalled(self, context):
+ """suds will send the specified soap envelope.
+ Provides the plugin with the opportunity to prune empty
+ nodes and fixup nodes before sending it to the server.
+ """
+ # suds builds the entire request object based on the wsdl schema.
+ # VI SDK throws server errors if optional SOAP nodes are sent
+ # without values, e.g. <test/> as opposed to <test>test</test>
+ context.envelope.prune()
+ context.envelope.walk(self.addAttributeForValue)
+
+
+class Vim:
+ """The VIM Object."""
+
+ def __init__(self,
+ protocol="https",
+ host="localhost"):
+ """
+ Creates the necessary Communication interfaces and gets the
+ ServiceContent for initiating SOAP transactions.
+
+ protocol: http or https
+ host : ESX IPAddress[:port] or ESX Hostname[:port]
+ """
+ if not suds:
+ raise Exception(_("Unable to import suds."))
+
+ self._protocol = protocol
+ self._host_name = host
+ wsdl_url = FLAGS.vmwareapi_wsdl_loc
+ if wsdl_url is None:
+ raise Exception(_("Must specify vmwareapi_wsdl_loc"))
+ # TODO(sateesh): Use this when VMware fixes their faulty wsdl
+ #wsdl_url = '%s://%s/sdk/vimService.wsdl' % (self._protocol,
+ # self._host_name)
+ url = '%s://%s/sdk' % (self._protocol, self._host_name)
+ self.client = suds.client.Client(wsdl_url, location=url,
+ plugins=[VIMMessagePlugin()])
+ self._service_content = \
+ self.RetrieveServiceContent("ServiceInstance")
+
+ def get_service_content(self):
+ """Gets the service content object."""
+ return self._service_content
+
+ def __getattr__(self, attr_name):
+ """Makes the API calls and gets the result."""
+ try:
+ return object.__getattr__(self, attr_name)
+ except AttributeError:
+
+ def vim_request_handler(managed_object, **kwargs):
+ """
+ Builds the SOAP message and parses the response for fault
+ checking and other errors.
+
+ managed_object : Managed Object Reference or Managed
+ Object Name
+ **kwargs : Keyword arguments of the call
+ """
+ # Dynamic handler for VI SDK Calls
+ try:
+ request_mo = \
+ self._request_managed_object_builder(managed_object)
+ request = getattr(self.client.service, attr_name)
+ response = request(request_mo, **kwargs)
+ # To check for the faults that are part of the message body
+ # and not returned as Fault object response from the ESX
+ # SOAP server
+ if hasattr(error_util.FaultCheckers,
+ attr_name.lower() + "_fault_checker"):
+ fault_checker = getattr(error_util.FaultCheckers,
+ attr_name.lower() + "_fault_checker")
+ fault_checker(response)
+ return response
+ # Catch the VimFaultException that is raised by the fault
+ # check of the SOAP response
+ except error_util.VimFaultException, excep:
+ raise
+ except suds.WebFault, excep:
+ doc = excep.document
+ detail = doc.childAtPath("/Envelope/Body/Fault/detail")
+ fault_list = []
+ for child in detail.getChildren():
+ fault_list.append(child.get("type"))
+ raise error_util.VimFaultException(fault_list, excep)
+ except AttributeError, excep:
+ raise error_util.VimAttributeError(_("No such SOAP method "
+ "'%s' provided by VI SDK") % (attr_name), excep)
+ except (httplib.CannotSendRequest,
+ httplib.ResponseNotReady,
+ httplib.CannotSendHeader), excep:
+ raise error_util.SessionOverLoadException(_("httplib "
+ "error in %s: ") % (attr_name), excep)
+ except Exception, excep:
+ # Socket errors which need special handling for they
+ # might be caused by ESX API call overload
+ if (str(excep).find(ADDRESS_IN_USE_ERROR) != -1 or
+ str(excep).find(CONN_ABORT_ERROR)) != -1:
+ raise error_util.SessionOverLoadException(_("Socket "
+ "error in %s: ") % (attr_name), excep)
+ # Type error that needs special handling for it might be
+ # caused by ESX host API call overload
+ elif str(excep).find(RESP_NOT_XML_ERROR) != -1:
+ raise error_util.SessionOverLoadException(_("Type "
+ "error in %s: ") % (attr_name), excep)
+ else:
+ raise error_util.VimException(
+ _("Exception in %s ") % (attr_name), excep)
+ return vim_request_handler
+
+ def _request_managed_object_builder(self, managed_object):
+ """Builds the request managed object."""
+ # Request Managed Object Builder
+ if type(managed_object) == type(""):
+ mo = suds.sudsobject.Property(managed_object)
+ mo._type = managed_object
+ else:
+ mo = managed_object
+ return mo
+
+ def __repr__(self):
+ return "VIM Object"
+
+ def __str__(self):
+ return "VIM Object"
diff --git a/nova/virt/vmwareapi/vim_util.py b/nova/virt/vmwareapi/vim_util.py new file mode 100644 index 000000000..11214231c --- /dev/null +++ b/nova/virt/vmwareapi/vim_util.py @@ -0,0 +1,217 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""
+The VMware API utility module.
+"""
+
+
+def build_selection_spec(client_factory, name):
+ """Builds the selection spec."""
+ sel_spec = client_factory.create('ns0:SelectionSpec')
+ sel_spec.name = name
+ return sel_spec
+
+
+def build_traversal_spec(client_factory, name, spec_type, path, skip,
+ select_set):
+ """Builds the traversal spec object."""
+ traversal_spec = client_factory.create('ns0:TraversalSpec')
+ traversal_spec.name = name
+ traversal_spec.type = spec_type
+ traversal_spec.path = path
+ traversal_spec.skip = skip
+ traversal_spec.selectSet = select_set
+ return traversal_spec
+
+
+def build_recursive_traversal_spec(client_factory):
+ """
+ Builds the Recursive Traversal Spec to traverse the object managed
+ object hierarchy.
+ """
+ visit_folders_select_spec = build_selection_spec(client_factory,
+ "visitFolders")
+ # For getting to hostFolder from datacenter
+ dc_to_hf = build_traversal_spec(client_factory, "dc_to_hf", "Datacenter",
+ "hostFolder", False,
+ [visit_folders_select_spec])
+ # For getting to vmFolder from datacenter
+ dc_to_vmf = build_traversal_spec(client_factory, "dc_to_vmf", "Datacenter",
+ "vmFolder", False,
+ [visit_folders_select_spec])
+ # For getting Host System to virtual machine
+ h_to_vm = build_traversal_spec(client_factory, "h_to_vm", "HostSystem",
+ "vm", False,
+ [visit_folders_select_spec])
+
+ # For getting to Host System from Compute Resource
+ cr_to_h = build_traversal_spec(client_factory, "cr_to_h",
+ "ComputeResource", "host", False, [])
+
+ # For getting to datastore from Compute Resource
+ cr_to_ds = build_traversal_spec(client_factory, "cr_to_ds",
+ "ComputeResource", "datastore", False, [])
+
+ rp_to_rp_select_spec = build_selection_spec(client_factory, "rp_to_rp")
+ rp_to_vm_select_spec = build_selection_spec(client_factory, "rp_to_vm")
+ # For getting to resource pool from Compute Resource
+ cr_to_rp = build_traversal_spec(client_factory, "cr_to_rp",
+ "ComputeResource", "resourcePool", False,
+ [rp_to_rp_select_spec, rp_to_vm_select_spec])
+
+ # For getting to child res pool from the parent res pool
+ rp_to_rp = build_traversal_spec(client_factory, "rp_to_rp", "ResourcePool",
+ "resourcePool", False,
+ [rp_to_rp_select_spec, rp_to_vm_select_spec])
+
+ # For getting to Virtual Machine from the Resource Pool
+ rp_to_vm = build_traversal_spec(client_factory, "rp_to_vm", "ResourcePool",
+ "vm", False,
+ [rp_to_rp_select_spec, rp_to_vm_select_spec])
+
+ # Get the assorted traversal spec which takes care of the objects to
+ # be searched for from the root folder
+ traversal_spec = build_traversal_spec(client_factory, "visitFolders",
+ "Folder", "childEntity", False,
+ [visit_folders_select_spec, dc_to_hf,
+ dc_to_vmf, cr_to_ds, cr_to_h, cr_to_rp,
+ rp_to_rp, h_to_vm, rp_to_vm])
+ return traversal_spec
+
+
+def build_property_spec(client_factory, type="VirtualMachine",
+ properties_to_collect=["name"],
+ all_properties=False):
+ """Builds the Property Spec."""
+ property_spec = client_factory.create('ns0:PropertySpec')
+ property_spec.all = all_properties
+ property_spec.pathSet = properties_to_collect
+ property_spec.type = type
+ return property_spec
+
+
+def build_object_spec(client_factory, root_folder, traversal_specs):
+ """Builds the object Spec."""
+ object_spec = client_factory.create('ns0:ObjectSpec')
+ object_spec.obj = root_folder
+ object_spec.skip = False
+ object_spec.selectSet = traversal_specs
+ return object_spec
+
+
+def build_property_filter_spec(client_factory, property_specs, object_specs):
+ """Builds the Property Filter Spec."""
+ property_filter_spec = client_factory.create('ns0:PropertyFilterSpec')
+ property_filter_spec.propSet = property_specs
+ property_filter_spec.objectSet = object_specs
+ return property_filter_spec
+
+
+def get_object_properties(vim, collector, mobj, type, properties):
+ """Gets the properties of the Managed object specified."""
+ client_factory = vim.client.factory
+ if mobj is None:
+ return None
+ usecoll = collector
+ if usecoll is None:
+ usecoll = vim.get_service_content().propertyCollector
+ property_filter_spec = client_factory.create('ns0:PropertyFilterSpec')
+ property_spec = client_factory.create('ns0:PropertySpec')
+ property_spec.all = (properties is None or len(properties) == 0)
+ property_spec.pathSet = properties
+ property_spec.type = type
+ object_spec = client_factory.create('ns0:ObjectSpec')
+ object_spec.obj = mobj
+ object_spec.skip = False
+ property_filter_spec.propSet = [property_spec]
+ property_filter_spec.objectSet = [object_spec]
+ return vim.RetrieveProperties(usecoll, specSet=[property_filter_spec])
+
+
+def get_dynamic_property(vim, mobj, type, property_name):
+ """Gets a particular property of the Managed Object."""
+ obj_content = \
+ get_object_properties(vim, None, mobj, type, [property_name])
+ property_value = None
+ if obj_content:
+ dynamic_property = obj_content[0].propSet
+ if dynamic_property:
+ property_value = dynamic_property[0].val
+ return property_value
+
+
+def get_objects(vim, type, properties_to_collect=["name"], all=False):
+ """Gets the list of objects of the type specified."""
+ client_factory = vim.client.factory
+ object_spec = build_object_spec(client_factory,
+ vim.get_service_content().rootFolder,
+ [build_recursive_traversal_spec(client_factory)])
+ property_spec = build_property_spec(client_factory, type=type,
+ properties_to_collect=properties_to_collect,
+ all_properties=all)
+ property_filter_spec = build_property_filter_spec(client_factory,
+ [property_spec],
+ [object_spec])
+ return vim.RetrieveProperties(vim.get_service_content().propertyCollector,
+ specSet=[property_filter_spec])
+
+
+def get_prop_spec(client_factory, spec_type, properties):
+ """Builds the Property Spec Object."""
+ prop_spec = client_factory.create('ns0:PropertySpec')
+ prop_spec.type = spec_type
+ prop_spec.pathSet = properties
+ return prop_spec
+
+
+def get_obj_spec(client_factory, obj, select_set=None):
+ """Builds the Object Spec object."""
+ obj_spec = client_factory.create('ns0:ObjectSpec')
+ obj_spec.obj = obj
+ obj_spec.skip = False
+ if select_set is not None:
+ obj_spec.selectSet = select_set
+ return obj_spec
+
+
+def get_prop_filter_spec(client_factory, obj_spec, prop_spec):
+ """Builds the Property Filter Spec Object."""
+ prop_filter_spec = \
+ client_factory.create('ns0:PropertyFilterSpec')
+ prop_filter_spec.propSet = prop_spec
+ prop_filter_spec.objectSet = obj_spec
+ return prop_filter_spec
+
+
+def get_properties_for_a_collection_of_objects(vim, type,
+ obj_list, properties):
+ """
+ Gets the list of properties for the collection of
+ objects of the type specified.
+ """
+ client_factory = vim.client.factory
+ if len(obj_list) == 0:
+ return []
+ prop_spec = get_prop_spec(client_factory, type, properties)
+ lst_obj_specs = []
+ for obj in obj_list:
+ lst_obj_specs.append(get_obj_spec(client_factory, obj))
+ prop_filter_spec = get_prop_filter_spec(client_factory,
+ lst_obj_specs, [prop_spec])
+ return vim.RetrieveProperties(vim.get_service_content().propertyCollector,
+ specSet=[prop_filter_spec])
diff --git a/nova/virt/vmwareapi/vm_util.py b/nova/virt/vmwareapi/vm_util.py new file mode 100644 index 000000000..a2fa7600c --- /dev/null +++ b/nova/virt/vmwareapi/vm_util.py @@ -0,0 +1,306 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+"""
+The VMware API VM utility module to build SOAP object specs.
+"""
+
+
+def build_datastore_path(datastore_name, path):
+ """Build the datastore compliant path."""
+ return "[%s] %s" % (datastore_name, path)
+
+
+def split_datastore_path(datastore_path):
+ """
+ Split the VMWare style datastore path to get the Datastore
+ name and the entity path.
+ """
+ spl = datastore_path.split('[', 1)[1].split(']', 1)
+ path = ""
+ if len(spl) == 1:
+ datastore_url = spl[0]
+ else:
+ datastore_url, path = spl
+ return datastore_url, path.strip()
+
+
+def get_vm_create_spec(client_factory, instance, data_store_name,
+ network_name="vmnet0",
+ os_type="otherGuest"):
+ """Builds the VM Create spec."""
+ config_spec = client_factory.create('ns0:VirtualMachineConfigSpec')
+ config_spec.name = instance.name
+ config_spec.guestId = os_type
+
+ vm_file_info = client_factory.create('ns0:VirtualMachineFileInfo')
+ vm_file_info.vmPathName = "[" + data_store_name + "]"
+ config_spec.files = vm_file_info
+
+ tools_info = client_factory.create('ns0:ToolsConfigInfo')
+ tools_info.afterPowerOn = True
+ tools_info.afterResume = True
+ tools_info.beforeGuestStandby = True
+ tools_info.beforeGuestShutdown = True
+ tools_info.beforeGuestReboot = True
+
+ config_spec.tools = tools_info
+ config_spec.numCPUs = int(instance.vcpus)
+ config_spec.memoryMB = int(instance.memory_mb)
+
+ nic_spec = create_network_spec(client_factory,
+ network_name, instance.mac_address)
+
+ device_config_spec = [nic_spec]
+
+ config_spec.deviceChange = device_config_spec
+ return config_spec
+
+
+def create_controller_spec(client_factory, key):
+ """
+ Builds a Config Spec for the LSI Logic Controller's addition
+ which acts as the controller for the virtual hard disk to be attached
+ to the VM.
+ """
+ # Create a controller for the Virtual Hard Disk
+ virtual_device_config = \
+ client_factory.create('ns0:VirtualDeviceConfigSpec')
+ virtual_device_config.operation = "add"
+ virtual_lsi = \
+ client_factory.create('ns0:VirtualLsiLogicController')
+ virtual_lsi.key = key
+ virtual_lsi.busNumber = 0
+ virtual_lsi.sharedBus = "noSharing"
+ virtual_device_config.device = virtual_lsi
+ return virtual_device_config
+
+
+def create_network_spec(client_factory, network_name, mac_address):
+ """
+ Builds a config spec for the addition of a new network
+ adapter to the VM.
+ """
+ network_spec = \
+ client_factory.create('ns0:VirtualDeviceConfigSpec')
+ network_spec.operation = "add"
+
+ # Get the recommended card type for the VM based on the guest OS of the VM
+ net_device = client_factory.create('ns0:VirtualPCNet32')
+
+ backing = \
+ client_factory.create('ns0:VirtualEthernetCardNetworkBackingInfo')
+ backing.deviceName = network_name
+
+ connectable_spec = \
+ client_factory.create('ns0:VirtualDeviceConnectInfo')
+ connectable_spec.startConnected = True
+ connectable_spec.allowGuestControl = True
+ connectable_spec.connected = True
+
+ net_device.connectable = connectable_spec
+ net_device.backing = backing
+
+ # The Server assigns a Key to the device. Here we pass a -ve temporary key.
+ # -ve because actual keys are +ve numbers and we don't
+ # want a clash with the key that server might associate with the device
+ net_device.key = -47
+ net_device.addressType = "manual"
+ net_device.macAddress = mac_address
+ net_device.wakeOnLanEnabled = True
+
+ network_spec.device = net_device
+ return network_spec
+
+
+def get_vmdk_attach_config_spec(client_factory, disksize, file_path,
+ adapter_type="lsiLogic"):
+ """Builds the vmdk attach config spec."""
+ config_spec = client_factory.create('ns0:VirtualMachineConfigSpec')
+
+ # The controller Key pertains to the Key of the LSI Logic Controller, which
+ # controls this Hard Disk
+ device_config_spec = []
+ # For IDE devices, there are these two default controllers created in the
+ # VM having keys 200 and 201
+ if adapter_type == "ide":
+ controller_key = 200
+ else:
+ controller_key = -101
+ controller_spec = create_controller_spec(client_factory,
+ controller_key)
+ device_config_spec.append(controller_spec)
+ virtual_device_config_spec = create_virtual_disk_spec(client_factory,
+ disksize, controller_key, file_path)
+
+ device_config_spec.append(virtual_device_config_spec)
+
+ config_spec.deviceChange = device_config_spec
+ return config_spec
+
+
+def get_vmdk_file_path_and_adapter_type(client_factory, hardware_devices):
+ """Gets the vmdk file path and the storage adapter type."""
+ if hardware_devices.__class__.__name__ == "ArrayOfVirtualDevice":
+ hardware_devices = hardware_devices.VirtualDevice
+ vmdk_file_path = None
+ vmdk_controler_key = None
+
+ adapter_type_dict = {}
+ for device in hardware_devices:
+ if device.__class__.__name__ == "VirtualDisk" and \
+ device.backing.__class__.__name__ \
+ == "VirtualDiskFlatVer2BackingInfo":
+ vmdk_file_path = device.backing.fileName
+ vmdk_controler_key = device.controllerKey
+ elif device.__class__.__name__ == "VirtualLsiLogicController":
+ adapter_type_dict[device.key] = "lsiLogic"
+ elif device.__class__.__name__ == "VirtualBusLogicController":
+ adapter_type_dict[device.key] = "busLogic"
+ elif device.__class__.__name__ == "VirtualIDEController":
+ adapter_type_dict[device.key] = "ide"
+ elif device.__class__.__name__ == "VirtualLsiLogicSASController":
+ adapter_type_dict[device.key] = "lsiLogic"
+
+ adapter_type = adapter_type_dict.get(vmdk_controler_key, "")
+
+ return vmdk_file_path, adapter_type
+
+
+def get_copy_virtual_disk_spec(client_factory, adapter_type="lsilogic"):
+ """Builds the Virtual Disk copy spec."""
+ dest_spec = client_factory.create('ns0:VirtualDiskSpec')
+ dest_spec.adapterType = adapter_type
+ dest_spec.diskType = "thick"
+ return dest_spec
+
+
+def get_vmdk_create_spec(client_factory, size_in_kb, adapter_type="lsiLogic"):
+ """Builds the virtual disk create spec."""
+ create_vmdk_spec = \
+ client_factory.create('ns0:FileBackedVirtualDiskSpec')
+ create_vmdk_spec.adapterType = adapter_type
+ create_vmdk_spec.diskType = "thick"
+ create_vmdk_spec.capacityKb = size_in_kb
+ return create_vmdk_spec
+
+
+def create_virtual_disk_spec(client_factory, disksize, controller_key,
+ file_path=None):
+ """
+ Builds spec for the creation of a new/ attaching of an already existing
+ Virtual Disk to the VM.
+ """
+ virtual_device_config = \
+ client_factory.create('ns0:VirtualDeviceConfigSpec')
+ virtual_device_config.operation = "add"
+ if file_path is None:
+ virtual_device_config.fileOperation = "create"
+
+ virtual_disk = client_factory.create('ns0:VirtualDisk')
+
+ disk_file_backing = \
+ client_factory.create('ns0:VirtualDiskFlatVer2BackingInfo')
+ disk_file_backing.diskMode = "persistent"
+ disk_file_backing.thinProvisioned = False
+ if file_path is not None:
+ disk_file_backing.fileName = file_path
+ else:
+ disk_file_backing.fileName = ""
+
+ connectable_spec = client_factory.create('ns0:VirtualDeviceConnectInfo')
+ connectable_spec.startConnected = True
+ connectable_spec.allowGuestControl = False
+ connectable_spec.connected = True
+
+ virtual_disk.backing = disk_file_backing
+ virtual_disk.connectable = connectable_spec
+
+ # The Server assigns a Key to the device. Here we pass a -ve random key.
+ # -ve because actual keys are +ve numbers and we don't
+ # want a clash with the key that server might associate with the device
+ virtual_disk.key = -100
+ virtual_disk.controllerKey = controller_key
+ virtual_disk.unitNumber = 0
+ virtual_disk.capacityInKB = disksize
+
+ virtual_device_config.device = virtual_disk
+
+ return virtual_device_config
+
+
+def get_dummy_vm_create_spec(client_factory, name, data_store_name):
+ """Builds the dummy VM create spec."""
+ config_spec = client_factory.create('ns0:VirtualMachineConfigSpec')
+
+ config_spec.name = name
+ config_spec.guestId = "otherGuest"
+
+ vm_file_info = client_factory.create('ns0:VirtualMachineFileInfo')
+ vm_file_info.vmPathName = "[" + data_store_name + "]"
+ config_spec.files = vm_file_info
+
+ tools_info = client_factory.create('ns0:ToolsConfigInfo')
+ tools_info.afterPowerOn = True
+ tools_info.afterResume = True
+ tools_info.beforeGuestStandby = True
+ tools_info.beforeGuestShutdown = True
+ tools_info.beforeGuestReboot = True
+
+ config_spec.tools = tools_info
+ config_spec.numCPUs = 1
+ config_spec.memoryMB = 4
+
+ controller_key = -101
+ controller_spec = create_controller_spec(client_factory, controller_key)
+ disk_spec = create_virtual_disk_spec(client_factory, 1024, controller_key)
+
+ device_config_spec = [controller_spec, disk_spec]
+
+ config_spec.deviceChange = device_config_spec
+ return config_spec
+
+
+def get_machine_id_change_spec(client_factory, mac, ip_addr, netmask, gateway):
+ """Builds the machine id change config spec."""
+ machine_id_str = "%s;%s;%s;%s" % (mac, ip_addr, netmask, gateway)
+ virtual_machine_config_spec = \
+ client_factory.create('ns0:VirtualMachineConfigSpec')
+
+ opt = client_factory.create('ns0:OptionValue')
+ opt.key = "machine.id"
+ opt.value = machine_id_str
+ virtual_machine_config_spec.extraConfig = [opt]
+ return virtual_machine_config_spec
+
+
+def get_add_vswitch_port_group_spec(client_factory, vswitch_name,
+ port_group_name, vlan_id):
+ """Builds the virtual switch port group add spec."""
+ vswitch_port_group_spec = client_factory.create('ns0:HostPortGroupSpec')
+ vswitch_port_group_spec.name = port_group_name
+ vswitch_port_group_spec.vswitchName = vswitch_name
+
+ # VLAN ID of 0 means that VLAN tagging is not to be done for the network.
+ vswitch_port_group_spec.vlanId = int(vlan_id)
+
+ policy = client_factory.create('ns0:HostNetworkPolicy')
+ nicteaming = client_factory.create('ns0:HostNicTeamingPolicy')
+ nicteaming.notifySwitches = True
+ policy.nicTeaming = nicteaming
+
+ vswitch_port_group_spec.policy = policy
+ return vswitch_port_group_spec
diff --git a/nova/virt/vmwareapi/vmops.py b/nova/virt/vmwareapi/vmops.py new file mode 100644 index 000000000..cf6c88bbd --- /dev/null +++ b/nova/virt/vmwareapi/vmops.py @@ -0,0 +1,789 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""
+Class for VM tasks like spawn, snapshot, suspend, resume etc.
+"""
+
+import base64
+import os
+import time
+import urllib
+import urllib2
+import uuid
+
+from nova import context
+from nova import db
+from nova import exception
+from nova import flags
+from nova import log as logging
+from nova.compute import power_state
+from nova.virt.vmwareapi import vim_util
+from nova.virt.vmwareapi import vm_util
+from nova.virt.vmwareapi import vmware_images
+from nova.virt.vmwareapi import network_utils
+
+FLAGS = flags.FLAGS
+LOG = logging.getLogger("nova.virt.vmwareapi.vmops")
+
+VMWARE_POWER_STATES = {
+ 'poweredOff': power_state.SHUTDOWN,
+ 'poweredOn': power_state.RUNNING,
+ 'suspended': power_state.PAUSED}
+
+
+class VMWareVMOps(object):
+ """Management class for VM-related tasks."""
+
+ def __init__(self, session):
+ """Initializer."""
+ self._session = session
+
+ def _wait_with_callback(self, instance_id, task, callback):
+ """Waits for the task to finish and does a callback after."""
+ ret = None
+ try:
+ ret = self._session._wait_for_task(instance_id, task)
+ except Exception, excep:
+ LOG.exception(excep)
+ callback(ret)
+
+ def list_instances(self):
+ """Lists the VM instances that are registered with the ESX host."""
+ LOG.debug(_("Getting list of instances"))
+ vms = self._session._call_method(vim_util, "get_objects",
+ "VirtualMachine",
+ ["name", "runtime.connectionState"])
+ lst_vm_names = []
+ for vm in vms:
+ vm_name = None
+ conn_state = None
+ for prop in vm.propSet:
+ if prop.name == "name":
+ vm_name = prop.val
+ elif prop.name == "runtime.connectionState":
+ conn_state = prop.val
+ # Ignoring the oprhaned or inaccessible VMs
+ if conn_state not in ["orphaned", "inaccessible"]:
+ lst_vm_names.append(vm_name)
+ LOG.debug(_("Got total of %s instances") % str(len(lst_vm_names)))
+ return lst_vm_names
+
+ def spawn(self, instance):
+ """
+ Creates a VM instance.
+
+ Steps followed are:
+ 1. Create a VM with no disk and the specifics in the instance object
+ like RAM size.
+ 2. Create a dummy vmdk of the size of the disk file that is to be
+ uploaded. This is required just to create the metadata file.
+ 3. Delete the -flat.vmdk file created in the above step and retain
+ the metadata .vmdk file.
+ 4. Upload the disk file.
+ 5. Attach the disk to the VM by reconfiguring the same.
+ 6. Power on the VM.
+ """
+ vm_ref = self._get_vm_ref_from_the_name(instance.name)
+ if vm_ref:
+ raise exception.Duplicate(_("Attempted to create a VM with a name"
+ " %s, but that already exists on the host") % instance.name)
+
+ client_factory = self._session._get_vim().client.factory
+ service_content = self._session._get_vim().get_service_content()
+
+ network = db.network_get_by_instance(context.get_admin_context(),
+ instance['id'])
+
+ net_name = network['bridge']
+
+ def _check_if_network_bridge_exists():
+ network_ref = \
+ network_utils.get_network_with_the_name(self._session,
+ net_name)
+ if network_ref is None:
+ raise exception.NotFound(_("Network with the name '%s' doesn't"
+ " exist on the ESX host") % net_name)
+
+ _check_if_network_bridge_exists()
+
+ def _get_datastore_ref():
+ """Get the datastore list and choose the first local storage."""
+ data_stores = self._session._call_method(vim_util, "get_objects",
+ "Datastore", ["summary.type", "summary.name"])
+ for elem in data_stores:
+ ds_name = None
+ ds_type = None
+ for prop in elem.propSet:
+ if prop.name == "summary.type":
+ ds_type = prop.val
+ elif prop.name == "summary.name":
+ ds_name = prop.val
+ # Local storage identifier
+ if ds_type == "VMFS":
+ data_store_name = ds_name
+ return data_store_name
+
+ if data_store_name is None:
+ msg = _("Couldn't get a local Datastore reference")
+ LOG.exception(msg)
+ raise exception.Error(msg)
+
+ data_store_name = _get_datastore_ref()
+
+ def _get_image_properties():
+ """
+ Get the Size of the flat vmdk file that is there on the storage
+ repository.
+ """
+ image_size, image_properties = \
+ vmware_images.get_vmdk_size_and_properties(
+ instance.image_id, instance)
+ vmdk_file_size_in_kb = int(image_size) / 1024
+ os_type = image_properties.get("vmware_ostype", "otherGuest")
+ adapter_type = image_properties.get("vmware_adaptertype",
+ "lsiLogic")
+ return vmdk_file_size_in_kb, os_type, adapter_type
+
+ vmdk_file_size_in_kb, os_type, adapter_type = _get_image_properties()
+
+ def _get_vmfolder_and_res_pool_mors():
+ """Get the Vm folder ref from the datacenter."""
+ dc_objs = self._session._call_method(vim_util, "get_objects",
+ "Datacenter", ["vmFolder"])
+ # There is only one default datacenter in a standalone ESX host
+ vm_folder_mor = dc_objs[0].propSet[0].val
+
+ # Get the resource pool. Taking the first resource pool coming our
+ # way. Assuming that is the default resource pool.
+ res_pool_mor = self._session._call_method(vim_util, "get_objects",
+ "ResourcePool")[0].obj
+ return vm_folder_mor, res_pool_mor
+
+ vm_folder_mor, res_pool_mor = _get_vmfolder_and_res_pool_mors()
+
+ # Get the create vm config spec
+ config_spec = vm_util.get_vm_create_spec(client_factory, instance,
+ data_store_name, net_name, os_type)
+
+ def _execute_create_vm():
+ """Create VM on ESX host."""
+ LOG.debug(_("Creating VM with the name %s on the ESX host") %
+ instance.name)
+ # Create the VM on the ESX host
+ vm_create_task = self._session._call_method(
+ self._session._get_vim(),
+ "CreateVM_Task", vm_folder_mor,
+ config=config_spec, pool=res_pool_mor)
+ self._session._wait_for_task(instance.id, vm_create_task)
+
+ LOG.debug(_("Created VM with the name %s on the ESX host") %
+ instance.name)
+
+ _execute_create_vm()
+
+ # Set the machine id for the VM for setting the IP
+ self._set_machine_id(client_factory, instance)
+
+ # Naming the VM files in correspondence with the VM instance name
+ # The flat vmdk file name
+ flat_uploaded_vmdk_name = "%s/%s-flat.vmdk" % (instance.name,
+ instance.name)
+ # The vmdk meta-data file
+ uploaded_vmdk_name = "%s/%s.vmdk" % (instance.name, instance.name)
+ flat_uploaded_vmdk_path = vm_util.build_datastore_path(data_store_name,
+ flat_uploaded_vmdk_name)
+ uploaded_vmdk_path = vm_util.build_datastore_path(data_store_name,
+ uploaded_vmdk_name)
+
+ def _create_virtual_disk():
+ """Create a virtual disk of the size of flat vmdk file."""
+ # Create a Virtual Disk of the size of the flat vmdk file. This is
+ # done just to generate the meta-data file whose specifics
+ # depend on the size of the disk, thin/thick provisioning and the
+ # storage adapter type.
+ # Here we assume thick provisioning and lsiLogic for the adapter
+ # type
+ LOG.debug(_("Creating Virtual Disk of size "
+ "%(vmdk_file_size_in_kb)s KB and adapter type "
+ "%(adapter_type)s on the ESX host local store"
+ " %(data_store_name)s") %
+ {"vmdk_file_size_in_kb": vmdk_file_size_in_kb,
+ "adapter_type": adapter_type,
+ "data_store_name": data_store_name})
+ vmdk_create_spec = vm_util.get_vmdk_create_spec(client_factory,
+ vmdk_file_size_in_kb, adapter_type)
+ vmdk_create_task = self._session._call_method(
+ self._session._get_vim(),
+ "CreateVirtualDisk_Task",
+ service_content.virtualDiskManager,
+ name=uploaded_vmdk_path,
+ datacenter=self._get_datacenter_name_and_ref()[0],
+ spec=vmdk_create_spec)
+ self._session._wait_for_task(instance.id, vmdk_create_task)
+ LOG.debug(_("Created Virtual Disk of size %(vmdk_file_size_in_kb)s"
+ " KB on the ESX host local store "
+ "%(data_store_name)s") %
+ {"vmdk_file_size_in_kb": vmdk_file_size_in_kb,
+ "data_store_name": data_store_name})
+
+ _create_virtual_disk()
+
+ def _delete_disk_file():
+ LOG.debug(_("Deleting the file %(flat_uploaded_vmdk_path)s "
+ "on the ESX host local"
+ "store %(data_store_name)s") %
+ {"flat_uploaded_vmdk_path": flat_uploaded_vmdk_path,
+ "data_store_name": data_store_name})
+ # Delete the -flat.vmdk file created. .vmdk file is retained.
+ vmdk_delete_task = self._session._call_method(
+ self._session._get_vim(),
+ "DeleteDatastoreFile_Task",
+ service_content.fileManager,
+ name=flat_uploaded_vmdk_path)
+ self._session._wait_for_task(instance.id, vmdk_delete_task)
+ LOG.debug(_("Deleted the file %(flat_uploaded_vmdk_path)s on the "
+ "ESX host local store %(data_store_name)s") %
+ {"flat_uploaded_vmdk_path": flat_uploaded_vmdk_path,
+ "data_store_name": data_store_name})
+
+ _delete_disk_file()
+
+ cookies = self._session._get_vim().client.options.transport.cookiejar
+
+ def _fetch_image_on_esx_datastore():
+ """Fetch image from Glance to ESX datastore."""
+ LOG.debug(_("Downloading image file data %(image_id)s to the ESX "
+ "data store %(data_store_name)s") %
+ ({'image_id': instance.image_id,
+ 'data_store_name': data_store_name}))
+ # Upload the -flat.vmdk file whose meta-data file we just created
+ # above
+ vmware_images.fetch_image(
+ instance.image_id,
+ instance,
+ host=self._session._host_ip,
+ data_center_name=self._get_datacenter_name_and_ref()[1],
+ datastore_name=data_store_name,
+ cookies=cookies,
+ file_path=flat_uploaded_vmdk_name)
+ LOG.debug(_("Downloaded image file data %(image_id)s to the ESX "
+ "data store %(data_store_name)s") %
+ ({'image_id': instance.image_id,
+ 'data_store_name': data_store_name}))
+ _fetch_image_on_esx_datastore()
+
+ vm_ref = self._get_vm_ref_from_the_name(instance.name)
+
+ def _attach_vmdk_to_the_vm():
+ """
+ Attach the vmdk uploaded to the VM. VM reconfigure is done
+ to do so.
+ """
+ vmdk_attach_config_spec = vm_util.get_vmdk_attach_config_spec(
+ client_factory,
+ vmdk_file_size_in_kb, uploaded_vmdk_path,
+ adapter_type)
+ LOG.debug(_("Reconfiguring VM instance %s to attach the image "
+ "disk") % instance.name)
+ reconfig_task = self._session._call_method(
+ self._session._get_vim(),
+ "ReconfigVM_Task", vm_ref,
+ spec=vmdk_attach_config_spec)
+ self._session._wait_for_task(instance.id, reconfig_task)
+ LOG.debug(_("Reconfigured VM instance %s to attach the image "
+ "disk") % instance.name)
+
+ _attach_vmdk_to_the_vm()
+
+ def _power_on_vm():
+ """Power on the VM."""
+ LOG.debug(_("Powering on the VM instance %s") % instance.name)
+ # Power On the VM
+ power_on_task = self._session._call_method(
+ self._session._get_vim(),
+ "PowerOnVM_Task", vm_ref)
+ self._session._wait_for_task(instance.id, power_on_task)
+ LOG.debug(_("Powered on the VM instance %s") % instance.name)
+ _power_on_vm()
+
+ def snapshot(self, instance, snapshot_name):
+ """
+ Create snapshot from a running VM instance.
+ Steps followed are:
+ 1. Get the name of the vmdk file which the VM points to right now.
+ Can be a chain of snapshots, so we need to know the last in the
+ chain.
+ 2. Create the snapshot. A new vmdk is created which the VM points to
+ now. The earlier vmdk becomes read-only.
+ 3. Call CopyVirtualDisk which coalesces the disk chain to form a single
+ vmdk, rather a .vmdk metadata file and a -flat.vmdk disk data file.
+ 4. Now upload the -flat.vmdk file to the image store.
+ 5. Delete the coalesced .vmdk and -flat.vmdk created.
+ """
+ vm_ref = self._get_vm_ref_from_the_name(instance.name)
+ if vm_ref is None:
+ raise exception.NotFound(_("instance - %s not present") %
+ instance.name)
+
+ client_factory = self._session._get_vim().client.factory
+ service_content = self._session._get_vim().get_service_content()
+
+ def _get_vm_and_vmdk_attribs():
+ # Get the vmdk file name that the VM is pointing to
+ hardware_devices = self._session._call_method(vim_util,
+ "get_dynamic_property", vm_ref,
+ "VirtualMachine", "config.hardware.device")
+ vmdk_file_path_before_snapshot, adapter_type = \
+ vm_util.get_vmdk_file_path_and_adapter_type(client_factory,
+ hardware_devices)
+ datastore_name = vm_util.split_datastore_path(
+ vmdk_file_path_before_snapshot)[0]
+ os_type = self._session._call_method(vim_util,
+ "get_dynamic_property", vm_ref,
+ "VirtualMachine", "summary.config.guestId")
+ return (vmdk_file_path_before_snapshot, adapter_type,
+ datastore_name, os_type)
+
+ vmdk_file_path_before_snapshot, adapter_type, datastore_name,\
+ os_type = _get_vm_and_vmdk_attribs()
+
+ def _create_vm_snapshot():
+ # Create a snapshot of the VM
+ LOG.debug(_("Creating Snapshot of the VM instance %s ") %
+ instance.name)
+ snapshot_task = self._session._call_method(
+ self._session._get_vim(),
+ "CreateSnapshot_Task", vm_ref,
+ name="%s-snapshot" % instance.name,
+ description="Taking Snapshot of the VM",
+ memory=True,
+ quiesce=True)
+ self._session._wait_for_task(instance.id, snapshot_task)
+ LOG.debug(_("Created Snapshot of the VM instance %s ") %
+ instance.name)
+
+ _create_vm_snapshot()
+
+ def _check_if_tmp_folder_exists():
+ # Copy the contents of the VM that were there just before the
+ # snapshot was taken
+ ds_ref_ret = vim_util.get_dynamic_property(
+ self._session._get_vim(),
+ vm_ref,
+ "VirtualMachine",
+ "datastore")
+ if not ds_ref_ret:
+ raise exception.NotFound(_("Failed to get the datastore "
+ "reference(s) which the VM uses"))
+ ds_ref = ds_ref_ret.ManagedObjectReference[0]
+ ds_browser = vim_util.get_dynamic_property(
+ self._session._get_vim(),
+ ds_ref,
+ "Datastore",
+ "browser")
+ # Check if the vmware-tmp folder exists or not. If not, create one
+ tmp_folder_path = vm_util.build_datastore_path(datastore_name,
+ "vmware-tmp")
+ if not self._path_exists(ds_browser, tmp_folder_path):
+ self._mkdir(vm_util.build_datastore_path(datastore_name,
+ "vmware-tmp"))
+
+ _check_if_tmp_folder_exists()
+
+ # Generate a random vmdk file name to which the coalesced vmdk content
+ # will be copied to. A random name is chosen so that we don't have
+ # name clashes.
+ random_name = str(uuid.uuid4())
+ dest_vmdk_file_location = vm_util.build_datastore_path(datastore_name,
+ "vmware-tmp/%s.vmdk" % random_name)
+ dc_ref = self._get_datacenter_name_and_ref()[0]
+
+ def _copy_vmdk_content():
+ # Copy the contents of the disk ( or disks, if there were snapshots
+ # done earlier) to a temporary vmdk file.
+ copy_spec = vm_util.get_copy_virtual_disk_spec(client_factory,
+ adapter_type)
+ LOG.debug(_("Copying disk data before snapshot of the VM "
+ " instance %s") % instance.name)
+ copy_disk_task = self._session._call_method(
+ self._session._get_vim(),
+ "CopyVirtualDisk_Task",
+ service_content.virtualDiskManager,
+ sourceName=vmdk_file_path_before_snapshot,
+ sourceDatacenter=dc_ref,
+ destName=dest_vmdk_file_location,
+ destDatacenter=dc_ref,
+ destSpec=copy_spec,
+ force=False)
+ self._session._wait_for_task(instance.id, copy_disk_task)
+ LOG.debug(_("Copied disk data before snapshot of the VM "
+ "instance %s") % instance.name)
+
+ _copy_vmdk_content()
+
+ cookies = self._session._get_vim().client.options.transport.cookiejar
+
+ def _upload_vmdk_to_image_repository():
+ # Upload the contents of -flat.vmdk file which has the disk data.
+ LOG.debug(_("Uploading image %s") % snapshot_name)
+ vmware_images.upload_image(
+ snapshot_name,
+ instance,
+ os_type=os_type,
+ adapter_type=adapter_type,
+ image_version=1,
+ host=self._session._host_ip,
+ data_center_name=self._get_datacenter_name_and_ref()[1],
+ datastore_name=datastore_name,
+ cookies=cookies,
+ file_path="vmware-tmp/%s-flat.vmdk" % random_name)
+ LOG.debug(_("Uploaded image %s") % snapshot_name)
+
+ _upload_vmdk_to_image_repository()
+
+ def _clean_temp_data():
+ """
+ Delete temporary vmdk files generated in image handling
+ operations.
+ """
+ # Delete the temporary vmdk created above.
+ LOG.debug(_("Deleting temporary vmdk file %s")
+ % dest_vmdk_file_location)
+ remove_disk_task = self._session._call_method(
+ self._session._get_vim(),
+ "DeleteVirtualDisk_Task",
+ service_content.virtualDiskManager,
+ name=dest_vmdk_file_location,
+ datacenter=dc_ref)
+ self._session._wait_for_task(instance.id, remove_disk_task)
+ LOG.debug(_("Deleted temporary vmdk file %s")
+ % dest_vmdk_file_location)
+
+ _clean_temp_data()
+
+ def reboot(self, instance):
+ """Reboot a VM instance."""
+ vm_ref = self._get_vm_ref_from_the_name(instance.name)
+ if vm_ref is None:
+ raise exception.NotFound(_("instance - %s not present") %
+ instance.name)
+ lst_properties = ["summary.guest.toolsStatus", "runtime.powerState",
+ "summary.guest.toolsRunningStatus"]
+ props = self._session._call_method(vim_util, "get_object_properties",
+ None, vm_ref, "VirtualMachine",
+ lst_properties)
+ pwr_state = None
+ tools_status = None
+ tools_running_status = False
+ for elem in props:
+ for prop in elem.propSet:
+ if prop.name == "runtime.powerState":
+ pwr_state = prop.val
+ elif prop.name == "summary.guest.toolsStatus":
+ tools_status = prop.val
+ elif prop.name == "summary.guest.toolsRunningStatus":
+ tools_running_status = prop.val
+
+ # Raise an exception if the VM is not powered On.
+ if pwr_state not in ["poweredOn"]:
+ raise exception.Invalid(_("instance - %s not poweredOn. So can't "
+ "be rebooted.") % instance.name)
+
+ # If latest vmware tools are installed in the VM, and that the tools
+ # are running, then only do a guest reboot. Otherwise do a hard reset.
+ if (tools_status == "toolsOk" and
+ tools_running_status == "guestToolsRunning"):
+ LOG.debug(_("Rebooting guest OS of VM %s") % instance.name)
+ self._session._call_method(self._session._get_vim(), "RebootGuest",
+ vm_ref)
+ LOG.debug(_("Rebooted guest OS of VM %s") % instance.name)
+ else:
+ LOG.debug(_("Doing hard reboot of VM %s") % instance.name)
+ reset_task = self._session._call_method(self._session._get_vim(),
+ "ResetVM_Task", vm_ref)
+ self._session._wait_for_task(instance.id, reset_task)
+ LOG.debug(_("Did hard reboot of VM %s") % instance.name)
+
+ def destroy(self, instance):
+ """
+ Destroy a VM instance. Steps followed are:
+ 1. Power off the VM, if it is in poweredOn state.
+ 2. Un-register a VM.
+ 3. Delete the contents of the folder holding the VM related data.
+ """
+ try:
+ vm_ref = self._get_vm_ref_from_the_name(instance.name)
+ if vm_ref is None:
+ LOG.debug(_("instance - %s not present") % instance.name)
+ return
+ lst_properties = ["config.files.vmPathName", "runtime.powerState"]
+ props = self._session._call_method(vim_util,
+ "get_object_properties",
+ None, vm_ref, "VirtualMachine", lst_properties)
+ pwr_state = None
+ for elem in props:
+ vm_config_pathname = None
+ for prop in elem.propSet:
+ if prop.name == "runtime.powerState":
+ pwr_state = prop.val
+ elif prop.name == "config.files.vmPathName":
+ vm_config_pathname = prop.val
+ if vm_config_pathname:
+ datastore_name, vmx_file_path = \
+ vm_util.split_datastore_path(vm_config_pathname)
+ # Power off the VM if it is in PoweredOn state.
+ if pwr_state == "poweredOn":
+ LOG.debug(_("Powering off the VM %s") % instance.name)
+ poweroff_task = self._session._call_method(
+ self._session._get_vim(),
+ "PowerOffVM_Task", vm_ref)
+ self._session._wait_for_task(instance.id, poweroff_task)
+ LOG.debug(_("Powered off the VM %s") % instance.name)
+
+ # Un-register the VM
+ try:
+ LOG.debug(_("Unregistering the VM %s") % instance.name)
+ self._session._call_method(self._session._get_vim(),
+ "UnregisterVM", vm_ref)
+ LOG.debug(_("Unregistered the VM %s") % instance.name)
+ except Exception, excep:
+ LOG.warn(_("In vmwareapi:vmops:destroy, got this exception"
+ " while un-registering the VM: %s") % str(excep))
+
+ # Delete the folder holding the VM related content on
+ # the datastore.
+ try:
+ dir_ds_compliant_path = vm_util.build_datastore_path(
+ datastore_name,
+ os.path.dirname(vmx_file_path))
+ LOG.debug(_("Deleting contents of the VM %(name)s from "
+ "datastore %(datastore_name)s") %
+ ({'name': instance.name,
+ 'datastore_name': datastore_name}))
+ delete_task = self._session._call_method(
+ self._session._get_vim(),
+ "DeleteDatastoreFile_Task",
+ self._session._get_vim().get_service_content().fileManager,
+ name=dir_ds_compliant_path)
+ self._session._wait_for_task(instance.id, delete_task)
+ LOG.debug(_("Deleted contents of the VM %(name)s from "
+ "datastore %(datastore_name)s") %
+ ({'name': instance.name,
+ 'datastore_name': datastore_name}))
+ except Exception, excep:
+ LOG.warn(_("In vmwareapi:vmops:destroy, "
+ "got this exception while deleting"
+ " the VM contents from the disk: %s")
+ % str(excep))
+ except Exception, exc:
+ LOG.exception(exc)
+
+ def pause(self, instance, callback):
+ """Pause a VM instance."""
+ raise exception.APIError("pause not supported for vmwareapi")
+
+ def unpause(self, instance, callback):
+ """Un-Pause a VM instance."""
+ raise exception.APIError("unpause not supported for vmwareapi")
+
+ def suspend(self, instance, callback):
+ """Suspend the specified instance."""
+ vm_ref = self._get_vm_ref_from_the_name(instance.name)
+ if vm_ref is None:
+ raise exception.NotFound(_("instance - %s not present") %
+ instance.name)
+
+ pwr_state = self._session._call_method(vim_util,
+ "get_dynamic_property", vm_ref,
+ "VirtualMachine", "runtime.powerState")
+ # Only PoweredOn VMs can be suspended.
+ if pwr_state == "poweredOn":
+ LOG.debug(_("Suspending the VM %s ") % instance.name)
+ suspend_task = self._session._call_method(self._session._get_vim(),
+ "SuspendVM_Task", vm_ref)
+ self._wait_with_callback(instance.id, suspend_task, callback)
+ LOG.debug(_("Suspended the VM %s ") % instance.name)
+ # Raise Exception if VM is poweredOff
+ elif pwr_state == "poweredOff":
+ raise exception.Invalid(_("instance - %s is poweredOff and hence "
+ " can't be suspended.") % instance.name)
+ LOG.debug(_("VM %s was already in suspended state. So returning "
+ "without doing anything") % instance.name)
+
+ def resume(self, instance, callback):
+ """Resume the specified instance."""
+ vm_ref = self._get_vm_ref_from_the_name(instance.name)
+ if vm_ref is None:
+ raise exception.NotFound(_("instance - %s not present") %
+ instance.name)
+
+ pwr_state = self._session._call_method(vim_util,
+ "get_dynamic_property", vm_ref,
+ "VirtualMachine", "runtime.powerState")
+ if pwr_state.lower() == "suspended":
+ LOG.debug(_("Resuming the VM %s") % instance.name)
+ suspend_task = self._session._call_method(
+ self._session._get_vim(),
+ "PowerOnVM_Task", vm_ref)
+ self._wait_with_callback(instance.id, suspend_task, callback)
+ LOG.debug(_("Resumed the VM %s ") % instance.name)
+ else:
+ raise exception.Invalid(_("instance - %s not in Suspended state "
+ "and hence can't be Resumed.") % instance.name)
+
+ def get_info(self, instance_name):
+ """Return data about the VM instance."""
+ vm_ref = self._get_vm_ref_from_the_name(instance_name)
+ if vm_ref is None:
+ raise exception.NotFound(_("instance - %s not present") %
+ instance_name)
+
+ lst_properties = ["summary.config.numCpu",
+ "summary.config.memorySizeMB",
+ "runtime.powerState"]
+ vm_props = self._session._call_method(vim_util,
+ "get_object_properties", None, vm_ref, "VirtualMachine",
+ lst_properties)
+ max_mem = None
+ pwr_state = None
+ num_cpu = None
+ for elem in vm_props:
+ for prop in elem.propSet:
+ if prop.name == "summary.config.numCpu":
+ num_cpu = int(prop.val)
+ elif prop.name == "summary.config.memorySizeMB":
+ # In MB, but we want in KB
+ max_mem = int(prop.val) * 1024
+ elif prop.name == "runtime.powerState":
+ pwr_state = VMWARE_POWER_STATES[prop.val]
+
+ return {'state': pwr_state,
+ 'max_mem': max_mem,
+ 'mem': max_mem,
+ 'num_cpu': num_cpu,
+ 'cpu_time': 0}
+
+ def get_diagnostics(self, instance):
+ """Return data about VM diagnostics."""
+ raise exception.APIError("get_diagnostics not implemented for "
+ "vmwareapi")
+
+ def get_console_output(self, instance):
+ """Return snapshot of console."""
+ vm_ref = self._get_vm_ref_from_the_name(instance.name)
+ if vm_ref is None:
+ raise exception.NotFound(_("instance - %s not present") %
+ instance.name)
+ param_list = {"id": str(vm_ref)}
+ base_url = "%s://%s/screen?%s" % (self._session._scheme,
+ self._session._host_ip,
+ urllib.urlencode(param_list))
+ request = urllib2.Request(base_url)
+ base64string = base64.encodestring(
+ '%s:%s' % (
+ self._session._host_username,
+ self._session._host_password)).replace('\n', '')
+ request.add_header("Authorization", "Basic %s" % base64string)
+ result = urllib2.urlopen(request)
+ if result.code == 200:
+ return result.read()
+ else:
+ return ""
+
+ def get_ajax_console(self, instance):
+ """Return link to instance's ajax console."""
+ return 'http://fakeajaxconsole/fake_url'
+
+ def _set_machine_id(self, client_factory, instance):
+ """
+ Set the machine id of the VM for guest tools to pick up and change
+ the IP.
+ """
+ vm_ref = self._get_vm_ref_from_the_name(instance.name)
+ if vm_ref is None:
+ raise exception.NotFound(_("instance - %s not present") %
+ instance.name)
+ network = db.network_get_by_instance(context.get_admin_context(),
+ instance['id'])
+ mac_addr = instance.mac_address
+ net_mask = network["netmask"]
+ gateway = network["gateway"]
+ ip_addr = db.instance_get_fixed_address(context.get_admin_context(),
+ instance['id'])
+ machine_id_chanfge_spec = \
+ vm_util.get_machine_id_change_spec(client_factory, mac_addr,
+ ip_addr, net_mask, gateway)
+ LOG.debug(_("Reconfiguring VM instance %(name)s to set the machine id "
+ "with ip - %(ip_addr)s") %
+ ({'name': instance.name,
+ 'ip_addr': ip_addr}))
+ reconfig_task = self._session._call_method(self._session._get_vim(),
+ "ReconfigVM_Task", vm_ref,
+ spec=machine_id_chanfge_spec)
+ self._session._wait_for_task(instance.id, reconfig_task)
+ LOG.debug(_("Reconfigured VM instance %(name)s to set the machine id "
+ "with ip - %(ip_addr)s") %
+ ({'name': instance.name,
+ 'ip_addr': ip_addr}))
+
+ def _get_datacenter_name_and_ref(self):
+ """Get the datacenter name and the reference."""
+ dc_obj = self._session._call_method(vim_util, "get_objects",
+ "Datacenter", ["name"])
+ return dc_obj[0].obj, dc_obj[0].propSet[0].val
+
+ def _path_exists(self, ds_browser, ds_path):
+ """Check if the path exists on the datastore."""
+ search_task = self._session._call_method(self._session._get_vim(),
+ "SearchDatastore_Task",
+ ds_browser,
+ datastorePath=ds_path)
+ # Wait till the state changes from queued or running.
+ # If an error state is returned, it means that the path doesn't exist.
+ while True:
+ task_info = self._session._call_method(vim_util,
+ "get_dynamic_property",
+ search_task, "Task", "info")
+ if task_info.state in ['queued', 'running']:
+ time.sleep(2)
+ continue
+ break
+ if task_info.state == "error":
+ return False
+ return True
+
+ def _mkdir(self, ds_path):
+ """
+ Creates a directory at the path specified. If it is just "NAME",
+ then a directory with this name is created at the topmost level of the
+ DataStore.
+ """
+ LOG.debug(_("Creating directory with path %s") % ds_path)
+ self._session._call_method(self._session._get_vim(), "MakeDirectory",
+ self._session._get_vim().get_service_content().fileManager,
+ name=ds_path, createParentDirectories=False)
+ LOG.debug(_("Created directory with path %s") % ds_path)
+
+ def _get_vm_ref_from_the_name(self, vm_name):
+ """Get reference to the VM with the name specified."""
+ vms = self._session._call_method(vim_util, "get_objects",
+ "VirtualMachine", ["name"])
+ for vm in vms:
+ if vm.propSet[0].val == vm_name:
+ return vm.obj
+ return None
diff --git a/nova/virt/vmwareapi/vmware_images.py b/nova/virt/vmwareapi/vmware_images.py new file mode 100644 index 000000000..50c6baedf --- /dev/null +++ b/nova/virt/vmwareapi/vmware_images.py @@ -0,0 +1,201 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+"""
+Utility functions for Image transfer.
+"""
+
+from glance import client
+
+from nova import exception
+from nova import flags
+from nova import log as logging
+from nova.virt.vmwareapi import io_util
+from nova.virt.vmwareapi import read_write_util
+
+LOG = logging.getLogger("nova.virt.vmwareapi.vmware_images")
+
+FLAGS = flags.FLAGS
+
+QUEUE_BUFFER_SIZE = 10
+
+
+def start_transfer(read_file_handle, data_size, write_file_handle=None,
+ glance_client=None, image_id=None, image_meta={}):
+ """Start the data transfer from the reader to the writer.
+ Reader writes to the pipe and the writer reads from the pipe. This means
+ that the total transfer time boils down to the slower of the read/write
+ and not the addition of the two times."""
+ # The pipe that acts as an intermediate store of data for reader to write
+ # to and writer to grab from.
+ thread_safe_pipe = io_util.ThreadSafePipe(QUEUE_BUFFER_SIZE, data_size)
+ # The read thread. In case of glance it is the instance of the
+ # GlanceFileRead class. The glance client read returns an iterator
+ # and this class wraps that iterator to provide datachunks in calls
+ # to read.
+ read_thread = io_util.IOThread(read_file_handle, thread_safe_pipe)
+
+ # In case of Glance - VMWare transfer, we just need a handle to the
+ # HTTP Connection that is to send transfer data to the VMWare datastore.
+ if write_file_handle:
+ write_thread = io_util.IOThread(thread_safe_pipe, write_file_handle)
+ # In case of VMWare - Glance transfer, we relinquish VMWare HTTP file read
+ # handle to Glance Client instance, but to be sure of the transfer we need
+ # to be sure of the status of the image on glnace changing to active.
+ # The GlanceWriteThread handles the same for us.
+ elif glance_client and image_id:
+ write_thread = io_util.GlanceWriteThread(thread_safe_pipe,
+ glance_client, image_id, image_meta)
+ # Start the read and write threads.
+ read_event = read_thread.start()
+ write_event = write_thread.start()
+ try:
+ # Wait on the read and write events to signal their end
+ read_event.wait()
+ write_event.wait()
+ except Exception, exc:
+ # In case of any of the reads or writes raising an exception,
+ # stop the threads so that we un-necessarily don't keep the other one
+ # waiting.
+ read_thread.stop()
+ write_thread.stop()
+
+ # Log and raise the exception.
+ LOG.exception(exc)
+ raise exception.Error(exc)
+ finally:
+ # No matter what, try closing the read and write handles, if it so
+ # applies.
+ read_file_handle.close()
+ if write_file_handle:
+ write_file_handle.close()
+
+
+def fetch_image(image, instance, **kwargs):
+ """Fetch an image for attaching to the newly created VM."""
+ # Depending upon the image service, make appropriate image service call
+ if FLAGS.image_service == "nova.image.glance.GlanceImageService":
+ func = _get_glance_image
+ elif FLAGS.image_service == "nova.image.s3.S3ImageService":
+ func = _get_s3_image
+ elif FLAGS.image_service == "nova.image.local.LocalImageService":
+ func = _get_local_image
+ else:
+ raise NotImplementedError(_("The Image Service %s is not implemented")
+ % FLAGS.image_service)
+ return func(image, instance, **kwargs)
+
+
+def upload_image(image, instance, **kwargs):
+ """Upload the newly snapshotted VM disk file."""
+ # Depending upon the image service, make appropriate image service call
+ if FLAGS.image_service == "nova.image.glance.GlanceImageService":
+ func = _put_glance_image
+ elif FLAGS.image_service == "nova.image.s3.S3ImageService":
+ func = _put_s3_image
+ elif FLAGS.image_service == "nova.image.local.LocalImageService":
+ func = _put_local_image
+ else:
+ raise NotImplementedError(_("The Image Service %s is not implemented")
+ % FLAGS.image_service)
+ return func(image, instance, **kwargs)
+
+
+def _get_glance_image(image, instance, **kwargs):
+ """Download image from the glance image server."""
+ LOG.debug(_("Downloading image %s from glance image server") % image)
+ glance_client = client.Client(FLAGS.glance_host, FLAGS.glance_port)
+ metadata, read_iter = glance_client.get_image(image)
+ read_file_handle = read_write_util.GlanceFileRead(read_iter)
+ file_size = int(metadata['size'])
+ write_file_handle = read_write_util.VMWareHTTPWriteFile(
+ kwargs.get("host"),
+ kwargs.get("data_center_name"),
+ kwargs.get("datastore_name"),
+ kwargs.get("cookies"),
+ kwargs.get("file_path"),
+ file_size)
+ start_transfer(read_file_handle, file_size,
+ write_file_handle=write_file_handle)
+ LOG.debug(_("Downloaded image %s from glance image server") % image)
+
+
+def _get_s3_image(image, instance, **kwargs):
+ """Download image from the S3 image server."""
+ raise NotImplementedError
+
+
+def _get_local_image(image, instance, **kwargs):
+ """Download image from the local nova compute node."""
+ raise NotImplementedError
+
+
+def _put_glance_image(image, instance, **kwargs):
+ """Upload the snapshotted vm disk file to Glance image server."""
+ LOG.debug(_("Uploading image %s to the Glance image server") % image)
+ read_file_handle = read_write_util.VmWareHTTPReadFile(
+ kwargs.get("host"),
+ kwargs.get("data_center_name"),
+ kwargs.get("datastore_name"),
+ kwargs.get("cookies"),
+ kwargs.get("file_path"))
+ file_size = read_file_handle.get_size()
+ glance_client = client.Client(FLAGS.glance_host, FLAGS.glance_port)
+ # The properties and other fields that we need to set for the image.
+ image_metadata = {"is_public": True,
+ "disk_format": "vmdk",
+ "container_format": "bare",
+ "type": "vmdk",
+ "properties": {"vmware_adaptertype":
+ kwargs.get("adapter_type"),
+ "vmware_ostype": kwargs.get("os_type"),
+ "vmware_image_version":
+ kwargs.get("image_version")}}
+ start_transfer(read_file_handle, file_size, glance_client=glance_client,
+ image_id=image, image_meta=image_metadata)
+ LOG.debug(_("Uploaded image %s to the Glance image server") % image)
+
+
+def _put_local_image(image, instance, **kwargs):
+ """Upload the snapshotted vm disk file to the local nova compute node."""
+ raise NotImplementedError
+
+
+def _put_s3_image(image, instance, **kwargs):
+ """Upload the snapshotted vm disk file to S3 image server."""
+ raise NotImplementedError
+
+
+def get_vmdk_size_and_properties(image, instance):
+ """
+ Get size of the vmdk file that is to be downloaded for attach in spawn.
+ Need this to create the dummy virtual disk for the meta-data file. The
+ geometry of the disk created depends on the size.
+ """
+
+ LOG.debug(_("Getting image size for the image %s") % image)
+ if FLAGS.image_service == "nova.image.glance.GlanceImageService":
+ glance_client = client.Client(FLAGS.glance_host,
+ FLAGS.glance_port)
+ meta_data = glance_client.get_image_meta(image)
+ size, properties = meta_data["size"], meta_data["properties"]
+ elif FLAGS.image_service == "nova.image.s3.S3ImageService":
+ raise NotImplementedError
+ elif FLAGS.image_service == "nova.image.local.LocalImageService":
+ raise NotImplementedError
+ LOG.debug(_("Got image size of %(size)s for the image %(image)s") %
+ locals())
+ return size, properties
diff --git a/nova/virt/vmwareapi_conn.py b/nova/virt/vmwareapi_conn.py new file mode 100644 index 000000000..20c1b2b45 --- /dev/null +++ b/nova/virt/vmwareapi_conn.py @@ -0,0 +1,376 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Citrix Systems, Inc.
+# Copyright 2011 OpenStack LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""
+A connection to the VMware ESX platform.
+
+**Related Flags**
+
+:vmwareapi_host_ip: IPAddress of VMware ESX server.
+:vmwareapi_host_username: Username for connection to VMware ESX Server.
+:vmwareapi_host_password: Password for connection to VMware ESX Server.
+:vmwareapi_task_poll_interval: The interval (seconds) used for polling of
+ remote tasks
+ (default: 1.0).
+:vmwareapi_api_retry_count: The API retry count in case of failure such as
+ network failures (socket errors etc.)
+ (default: 10).
+
+"""
+
+import time
+
+from eventlet import event
+
+from nova import context
+from nova import db
+from nova import exception
+from nova import flags
+from nova import log as logging
+from nova import utils
+from nova.virt.vmwareapi import error_util
+from nova.virt.vmwareapi import vim
+from nova.virt.vmwareapi import vim_util
+from nova.virt.vmwareapi.vmops import VMWareVMOps
+
+
+LOG = logging.getLogger("nova.virt.vmwareapi_conn")
+
+FLAGS = flags.FLAGS
+flags.DEFINE_string('vmwareapi_host_ip',
+ None,
+ 'URL for connection to VMWare ESX host.'
+ 'Required if connection_type is vmwareapi.')
+flags.DEFINE_string('vmwareapi_host_username',
+ None,
+ 'Username for connection to VMWare ESX host.'
+ 'Used only if connection_type is vmwareapi.')
+flags.DEFINE_string('vmwareapi_host_password',
+ None,
+ 'Password for connection to VMWare ESX host.'
+ 'Used only if connection_type is vmwareapi.')
+flags.DEFINE_float('vmwareapi_task_poll_interval',
+ 5.0,
+ 'The interval used for polling of remote tasks '
+ 'Used only if connection_type is vmwareapi')
+flags.DEFINE_float('vmwareapi_api_retry_count',
+ 10,
+ 'The number of times we retry on failures, '
+ 'e.g., socket error, etc.'
+ 'Used only if connection_type is vmwareapi')
+flags.DEFINE_string('vmwareapi_vlan_interface',
+ 'vmnic0',
+ 'Physical ethernet adapter name for vlan networking')
+
+TIME_BETWEEN_API_CALL_RETRIES = 2.0
+
+
+class Failure(Exception):
+ """Base Exception class for handling task failures."""
+
+ def __init__(self, details):
+ self.details = details
+
+ def __str__(self):
+ return str(self.details)
+
+
+def get_connection(_):
+ """Sets up the ESX host connection."""
+ host_ip = FLAGS.vmwareapi_host_ip
+ host_username = FLAGS.vmwareapi_host_username
+ host_password = FLAGS.vmwareapi_host_password
+ api_retry_count = FLAGS.vmwareapi_api_retry_count
+ if not host_ip or host_username is None or host_password is None:
+ raise Exception(_("Must specify vmwareapi_host_ip,"
+ "vmwareapi_host_username "
+ "and vmwareapi_host_password to use"
+ "connection_type=vmwareapi"))
+ return VMWareESXConnection(host_ip, host_username, host_password,
+ api_retry_count)
+
+
+class VMWareESXConnection(object):
+ """The ESX host connection object."""
+
+ def __init__(self, host_ip, host_username, host_password,
+ api_retry_count, scheme="https"):
+ session = VMWareAPISession(host_ip, host_username, host_password,
+ api_retry_count, scheme=scheme)
+ self._vmops = VMWareVMOps(session)
+
+ def init_host(self, host):
+ """Do the initialization that needs to be done."""
+ # FIXME(sateesh): implement this
+ pass
+
+ def list_instances(self):
+ """List VM instances."""
+ return self._vmops.list_instances()
+
+ def spawn(self, instance):
+ """Create VM instance."""
+ self._vmops.spawn(instance)
+
+ def snapshot(self, instance, name):
+ """Create snapshot from a running VM instance."""
+ self._vmops.snapshot(instance, name)
+
+ def reboot(self, instance):
+ """Reboot VM instance."""
+ self._vmops.reboot(instance)
+
+ def destroy(self, instance):
+ """Destroy VM instance."""
+ self._vmops.destroy(instance)
+
+ def pause(self, instance, callback):
+ """Pause VM instance."""
+ self._vmops.pause(instance, callback)
+
+ def unpause(self, instance, callback):
+ """Unpause paused VM instance."""
+ self._vmops.unpause(instance, callback)
+
+ def suspend(self, instance, callback):
+ """Suspend the specified instance."""
+ self._vmops.suspend(instance, callback)
+
+ def resume(self, instance, callback):
+ """Resume the suspended VM instance."""
+ self._vmops.resume(instance, callback)
+
+ def get_info(self, instance_id):
+ """Return info about the VM instance."""
+ return self._vmops.get_info(instance_id)
+
+ def get_diagnostics(self, instance):
+ """Return data about VM diagnostics."""
+ return self._vmops.get_info(instance)
+
+ def get_console_output(self, instance):
+ """Return snapshot of console."""
+ return self._vmops.get_console_output(instance)
+
+ def get_ajax_console(self, instance):
+ """Return link to instance's ajax console."""
+ return self._vmops.get_ajax_console(instance)
+
+ def attach_volume(self, instance_name, device_path, mountpoint):
+ """Attach volume storage to VM instance."""
+ pass
+
+ def detach_volume(self, instance_name, mountpoint):
+ """Detach volume storage to VM instance."""
+ pass
+
+ def get_console_pool_info(self, console_type):
+ """Get info about the host on which the VM resides."""
+ return {'address': FLAGS.vmwareapi_host_ip,
+ 'username': FLAGS.vmwareapi_host_username,
+ 'password': FLAGS.vmwareapi_host_password}
+
+ def update_available_resource(self, ctxt, host):
+ """This method is supported only by libvirt."""
+ return
+
+
+class VMWareAPISession(object):
+ """
+ Sets up a session with the ESX host and handles all
+ the calls made to the host.
+ """
+
+ def __init__(self, host_ip, host_username, host_password,
+ api_retry_count, scheme="https"):
+ self._host_ip = host_ip
+ self._host_username = host_username
+ self._host_password = host_password
+ self.api_retry_count = api_retry_count
+ self._scheme = scheme
+ self._session_id = None
+ self.vim = None
+ self._create_session()
+
+ def _get_vim_object(self):
+ """Create the VIM Object instance."""
+ return vim.Vim(protocol=self._scheme, host=self._host_ip)
+
+ def _create_session(self):
+ """Creates a session with the ESX host."""
+ while True:
+ try:
+ # Login and setup the session with the ESX host for making
+ # API calls
+ self.vim = self._get_vim_object()
+ session = self.vim.Login(
+ self.vim.get_service_content().sessionManager,
+ userName=self._host_username,
+ password=self._host_password)
+ # Terminate the earlier session, if possible ( For the sake of
+ # preserving sessions as there is a limit to the number of
+ # sessions we can have )
+ if self._session_id:
+ try:
+ self.vim.TerminateSession(
+ self.vim.get_service_content().sessionManager,
+ sessionId=[self._session_id])
+ except Exception, excep:
+ # This exception is something we can live with. It is
+ # just an extra caution on our side. The session may
+ # have been cleared. We could have made a call to
+ # SessionIsActive, but that is an overhead because we
+ # anyway would have to call TerminateSession.
+ LOG.debug(excep)
+ self._session_id = session.key
+ return
+ except Exception, excep:
+ LOG.critical(_("In vmwareapi:_create_session, "
+ "got this exception: %s") % excep)
+ raise exception.Error(excep)
+
+ def __del__(self):
+ """Logs-out the session."""
+ # Logout to avoid un-necessary increase in session count at the
+ # ESX host
+ try:
+ self.vim.Logout(self.vim.get_service_content().sessionManager)
+ except Exception, excep:
+ # It is just cautionary on our part to do a logout in del just
+ # to ensure that the session is not left active.
+ LOG.debug(excep)
+
+ def _is_vim_object(self, module):
+ """Check if the module is a VIM Object instance."""
+ return isinstance(module, vim.Vim)
+
+ def _call_method(self, module, method, *args, **kwargs):
+ """
+ Calls a method within the module specified with
+ args provided.
+ """
+ args = list(args)
+ retry_count = 0
+ exc = None
+ last_fault_list = []
+ while True:
+ try:
+ if not self._is_vim_object(module):
+ # If it is not the first try, then get the latest
+ # vim object
+ if retry_count > 0:
+ args = args[1:]
+ args = [self.vim] + args
+ retry_count += 1
+ temp_module = module
+
+ for method_elem in method.split("."):
+ temp_module = getattr(temp_module, method_elem)
+
+ return temp_module(*args, **kwargs)
+ except error_util.VimFaultException, excep:
+ # If it is a Session Fault Exception, it may point
+ # to a session gone bad. So we try re-creating a session
+ # and then proceeding ahead with the call.
+ exc = excep
+ if error_util.FAULT_NOT_AUTHENTICATED in excep.fault_list:
+ # Because of the idle session returning an empty
+ # RetrievePropertiesResponse and also the same is returned
+ # when there is say empty answer to the query for
+ # VMs on the host ( as in no VMs on the host), we have no
+ # way to differentiate.
+ # So if the previous response was also am empty response
+ # and after creating a new session, we get the same empty
+ # response, then we are sure of the response being supposed
+ # to be empty.
+ if error_util.FAULT_NOT_AUTHENTICATED in last_fault_list:
+ return []
+ last_fault_list = excep.fault_list
+ self._create_session()
+ else:
+ # No re-trying for errors for API call has gone through
+ # and is the caller's fault. Caller should handle these
+ # errors. e.g, InvalidArgument fault.
+ break
+ except error_util.SessionOverLoadException, excep:
+ # For exceptions which may come because of session overload,
+ # we retry
+ exc = excep
+ except Exception, excep:
+ # If it is a proper exception, say not having furnished
+ # proper data in the SOAP call or the retry limit having
+ # exceeded, we raise the exception
+ exc = excep
+ break
+ # If retry count has been reached then break and
+ # raise the exception
+ if retry_count > self.api_retry_count:
+ break
+ time.sleep(TIME_BETWEEN_API_CALL_RETRIES)
+
+ LOG.critical(_("In vmwareapi:_call_method, "
+ "got this exception: %s") % exc)
+ raise
+
+ def _get_vim(self):
+ """Gets the VIM object reference."""
+ if self.vim is None:
+ self._create_session()
+ return self.vim
+
+ def _wait_for_task(self, instance_id, task_ref):
+ """
+ Return a Deferred that will give the result of the given task.
+ The task is polled until it completes.
+ """
+ done = event.Event()
+ loop = utils.LoopingCall(self._poll_task, instance_id, task_ref,
+ done)
+ loop.start(FLAGS.vmwareapi_task_poll_interval, now=True)
+ ret_val = done.wait()
+ loop.stop()
+ return ret_val
+
+ def _poll_task(self, instance_id, task_ref, done):
+ """
+ Poll the given task, and fires the given Deferred if we
+ get a result.
+ """
+ try:
+ task_info = self._call_method(vim_util, "get_dynamic_property",
+ task_ref, "Task", "info")
+ task_name = task_info.name
+ action = dict(
+ instance_id=int(instance_id),
+ action=task_name[0:255],
+ error=None)
+ if task_info.state in ['queued', 'running']:
+ return
+ elif task_info.state == 'success':
+ LOG.debug(_("Task [%(task_name)s] %(task_ref)s "
+ "status: success") % locals())
+ done.send("success")
+ else:
+ error_info = str(task_info.error.localizedMessage)
+ action["error"] = error_info
+ LOG.warn(_("Task [%(task_name)s] %(task_ref)s "
+ "status: error %(error_info)s") % locals())
+ done.send_exception(exception.Error(error_info))
+ db.instance_action_create(context.get_admin_context(), action)
+ except Exception, excep:
+ LOG.warn(_("In vmwareapi:_poll_task, Got this error %s") % excep)
+ done.send_exception(excep)
diff --git a/nova/virt/xenapi/fake.py b/nova/virt/xenapi/fake.py index 018d0dcd3..4434dbf0b 100644 --- a/nova/virt/xenapi/fake.py +++ b/nova/virt/xenapi/fake.py @@ -60,8 +60,8 @@ from nova import exception from nova import log as logging -_CLASSES = ['host', 'network', 'session', 'SR', 'VBD',\ - 'PBD', 'VDI', 'VIF', 'VM', 'task'] +_CLASSES = ['host', 'network', 'session', 'SR', 'VBD', + 'PBD', 'VDI', 'VIF', 'PIF', 'VM', 'VLAN', 'task'] _db_content = {} @@ -78,30 +78,36 @@ def reset(): for c in _CLASSES: _db_content[c] = {} create_host('fake') - create_vm('fake', 'Running', is_a_template=False, is_control_domain=True) + create_vm('fake', + 'Running', + is_a_template=False, + is_control_domain=True) + + +def reset_table(table): + if not table in _CLASSES: + return + _db_content[table] = {} def create_host(name_label): - return _create_object('host', { - 'name_label': name_label, - }) + return _create_object('host', + {'name_label': name_label}) def create_network(name_label, bridge): - return _create_object('network', { - 'name_label': name_label, - 'bridge': bridge, - }) + return _create_object('network', + {'name_label': name_label, + 'bridge': bridge}) def create_vm(name_label, status, is_a_template=False, is_control_domain=False): - return _create_object('VM', { - 'name_label': name_label, - 'power-state': status, - 'is_a_template': is_a_template, - 'is_control_domain': is_control_domain, - }) + return _create_object('VM', + {'name_label': name_label, + 'power-state': status, + 'is_a_template': is_a_template, + 'is_control_domain': is_control_domain}) def destroy_vm(vm_ref): @@ -123,27 +129,24 @@ def destroy_vdi(vdi_ref): def create_vdi(name_label, read_only, sr_ref, sharable): - return _create_object('VDI', { - 'name_label': name_label, - 'read_only': read_only, - 'SR': sr_ref, - 'type': '', - 'name_description': '', - 'sharable': sharable, - 'other_config': {}, - 'location': '', - 'xenstore_data': '', - 'sm_config': {}, - 'VBDs': {}, - }) + return _create_object('VDI', + {'name_label': name_label, + 'read_only': read_only, + 'SR': sr_ref, + 'type': '', + 'name_description': '', + 'sharable': sharable, + 'other_config': {}, + 'location': '', + 'xenstore_data': '', + 'sm_config': {}, + 'VBDs': {}}) def create_vbd(vm_ref, vdi_ref): - vbd_rec = { - 'VM': vm_ref, - 'VDI': vdi_ref, - 'currently_attached': False, - } + vbd_rec = {'VM': vm_ref, + 'VDI': vdi_ref, + 'currently_attached': False} vbd_ref = _create_object('VBD', vbd_rec) after_VBD_create(vbd_ref, vbd_rec) return vbd_ref @@ -162,20 +165,31 @@ def after_VBD_create(vbd_ref, vbd_rec): vbd_rec['vm_name_label'] = vm_name_label +def after_VM_create(vm_ref, vm_rec): + """Create read-only fields in the VM record.""" + if 'is_control_domain' not in vm_rec: + vm_rec['is_control_domain'] = False + + def create_pbd(config, host_ref, sr_ref, attached): - return _create_object('PBD', { - 'device-config': config, - 'host': host_ref, - 'SR': sr_ref, - 'currently-attached': attached, - }) + return _create_object('PBD', + {'device-config': config, + 'host': host_ref, + 'SR': sr_ref, + 'currently-attached': attached}) def create_task(name_label): - return _create_object('task', { - 'name_label': name_label, - 'status': 'pending', - }) + return _create_object('task', + {'name_label': name_label, + 'status': 'pending'}) + + +def create_local_pifs(): + """Adds a PIF for each to the local database with VLAN=-1. + Do this one per host.""" + for host_ref in _db_content['host'].keys(): + _create_local_pif(host_ref) def create_local_srs(): @@ -186,25 +200,34 @@ def create_local_srs(): def _create_local_sr(host_ref): - sr_ref = _create_object('SR', { - 'name_label': 'Local storage', - 'type': 'lvm', - 'content_type': 'user', - 'shared': False, - 'physical_size': str(1 << 30), - 'physical_utilisation': str(0), - 'virtual_allocation': str(0), - 'other_config': { - 'i18n-original-value-name_label': 'Local storage', - 'i18n-key': 'local-storage', - }, - 'VDIs': [] - }) + sr_ref = _create_object( + 'SR', + {'name_label': 'Local storage', + 'type': 'lvm', + 'content_type': 'user', + 'shared': False, + 'physical_size': str(1 << 30), + 'physical_utilisation': str(0), + 'virtual_allocation': str(0), + 'other_config': { + 'i18n-original-value-name_label': 'Local storage', + 'i18n-key': 'local-storage'}, + 'VDIs': []}) pbd_ref = create_pbd('', host_ref, sr_ref, True) _db_content['SR'][sr_ref]['PBDs'] = [pbd_ref] return sr_ref +def _create_local_pif(host_ref): + pif_ref = _create_object('PIF', + {'name-label': 'Fake PIF', + 'MAC': '00:11:22:33:44:55', + 'physical': True, + 'VLAN': -1, + 'device': 'fake0', + 'host_uuid': host_ref}) + + def _create_object(table, obj): ref = str(uuid.uuid4()) obj['uuid'] = str(uuid.uuid4()) @@ -228,6 +251,21 @@ def _create_sr(table, obj): return sr_ref +def _create_vlan(pif_ref, vlan_num, network_ref): + pif_rec = get_record('PIF', pif_ref) + vlan_pif_ref = _create_object('PIF', + {'name-label': 'Fake VLAN PIF', + 'MAC': '00:11:22:33:44:55', + 'physical': True, + 'VLAN': vlan_num, + 'device': pif_rec['device'], + 'host_uuid': pif_rec['host_uuid']}) + return _create_object('VLAN', + {'tagged-pif': pif_ref, + 'untagged-pif': vlan_pif_ref, + 'tag': vlan_num}) + + def get_all(table): return _db_content[table].keys() @@ -286,10 +324,39 @@ class SessionBase(object): rec['currently_attached'] = False rec['device'] = '' + def PIF_get_all_records_where(self, _1, _2): + # TODO (salvatore-orlando): filter table on _2 + return _db_content['PIF'] + + def VM_get_xenstore_data(self, _1, vm_ref): + return _db_content['VM'][vm_ref].get('xenstore_data', '') + + def VM_remove_from_xenstore_data(self, _1, vm_ref, key): + db_ref = _db_content['VM'][vm_ref] + if not 'xenstore_data' in db_ref: + return + db_ref['xenstore_data'][key] = None + + def network_get_all_records_where(self, _1, _2): + # TODO (salvatore-orlando): filter table on _2 + return _db_content['network'] + + def VM_add_to_xenstore_data(self, _1, vm_ref, key, value): + db_ref = _db_content['VM'][vm_ref] + if not 'xenstore_data' in db_ref: + db_ref['xenstore_data'] = {} + db_ref['xenstore_data'][key] = value + def host_compute_free_memory(self, _1, ref): #Always return 12GB available return 12 * 1024 * 1024 * 1024 + def host_call_plugin(*args): + return 'herp' + + def network_get_all_records_where(self, _1, filter): + return self.xenapi.network.get_all_records() + def xenapi_request(self, methodname, params): if methodname.startswith('login'): self._login(methodname, params) @@ -309,10 +376,9 @@ class SessionBase(object): def _login(self, method, params): self._session = str(uuid.uuid4()) - _db_content['session'][self._session] = { - 'uuid': str(uuid.uuid4()), - 'this_host': _db_content['host'].keys()[0], - } + _db_content['session'][self._session] = \ + {'uuid': str(uuid.uuid4()), + 'this_host': _db_content['host'].keys()[0]} def _logout(self): s = self._session @@ -373,7 +439,6 @@ class SessionBase(object): def _getter(self, name, params): self._check_session(params) (cls, func) = name.split('.') - if func == 'get_all': self._check_arg_count(params, 1) return get_all(cls) @@ -396,12 +461,13 @@ class SessionBase(object): if len(params) == 2: field = func[len('get_'):] ref = params[1] + if (ref in _db_content[cls]): + if (field in _db_content[cls][ref]): + return _db_content[cls][ref][field] + else: + raise Failure(['HANDLE_INVALID', cls, ref]) - if (ref in _db_content[cls] and - field in _db_content[cls][ref]): - return _db_content[cls][ref][field] - - LOG.debuug(_('Raising NotImplemented')) + LOG.debug(_('Raising NotImplemented')) raise NotImplementedError( _('xenapi.fake does not have an implementation for %s or it has ' 'been called with the wrong number of arguments') % name) @@ -428,12 +494,16 @@ class SessionBase(object): def _create(self, name, params): self._check_session(params) is_sr_create = name == 'SR.create' + is_vlan_create = name == 'VLAN.create' # Storage Repositories have a different API - expected = is_sr_create and 10 or 2 + expected = is_sr_create and 10 or is_vlan_create and 4 or 2 self._check_arg_count(params, expected) (cls, _) = name.split('.') ref = is_sr_create and \ - _create_sr(cls, params) or _create_object(cls, params[1]) + _create_sr(cls, params) or \ + is_vlan_create and \ + _create_vlan(params[1], params[2], params[3]) or \ + _create_object(cls, params[1]) # Call hook to provide any fixups needed (ex. creating backrefs) after_hook = 'after_%s_create' % cls @@ -473,7 +543,7 @@ class SessionBase(object): def _check_session(self, params): if (self._session is None or self._session not in _db_content['session']): - raise Failure(['HANDLE_INVALID', 'session', self._session]) + raise Failure(['HANDLE_INVALID', 'session', self._session]) if len(params) == 0 or params[0] != self._session: LOG.debug(_('Raising NotImplemented')) raise NotImplementedError('Call to XenAPI without using .xenapi') diff --git a/nova/virt/xenapi/network_utils.py b/nova/virt/xenapi/network_utils.py index c0406d8f0..94d8e5199 100644 --- a/nova/virt/xenapi/network_utils.py +++ b/nova/virt/xenapi/network_utils.py @@ -28,11 +28,26 @@ class NetworkHelper(HelperBase): """ The class that wraps the helper methods together. """ + @classmethod + def find_network_with_name_label(cls, session, name_label): + networks = session.call_xenapi('network.get_by_name_label', name_label) + if len(networks) == 1: + return networks[0] + elif len(networks) > 1: + raise Exception(_('Found non-unique network' + ' for name_label %s') % name_label) + else: + return None @classmethod def find_network_with_bridge(cls, session, bridge): - """Return the network on which the bridge is attached, if found.""" - expr = 'field "bridge" = "%s"' % bridge + """ + Return the network on which the bridge is attached, if found. + The bridge is defined in the nova db and can be found either in the + 'bridge' or 'name_label' fields of the XenAPI network record. + """ + expr = 'field "name__label" = "%s" or ' \ + 'field "bridge" = "%s"' % (bridge, bridge) networks = session.call_xenapi('network.get_all_records_where', expr) if len(networks) == 1: return networks.keys()[0] diff --git a/nova/virt/xenapi/vm_utils.py b/nova/virt/xenapi/vm_utils.py index 80cc3035d..d07d60800 100644 --- a/nova/virt/xenapi/vm_utils.py +++ b/nova/virt/xenapi/vm_utils.py @@ -22,12 +22,16 @@ their attributes like VDIs, VIFs, as well as their lookup functions. import os import pickle import re +import tempfile import time import urllib +import uuid from xml.dom import minidom from eventlet import event import glance.client +from nova import context +from nova import db from nova import exception from nova import flags from nova import log as logging @@ -35,14 +39,17 @@ from nova import utils from nova.auth.manager import AuthManager from nova.compute import instance_types from nova.compute import power_state +from nova.virt import disk from nova.virt import images from nova.virt.xenapi import HelperBase from nova.virt.xenapi.volume_utils import StorageError -FLAGS = flags.FLAGS LOG = logging.getLogger("nova.virt.xenapi.vm_utils") +FLAGS = flags.FLAGS +flags.DEFINE_string('default_os_type', 'linux', 'Default OS type') + XENAPI_POWER_STATE = { 'Halted': power_state.SHUTDOWN, 'Running': power_state.RUNNING, @@ -63,11 +70,14 @@ class ImageType: 0 - kernel/ramdisk image (goes on dom0's filesystem) 1 - disk image (local SR, partitioned by objectstore plugin) 2 - raw disk image (local SR, NOT partitioned by plugin) + 3 - vhd disk image (local SR, NOT inspected by XS, PV assumed for + linux, HVM assumed for Windows) """ KERNEL_RAMDISK = 0 DISK = 1 DISK_RAW = 2 + DISK_VHD = 3 class VMHelper(HelperBase): @@ -76,62 +86,81 @@ class VMHelper(HelperBase): """ @classmethod - def create_vm(cls, session, instance, kernel, ramdisk, pv_kernel=False): + def create_vm(cls, session, instance, kernel, ramdisk, + use_pv_kernel=False): """Create a VM record. Returns a Deferred that gives the new VM reference. - the pv_kernel flag indicates whether the guest is HVM or PV + the use_pv_kernel flag indicates whether the guest is HVM or PV + + There are 3 scenarios: + + 1. Using paravirtualization, kernel passed in + + 2. Using paravirtualization, kernel within the image + + 3. Using hardware virtualization """ - instance_type = instance_types.INSTANCE_TYPES[instance.instance_type] + instance_type = instance_types.\ + get_instance_type(instance.instance_type) mem = str(long(instance_type['memory_mb']) * 1024 * 1024) vcpus = str(instance_type['vcpus']) rec = { - 'name_label': instance.name, - 'name_description': '', + 'actions_after_crash': 'destroy', + 'actions_after_reboot': 'restart', + 'actions_after_shutdown': 'destroy', + 'affinity': '', + 'blocked_operations': {}, + 'ha_always_run': False, + 'ha_restart_priority': '', + 'HVM_boot_params': {}, + 'HVM_boot_policy': '', 'is_a_template': False, - 'memory_static_min': '0', - 'memory_static_max': mem, 'memory_dynamic_min': mem, 'memory_dynamic_max': mem, - 'VCPUs_at_startup': vcpus, - 'VCPUs_max': vcpus, - 'VCPUs_params': {}, - 'actions_after_shutdown': 'destroy', - 'actions_after_reboot': 'restart', - 'actions_after_crash': 'destroy', - 'PV_bootloader': '', - 'PV_kernel': '', - 'PV_ramdisk': '', + 'memory_static_min': '0', + 'memory_static_max': mem, + 'memory_target': mem, + 'name_description': '', + 'name_label': instance.name, + 'other_config': {'allowvssprovider': False}, + 'other_config': {}, + 'PCI_bus': '', + 'platform': {'acpi': 'true', 'apic': 'true', 'pae': 'true', + 'viridian': 'true', 'timeoffset': '0'}, 'PV_args': '', + 'PV_bootloader': '', 'PV_bootloader_args': '', + 'PV_kernel': '', 'PV_legacy_args': '', - 'HVM_boot_policy': '', - 'HVM_boot_params': {}, - 'platform': {}, - 'PCI_bus': '', + 'PV_ramdisk': '', 'recommendations': '', - 'affinity': '', + 'tags': [], 'user_version': '0', - 'other_config': {}, - } - #Complete VM configuration record according to the image type - #non-raw/raw with PV kernel/raw in HVM mode - if instance.kernel_id: - rec['PV_bootloader'] = '' - rec['PV_kernel'] = kernel - rec['PV_ramdisk'] = ramdisk - rec['PV_args'] = 'root=/dev/xvda1' - rec['PV_bootloader_args'] = '' - rec['PV_legacy_args'] = '' - else: - if pv_kernel: - rec['PV_args'] = 'noninteractive' - rec['PV_bootloader'] = 'pygrub' + 'VCPUs_at_startup': vcpus, + 'VCPUs_max': vcpus, + 'VCPUs_params': {}, + 'xenstore_data': {}} + + # Complete VM configuration record according to the image type + # non-raw/raw with PV kernel/raw in HVM mode + if use_pv_kernel: + rec['platform']['nx'] = 'false' + if instance.kernel_id: + # 1. Kernel explicitly passed in, use that + rec['PV_args'] = 'root=/dev/xvda1' + rec['PV_kernel'] = kernel + rec['PV_ramdisk'] = ramdisk else: - rec['HVM_boot_policy'] = 'BIOS order' - rec['HVM_boot_params'] = {'order': 'dc'} - rec['platform'] = {'acpi': 'true', 'apic': 'true', - 'pae': 'true', 'viridian': 'true'} + # 2. Use kernel within the image + rec['PV_args'] = 'clocksource=jiffies' + rec['PV_bootloader'] = 'pygrub' + else: + # 3. Using hardware virtualization + rec['platform']['nx'] = 'true' + rec['HVM_boot_params'] = {'order': 'dc'} + rec['HVM_boot_policy'] = 'BIOS order' + LOG.debug(_('Created VM %s...'), instance.name) vm_ref = session.call_xenapi('VM.create', rec) instance_name = instance.name @@ -140,7 +169,8 @@ class VMHelper(HelperBase): @classmethod def ensure_free_mem(cls, session, instance): - instance_type = instance_types.INSTANCE_TYPES[instance.instance_type] + instance_type = instance_types.get_instance_type( + instance.instance_type) mem = long(instance_type['memory_mb']) * 1024 * 1024 #get free memory from host host = session.get_xenapi_host() @@ -175,13 +205,13 @@ class VMHelper(HelperBase): @classmethod def find_vbd_by_number(cls, session, vm_ref, number): """Get the VBD reference from the device number""" - vbds = session.get_xenapi().VM.get_VBDs(vm_ref) - if vbds: - for vbd in vbds: + vbd_refs = session.get_xenapi().VM.get_VBDs(vm_ref) + if vbd_refs: + for vbd_ref in vbd_refs: try: - vbd_rec = session.get_xenapi().VBD.get_record(vbd) + vbd_rec = session.get_xenapi().VBD.get_record(vbd_ref) if vbd_rec['userdevice'] == str(number): - return vbd + return vbd_ref except cls.XenAPI.Failure, exc: LOG.exception(exc) raise StorageError(_('VBD not found in instance %s') % vm_ref) @@ -201,26 +231,26 @@ class VMHelper(HelperBase): """Destroy VBD from host database""" try: task = session.call_xenapi('Async.VBD.destroy', vbd_ref) - #FIXME(armando): find a solution to missing instance_id - #with Josh Kearney - session.wait_for_task(0, task) + session.wait_for_task(task) except cls.XenAPI.Failure, exc: LOG.exception(exc) raise StorageError(_('Unable to destroy VBD %s') % vbd_ref) @classmethod - def create_vif(cls, session, vm_ref, network_ref, mac_address): + def create_vif(cls, session, vm_ref, network_ref, mac_address, + dev, rxtx_cap=0): """Create a VIF record. Returns a Deferred that gives the new VIF reference.""" vif_rec = {} - vif_rec['device'] = '0' + vif_rec['device'] = str(dev) vif_rec['network'] = network_ref vif_rec['VM'] = vm_ref vif_rec['MAC'] = mac_address vif_rec['MTU'] = '1500' vif_rec['other_config'] = {} - vif_rec['qos_algorithm_type'] = '' - vif_rec['qos_algorithm_params'] = {} + vif_rec['qos_algorithm_type'] = "ratelimit" if rxtx_cap else '' + vif_rec['qos_algorithm_params'] = \ + {"kbps": str(rxtx_cap * 1024)} if rxtx_cap else {} LOG.debug(_('Creating VIF for VM %(vm_ref)s,' ' network %(network_ref)s.') % locals()) vif_ref = session.call_xenapi('VIF.create', vif_rec) @@ -249,24 +279,40 @@ class VMHelper(HelperBase): return vdi_ref @classmethod + def get_vdi_for_vm_safely(cls, session, vm_ref): + vdi_refs = VMHelper.lookup_vm_vdis(session, vm_ref) + if vdi_refs is None: + raise Exception(_("No VDIs found for VM %s") % vm_ref) + else: + num_vdis = len(vdi_refs) + if num_vdis != 1: + raise Exception( + _("Unexpected number of VDIs (%(num_vdis)s) found" + " for VM %(vm_ref)s") % locals()) + + vdi_ref = vdi_refs[0] + vdi_rec = session.get_xenapi().VDI.get_record(vdi_ref) + return vdi_ref, vdi_rec + + @classmethod def create_snapshot(cls, session, instance_id, vm_ref, label): - """ Creates Snapshot (Template) VM, Snapshot VBD, Snapshot VDI, - Snapshot VHD - """ + """Creates Snapshot (Template) VM, Snapshot VBD, Snapshot VDI, + Snapshot VHD""" #TODO(sirp): Add quiesce and VSS locking support when Windows support # is added LOG.debug(_("Snapshotting VM %(vm_ref)s with label '%(label)s'...") % locals()) - vm_vdi_ref, vm_vdi_rec = get_vdi_for_vm_safely(session, vm_ref) + vm_vdi_ref, vm_vdi_rec = cls.get_vdi_for_vm_safely(session, vm_ref) vm_vdi_uuid = vm_vdi_rec["uuid"] sr_ref = vm_vdi_rec["SR"] original_parent_uuid = get_vhd_parent_uuid(session, vm_vdi_ref) task = session.call_xenapi('Async.VM.snapshot', vm_ref, label) - template_vm_ref = session.wait_for_task(instance_id, task) - template_vdi_rec = get_vdi_for_vm_safely(session, template_vm_ref)[1] + template_vm_ref = session.wait_for_task(task, instance_id) + template_vdi_rec = cls.get_vdi_for_vm_safely(session, + template_vm_ref)[1] template_vdi_uuid = template_vdi_rec["uuid"] LOG.debug(_('Created snapshot %(template_vm_ref)s from' @@ -276,29 +322,56 @@ class VMHelper(HelperBase): session, instance_id, sr_ref, vm_vdi_ref, original_parent_uuid) #TODO(sirp): we need to assert only one parent, not parents two deep - return template_vm_ref, [template_vdi_uuid, parent_uuid] + template_vdi_uuids = {'image': parent_uuid, + 'snap': template_vdi_uuid} + return template_vm_ref, template_vdi_uuids @classmethod - def upload_image(cls, session, instance_id, vdi_uuids, image_id): + def get_sr(cls, session, sr_label='slices'): + """Finds the SR named by the given name label and returns + the UUID""" + return session.call_xenapi('SR.get_by_name_label', sr_label)[0] + + @classmethod + def get_sr_path(cls, session): + """Return the path to our storage repository + + This is used when we're dealing with VHDs directly, either by taking + snapshots or by restoring an image in the DISK_VHD format. + """ + sr_ref = safe_find_sr(session) + sr_rec = session.get_xenapi().SR.get_record(sr_ref) + sr_uuid = sr_rec["uuid"] + return os.path.join(FLAGS.xenapi_sr_base_path, sr_uuid) + + @classmethod + def upload_image(cls, session, instance, vdi_uuids, image_id): """ Requests that the Glance plugin bundle the specified VDIs and push them into Glance using the specified human-friendly name. """ + # NOTE(sirp): Currently we only support uploading images as VHD, there + # is no RAW equivalent (yet) logging.debug(_("Asking xapi to upload %(vdi_uuids)s as" " ID %(image_id)s") % locals()) + os_type = instance.os_type or FLAGS.default_os_type + params = {'vdi_uuids': vdi_uuids, 'image_id': image_id, 'glance_host': FLAGS.glance_host, - 'glance_port': FLAGS.glance_port} + 'glance_port': FLAGS.glance_port, + 'sr_path': cls.get_sr_path(session), + 'os_type': os_type} kwargs = {'params': pickle.dumps(params)} - task = session.async_call_plugin('glance', 'put_vdis', kwargs) - session.wait_for_task(instance_id, task) + task = session.async_call_plugin('glance', 'upload_vhd', kwargs) + session.wait_for_task(task, instance.id) @classmethod - def fetch_image(cls, session, instance_id, image, user, project, type): + def fetch_image(cls, session, instance_id, image, user, project, + image_type): """ - type is interpreted as an ImageType instance + image_type is interpreted as an ImageType instance Related flags: xenapi_image_service = ['glance', 'objectstore'] glance_address = 'address for glance services' @@ -308,87 +381,220 @@ class VMHelper(HelperBase): if FLAGS.xenapi_image_service == 'glance': return cls._fetch_image_glance(session, instance_id, image, - access, type) + access, image_type) else: return cls._fetch_image_objectstore(session, instance_id, image, - access, user.secret, type) + access, user.secret, + image_type) + + @classmethod + def _fetch_image_glance_vhd(cls, session, instance_id, image, access, + image_type): + LOG.debug(_("Asking xapi to fetch vhd image %(image)s") + % locals()) + + sr_ref = safe_find_sr(session) + + # NOTE(sirp): The Glance plugin runs under Python 2.4 which does not + # have the `uuid` module. To work around this, we generate the uuids + # here (under Python 2.6+) and pass them as arguments + uuid_stack = [str(uuid.uuid4()) for i in xrange(2)] + + params = {'image_id': image, + 'glance_host': FLAGS.glance_host, + 'glance_port': FLAGS.glance_port, + 'uuid_stack': uuid_stack, + 'sr_path': cls.get_sr_path(session)} + + kwargs = {'params': pickle.dumps(params)} + task = session.async_call_plugin('glance', 'download_vhd', kwargs) + vdi_uuid = session.wait_for_task(task, instance_id) + + cls.scan_sr(session, instance_id, sr_ref) + + # Set the name-label to ease debugging + vdi_ref = session.get_xenapi().VDI.get_by_uuid(vdi_uuid) + name_label = get_name_label_for_image(image) + session.get_xenapi().VDI.set_name_label(vdi_ref, name_label) + + LOG.debug(_("xapi 'download_vhd' returned VDI UUID %(vdi_uuid)s") + % locals()) + return vdi_uuid @classmethod - def _fetch_image_glance(cls, session, instance_id, image, access, type): - sr = find_sr(session) - if sr is None: - raise exception.NotFound('Cannot find SR to write VDI to') + def _fetch_image_glance_disk(cls, session, instance_id, image, access, + image_type): + """Fetch the image from Glance + + NOTE: + Unlike _fetch_image_glance_vhd, this method does not use the Glance + plugin; instead, it streams the disks through domU to the VDI + directly. - c = glance.client.Client(FLAGS.glance_host, FLAGS.glance_port) + """ + # FIXME(sirp): Since the Glance plugin seems to be required for the + # VHD disk, it may be worth using the plugin for both VHD and RAW and + # DISK restores + sr_ref = safe_find_sr(session) - meta, image_file = c.get_image(image) + client = glance.client.Client(FLAGS.glance_host, FLAGS.glance_port) + meta, image_file = client.get_image(image) virtual_size = int(meta['size']) vdi_size = virtual_size LOG.debug(_("Size for image %(image)s:%(virtual_size)d") % locals()) - if type == ImageType.DISK: + + if image_type == ImageType.DISK: # Make room for MBR. vdi_size += MBR_SIZE_BYTES - vdi = cls.create_vdi(session, sr, _('Glance image %s') % image, - vdi_size, False) + name_label = get_name_label_for_image(image) + vdi_ref = cls.create_vdi(session, sr_ref, name_label, vdi_size, False) - with_vdi_attached_here(session, vdi, False, + with_vdi_attached_here(session, vdi_ref, False, lambda dev: - _stream_disk(dev, type, + _stream_disk(dev, image_type, virtual_size, image_file)) - if (type == ImageType.KERNEL_RAMDISK): + if image_type == ImageType.KERNEL_RAMDISK: #we need to invoke a plugin for copying VDI's #content into proper path - LOG.debug(_("Copying VDI %s to /boot/guest on dom0"), vdi) + LOG.debug(_("Copying VDI %s to /boot/guest on dom0"), vdi_ref) fn = "copy_kernel_vdi" args = {} - args['vdi-ref'] = vdi + args['vdi-ref'] = vdi_ref #let the plugin copy the correct number of bytes args['image-size'] = str(vdi_size) task = session.async_call_plugin('glance', fn, args) - filename = session.wait_for_task(instance_id, task) + filename = session.wait_for_task(task, instance_id) #remove the VDI as it is not needed anymore - session.get_xenapi().VDI.destroy(vdi) - LOG.debug(_("Kernel/Ramdisk VDI %s destroyed"), vdi) + session.get_xenapi().VDI.destroy(vdi_ref) + LOG.debug(_("Kernel/Ramdisk VDI %s destroyed"), vdi_ref) return filename else: - return session.get_xenapi().VDI.get_uuid(vdi) + return session.get_xenapi().VDI.get_uuid(vdi_ref) + + @classmethod + def determine_disk_image_type(cls, instance): + """Disk Image Types are used to determine where the kernel will reside + within an image. To figure out which type we're dealing with, we use + the following rules: + + 1. If we're using Glance, we can use the image_type field to + determine the image_type + + 2. If we're not using Glance, then we need to deduce this based on + whether a kernel_id is specified. + """ + def log_disk_format(image_type): + pretty_format = {ImageType.KERNEL_RAMDISK: 'KERNEL_RAMDISK', + ImageType.DISK: 'DISK', + ImageType.DISK_RAW: 'DISK_RAW', + ImageType.DISK_VHD: 'DISK_VHD'} + disk_format = pretty_format[image_type] + image_id = instance.image_id + instance_id = instance.id + LOG.debug(_("Detected %(disk_format)s format for image " + "%(image_id)s, instance %(instance_id)s") % locals()) + + def determine_from_glance(): + glance_disk_format2nova_type = { + 'ami': ImageType.DISK, + 'aki': ImageType.KERNEL_RAMDISK, + 'ari': ImageType.KERNEL_RAMDISK, + 'raw': ImageType.DISK_RAW, + 'vhd': ImageType.DISK_VHD} + client = glance.client.Client(FLAGS.glance_host, FLAGS.glance_port) + meta = client.get_image_meta(instance.image_id) + disk_format = meta['disk_format'] + try: + return glance_disk_format2nova_type[disk_format] + except KeyError: + raise exception.NotFound( + _("Unrecognized disk_format '%(disk_format)s'") + % locals()) + + def determine_from_instance(): + if instance.kernel_id: + return ImageType.DISK + else: + return ImageType.DISK_RAW + + # FIXME(sirp): can we unify the ImageService and xenapi_image_service + # abstractions? + if FLAGS.xenapi_image_service == 'glance': + image_type = determine_from_glance() + else: + image_type = determine_from_instance() + + log_disk_format(image_type) + return image_type + + @classmethod + def _fetch_image_glance(cls, session, instance_id, image, access, + image_type): + if image_type == ImageType.DISK_VHD: + return cls._fetch_image_glance_vhd( + session, instance_id, image, access, image_type) + else: + return cls._fetch_image_glance_disk( + session, instance_id, image, access, image_type) @classmethod def _fetch_image_objectstore(cls, session, instance_id, image, access, - secret, type): + secret, image_type): url = images.image_url(image) LOG.debug(_("Asking xapi to fetch %(url)s as %(access)s") % locals()) - fn = (type != ImageType.KERNEL_RAMDISK) and 'get_vdi' or 'get_kernel' + if image_type == ImageType.KERNEL_RAMDISK: + fn = 'get_kernel' + else: + fn = 'get_vdi' args = {} args['src_url'] = url args['username'] = access args['password'] = secret args['add_partition'] = 'false' args['raw'] = 'false' - if type != ImageType.KERNEL_RAMDISK: + if image_type != ImageType.KERNEL_RAMDISK: args['add_partition'] = 'true' - if type == ImageType.DISK_RAW: + if image_type == ImageType.DISK_RAW: args['raw'] = 'true' task = session.async_call_plugin('objectstore', fn, args) - uuid = session.wait_for_task(instance_id, task) + uuid = session.wait_for_task(task, instance_id) return uuid @classmethod - def lookup_image(cls, session, instance_id, vdi_ref): + def determine_is_pv(cls, session, instance_id, vdi_ref, disk_image_type, + os_type): + """ + Determine whether the VM will use a paravirtualized kernel or if it + will use hardware virtualization. + + 1. Objectstore (any image type): + We use plugin to figure out whether the VDI uses PV + + 2. Glance (VHD): then we use `os_type`, raise if not set + + 3. Glance (DISK_RAW): use Pygrub to figure out if pv kernel is + available + + 4. Glance (DISK): pv is assumed + """ if FLAGS.xenapi_image_service == 'glance': - return cls._lookup_image_glance(session, vdi_ref) + # 2, 3, 4: Glance + return cls._determine_is_pv_glance( + session, vdi_ref, disk_image_type, os_type) else: - return cls._lookup_image_objectstore(session, instance_id, vdi_ref) + # 1. Objecstore + return cls._determine_is_pv_objectstore(session, instance_id, + vdi_ref) @classmethod - def _lookup_image_objectstore(cls, session, instance_id, vdi_ref): + def _determine_is_pv_objectstore(cls, session, instance_id, vdi_ref): LOG.debug(_("Looking up vdi %s for PV kernel"), vdi_ref) fn = "is_vdi_pv" args = {} args['vdi-ref'] = vdi_ref task = session.async_call_plugin('objectstore', fn, args) - pv_str = session.wait_for_task(instance_id, task) + pv_str = session.wait_for_task(task, instance_id) pv = None if pv_str.lower() == 'true': pv = True @@ -398,58 +604,93 @@ class VMHelper(HelperBase): return pv @classmethod - def _lookup_image_glance(cls, session, vdi_ref): + def _determine_is_pv_glance(cls, session, vdi_ref, disk_image_type, + os_type): + """ + For a Glance image, determine if we need paravirtualization. + + The relevant scenarios are: + 2. Glance (VHD): then we use `os_type`, raise if not set + + 3. Glance (DISK_RAW): use Pygrub to figure out if pv kernel is + available + + 4. Glance (DISK): pv is assumed + """ + LOG.debug(_("Looking up vdi %s for PV kernel"), vdi_ref) + if disk_image_type == ImageType.DISK_VHD: + # 2. VHD + if os_type == 'windows': + is_pv = False + else: + is_pv = True + elif disk_image_type == ImageType.DISK_RAW: + # 3. RAW + is_pv = with_vdi_attached_here(session, vdi_ref, True, _is_vdi_pv) + elif disk_image_type == ImageType.DISK: + # 4. Disk + is_pv = True + else: + raise exception.Error(_("Unknown image format %(disk_image_type)s") + % locals()) - def is_vdi_pv(dev): - LOG.debug(_("Running pygrub against %s"), dev) - output = os.popen('pygrub -qn /dev/%s' % dev) - for line in output.readlines(): - #try to find kernel string - m = re.search('(?<=kernel:)/.*(?:>)', line) - if m and m.group(0).find('xen') != -1: - LOG.debug(_("Found Xen kernel %s") % m.group(0)) - return True - LOG.debug(_("No Xen kernel found. Booting HVM.")) - return False - return with_vdi_attached_here(session, vdi_ref, True, is_vdi_pv) + return is_pv @classmethod - def lookup(cls, session, i): + def lookup(cls, session, name_label): """Look the instance i up, and returns it if available""" - vms = session.get_xenapi().VM.get_by_name_label(i) - n = len(vms) + vm_refs = session.get_xenapi().VM.get_by_name_label(name_label) + n = len(vm_refs) if n == 0: return None elif n > 1: - raise exception.Duplicate(_('duplicate name found: %s') % i) + raise exception.Duplicate(_('duplicate name found: %s') % + name_label) else: - return vms[0] + return vm_refs[0] @classmethod - def lookup_vm_vdis(cls, session, vm): + def lookup_vm_vdis(cls, session, vm_ref): """Look for the VDIs that are attached to the VM""" # Firstly we get the VBDs, then the VDIs. # TODO(Armando): do we leave the read-only devices? - vbds = session.get_xenapi().VM.get_VBDs(vm) - vdis = [] - if vbds: - for vbd in vbds: + vbd_refs = session.get_xenapi().VM.get_VBDs(vm_ref) + vdi_refs = [] + if vbd_refs: + for vbd_ref in vbd_refs: try: - vdi = session.get_xenapi().VBD.get_VDI(vbd) + vdi_ref = session.get_xenapi().VBD.get_VDI(vbd_ref) # Test valid VDI - record = session.get_xenapi().VDI.get_record(vdi) + record = session.get_xenapi().VDI.get_record(vdi_ref) LOG.debug(_('VDI %s is still available'), record['uuid']) except cls.XenAPI.Failure, exc: LOG.exception(exc) else: - vdis.append(vdi) - if len(vdis) > 0: - return vdis + vdi_refs.append(vdi_ref) + if len(vdi_refs) > 0: + return vdi_refs else: return None @classmethod + def preconfigure_instance(cls, session, instance, vdi_ref, network_info): + """Makes alterations to the image before launching as part of spawn. + """ + + # As mounting the image VDI is expensive, we only want do do it once, + # if at all, so determine whether it's required first, and then do + # everything + mount_required = False + key, net = _prepare_injectables(instance, network_info) + mount_required = key or net + if not mount_required: + return + + with_vdi_attached_here(session, vdi_ref, False, + lambda dev: _mounted_processing(dev, key, net)) + + @classmethod def lookup_kernel_ramdisk(cls, session, vm): vm_rec = session.get_xenapi().VM.get_record(vm) if 'PV_kernel' in vm_rec and 'PV_ramdisk' in vm_rec: @@ -496,6 +737,21 @@ class VMHelper(HelperBase): except cls.XenAPI.Failure as e: return {"Unable to retrieve diagnostics": e} + @classmethod + def scan_sr(cls, session, instance_id=None, sr_ref=None): + """Scans the SR specified by sr_ref""" + if sr_ref: + LOG.debug(_("Re-scanning SR %s"), sr_ref) + task = session.call_xenapi('Async.SR.scan', sr_ref) + session.wait_for_task(task, instance_id) + + @classmethod + def scan_default_sr(cls, session): + """Looks for the system default SR and triggers a re-scan""" + #FIXME(sirp/mdietz): refactor scan_default_sr in there + sr_ref = cls.get_sr(session) + session.call_xenapi('SR.scan', sr_ref) + def get_rrd(host, uuid): """Return the VM RRD XML as a string""" @@ -538,12 +794,6 @@ def get_vhd_parent_uuid(session, vdi_ref): return None -def scan_sr(session, instance_id, sr_ref): - LOG.debug(_("Re-scanning SR %s"), sr_ref) - task = session.call_xenapi('Async.SR.scan', sr_ref) - session.wait_for_task(instance_id, task) - - def wait_for_vhd_coalesce(session, instance_id, sr_ref, vdi_ref, original_parent_uuid): """ Spin until the parent VHD is coalesced into its parent VHD @@ -568,7 +818,7 @@ def wait_for_vhd_coalesce(session, instance_id, sr_ref, vdi_ref, " %(max_attempts)d), giving up...") % locals()) raise exception.Error(msg) - scan_sr(session, instance_id, sr_ref) + VMHelper.scan_sr(session, instance_id, sr_ref) parent_uuid = get_vhd_parent_uuid(session, vdi_ref) if original_parent_uuid and (parent_uuid != original_parent_uuid): LOG.debug(_("Parent %(parent_uuid)s doesn't match original parent" @@ -599,18 +849,29 @@ def get_vdi_for_vm_safely(session, vm_ref): return vdi_ref, vdi_rec +def safe_find_sr(session): + """Same as find_sr except raises a NotFound exception if SR cannot be + determined + """ + sr_ref = find_sr(session) + if sr_ref is None: + raise exception.NotFound(_('Cannot find SR to read/write VDI')) + return sr_ref + + def find_sr(session): + """Return the storage repository to hold VM images""" host = session.get_xenapi_host() - srs = session.get_xenapi().SR.get_all() - for sr in srs: - sr_rec = session.get_xenapi().SR.get_record(sr) + sr_refs = session.get_xenapi().SR.get_all() + for sr_ref in sr_refs: + sr_rec = session.get_xenapi().SR.get_record(sr_ref) if not ('i18n-key' in sr_rec['other_config'] and sr_rec['other_config']['i18n-key'] == 'local-storage'): continue - for pbd in sr_rec['PBDs']: - pbd_rec = session.get_xenapi().PBD.get_record(pbd) + for pbd_ref in sr_rec['PBDs']: + pbd_rec = session.get_xenapi().PBD.get_record(pbd_ref) if pbd_rec['host'] == host: - return sr + return sr_ref return None @@ -635,11 +896,11 @@ def remap_vbd_dev(dev): return remapped_dev -def with_vdi_attached_here(session, vdi, read_only, f): +def with_vdi_attached_here(session, vdi_ref, read_only, f): this_vm_ref = get_this_vm_ref(session) vbd_rec = {} vbd_rec['VM'] = this_vm_ref - vbd_rec['VDI'] = vdi + vbd_rec['VDI'] = vdi_ref vbd_rec['userdevice'] = 'autodetect' vbd_rec['bootable'] = False vbd_rec['mode'] = read_only and 'RO' or 'RW' @@ -650,28 +911,28 @@ def with_vdi_attached_here(session, vdi, read_only, f): vbd_rec['qos_algorithm_type'] = '' vbd_rec['qos_algorithm_params'] = {} vbd_rec['qos_supported_algorithms'] = [] - LOG.debug(_('Creating VBD for VDI %s ... '), vdi) - vbd = session.get_xenapi().VBD.create(vbd_rec) - LOG.debug(_('Creating VBD for VDI %s done.'), vdi) + LOG.debug(_('Creating VBD for VDI %s ... '), vdi_ref) + vbd_ref = session.get_xenapi().VBD.create(vbd_rec) + LOG.debug(_('Creating VBD for VDI %s done.'), vdi_ref) try: - LOG.debug(_('Plugging VBD %s ... '), vbd) - session.get_xenapi().VBD.plug(vbd) - LOG.debug(_('Plugging VBD %s done.'), vbd) - orig_dev = session.get_xenapi().VBD.get_device(vbd) - LOG.debug(_('VBD %(vbd)s plugged as %(orig_dev)s') % locals()) + LOG.debug(_('Plugging VBD %s ... '), vbd_ref) + session.get_xenapi().VBD.plug(vbd_ref) + LOG.debug(_('Plugging VBD %s done.'), vbd_ref) + orig_dev = session.get_xenapi().VBD.get_device(vbd_ref) + LOG.debug(_('VBD %(vbd_ref)s plugged as %(orig_dev)s') % locals()) dev = remap_vbd_dev(orig_dev) if dev != orig_dev: - LOG.debug(_('VBD %(vbd)s plugged into wrong dev, ' + LOG.debug(_('VBD %(vbd_ref)s plugged into wrong dev, ' 'remapping to %(dev)s') % locals()) return f(dev) finally: - LOG.debug(_('Destroying VBD for VDI %s ... '), vdi) - vbd_unplug_with_retry(session, vbd) - ignore_failure(session.get_xenapi().VBD.destroy, vbd) - LOG.debug(_('Destroying VBD for VDI %s done.'), vdi) + LOG.debug(_('Destroying VBD for VDI %s ... '), vdi_ref) + vbd_unplug_with_retry(session, vbd_ref) + ignore_failure(session.get_xenapi().VBD.destroy, vbd_ref) + LOG.debug(_('Destroying VBD for VDI %s done.'), vdi_ref) -def vbd_unplug_with_retry(session, vbd): +def vbd_unplug_with_retry(session, vbd_ref): """Call VBD.unplug on the given VBD, with a retry if we get DEVICE_DETACH_REJECTED. For reasons which I don't understand, we're seeing the device still in use, even when all processes using the device @@ -679,7 +940,7 @@ def vbd_unplug_with_retry(session, vbd): # FIXME(sirp): We can use LoopingCall here w/o blocking sleep() while True: try: - session.get_xenapi().VBD.unplug(vbd) + session.get_xenapi().VBD.unplug(vbd_ref) LOG.debug(_('VBD.unplug successful first time.')) return except VMHelper.XenAPI.Failure, e: @@ -687,6 +948,7 @@ def vbd_unplug_with_retry(session, vbd): e.details[0] == 'DEVICE_DETACH_REJECTED'): LOG.debug(_('VBD.unplug rejected: retrying...')) time.sleep(1) + LOG.debug(_('Not sleeping anymore!')) elif (len(e.details) > 0 and e.details[0] == 'DEVICE_ALREADY_DETACHED'): LOG.debug(_('VBD.unplug successful eventually.')) @@ -714,9 +976,22 @@ def get_this_vm_ref(session): return session.get_xenapi().VM.get_by_uuid(get_this_vm_uuid()) -def _stream_disk(dev, type, virtual_size, image_file): +def _is_vdi_pv(dev): + LOG.debug(_("Running pygrub against %s"), dev) + output = os.popen('pygrub -qn /dev/%s' % dev) + for line in output.readlines(): + #try to find kernel string + m = re.search('(?<=kernel:)/.*(?:>)', line) + if m and m.group(0).find('xen') != -1: + LOG.debug(_("Found Xen kernel %s") % m.group(0)) + return True + LOG.debug(_("No Xen kernel found. Booting HVM.")) + return False + + +def _stream_disk(dev, image_type, virtual_size, image_file): offset = 0 - if type == ImageType.DISK: + if image_type == ImageType.DISK: offset = MBR_SIZE_BYTES _write_partition(virtual_size, dev) @@ -735,13 +1010,132 @@ def _write_partition(virtual_size, dev): LOG.debug(_('Writing partition table %(primary_first)d %(primary_last)d' ' to %(dest)s...') % locals()) - def execute(cmd, process_input=None, check_exit_code=True): - return utils.execute(cmd=cmd, - process_input=process_input, - check_exit_code=check_exit_code) + def execute(*cmd, **kwargs): + return utils.execute(*cmd, **kwargs) - execute('parted --script %s mklabel msdos' % dest) - execute('parted --script %s mkpart primary %ds %ds' % - (dest, primary_first, primary_last)) + execute('sudo', 'parted', '--script', dest, 'mklabel', 'msdos') + execute('sudo', 'parted', '--script', dest, 'mkpart', 'primary', + '%ds' % primary_first, + '%ds' % primary_last) LOG.debug(_('Writing partition table %s done.'), dest) + + +def get_name_label_for_image(image): + # TODO(sirp): This should eventually be the URI for the Glance image + return _('Glance image %s') % image + + +def _mount_filesystem(dev_path, dir): + """mounts the device specified by dev_path in dir""" + try: + out, err = utils.execute('sudo', 'mount', + '-t', 'ext2,ext3', + dev_path, dir) + except exception.ProcessExecutionError as e: + err = str(e) + return err + + +def _find_guest_agent(base_dir, agent_rel_path): + """ + tries to locate a guest agent at the path + specificed by agent_rel_path + """ + agent_path = os.path.join(base_dir, agent_rel_path) + if os.path.isfile(agent_path): + # The presence of the guest agent + # file indicates that this instance can + # reconfigure the network from xenstore data, + # so manipulation of files in /etc is not + # required + LOG.info(_('XenServer tools installed in this ' + 'image are capable of network injection. ' + 'Networking files will not be' + 'manipulated')) + return True + xe_daemon_filename = os.path.join(base_dir, + 'usr', 'sbin', 'xe-daemon') + if os.path.isfile(xe_daemon_filename): + LOG.info(_('XenServer tools are present ' + 'in this image but are not capable ' + 'of network injection')) + else: + LOG.info(_('XenServer tools are not ' + 'installed in this image')) + return False + + +def _mounted_processing(device, key, net): + """Callback which runs with the image VDI attached""" + + dev_path = '/dev/' + device + '1' # NB: Partition 1 hardcoded + tmpdir = tempfile.mkdtemp() + try: + # Mount only Linux filesystems, to avoid disturbing NTFS images + err = _mount_filesystem(dev_path, tmpdir) + if not err: + try: + # This try block ensures that the umount occurs + if not _find_guest_agent(tmpdir, FLAGS.xenapi_agent_path): + LOG.info(_('Manipulating interface files ' + 'directly')) + disk.inject_data_into_fs(tmpdir, key, net, + utils.execute) + finally: + utils.execute('sudo', 'umount', dev_path) + else: + LOG.info(_('Failed to mount filesystem (expected for ' + 'non-linux instances): %s') % err) + finally: + # remove temporary directory + os.rmdir(tmpdir) + + +def _prepare_injectables(inst, networks_info): + """ + prepares the ssh key and the network configuration file to be + injected into the disk image + """ + #do the import here - Cheetah.Template will be loaded + #only if injection is performed + from Cheetah import Template as t + template = t.Template + template_data = open(FLAGS.injected_network_template).read() + + key = str(inst['key_data']) + net = None + if networks_info: + ifc_num = -1 + interfaces_info = [] + have_injected_networks = False + for (network_ref, info) in networks_info: + ifc_num += 1 + if not network_ref['injected']: + continue + + have_injected_networks = True + ip_v4 = ip_v6 = None + if 'ips' in info and len(info['ips']) > 0: + ip_v4 = info['ips'][0] + if 'ip6s' in info and len(info['ip6s']) > 0: + ip_v6 = info['ip6s'][0] + if len(info['dns']) > 0: + dns = info['dns'][0] + interface_info = {'name': 'eth%d' % ifc_num, + 'address': ip_v4 and ip_v4['ip'] or '', + 'netmask': ip_v4 and ip_v4['netmask'] or '', + 'gateway': info['gateway'], + 'broadcast': info['broadcast'], + 'dns': dns, + 'address_v6': ip_v6 and ip_v6['ip'] or '', + 'netmask_v6': ip_v6 and ip_v6['netmask'] or '', + 'gateway_v6': ip_v6 and ip_v6['gateway'] or '', + 'use_ipv6': FLAGS.use_ipv6} + interfaces_info.append(interface_info) + + if have_injected_networks: + net = str(template(template_data, + searchList=[{'interfaces': interfaces_info, + 'use_ipv6': FLAGS.use_ipv6}])) + return key, net diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 98f8ab46e..c96c35a6e 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -19,9 +19,11 @@ Management class for VM-related functions (spawn, reboot, etc). """ +import base64 import json import M2Crypto import os +import pickle import subprocess import tempfile import uuid @@ -31,15 +33,18 @@ from nova import context from nova import log as logging from nova import exception from nova import utils +from nova import flags from nova.auth.manager import AuthManager from nova.compute import power_state +from nova.virt import driver from nova.virt.xenapi.network_utils import NetworkHelper from nova.virt.xenapi.vm_utils import VMHelper from nova.virt.xenapi.vm_utils import ImageType XenAPI = None LOG = logging.getLogger("nova.virt.xenapi.vmops") +FLAGS = flags.FLAGS class VMOps(object): @@ -49,132 +54,232 @@ class VMOps(object): def __init__(self, session): self.XenAPI = session.get_imported_xenapi() self._session = session + self.poll_rescue_last_ran = None VMHelper.XenAPI = self.XenAPI def list_instances(self): - """List VM instances""" - vms = [] - for vm in self._session.get_xenapi().VM.get_all(): - rec = self._session.get_xenapi().VM.get_record(vm) - if not rec["is_a_template"] and not rec["is_control_domain"]: - vms.append(rec["name_label"]) - return vms - - def spawn(self, instance): - """Create VM instance""" - vm = VMHelper.lookup(self._session, instance.name) - if vm is not None: + """List VM instances.""" + # TODO(justinsb): Should we just always use the details method? + # Seems to be the same number of API calls.. + vm_refs = [] + for vm_ref in self._session.get_xenapi().VM.get_all(): + vm_rec = self._session.get_xenapi().VM.get_record(vm_ref) + if not vm_rec["is_a_template"] and not vm_rec["is_control_domain"]: + vm_refs.append(vm_rec["name_label"]) + return vm_refs + + def list_instances_detail(self): + """List VM instances, returning InstanceInfo objects.""" + instance_infos = [] + for vm_ref in self._session.get_xenapi().VM.get_all(): + vm_rec = self._session.get_xenapi().VM.get_record(vm_ref) + if not vm_rec["is_a_template"] and not vm_rec["is_control_domain"]: + name = vm_rec["name_label"] + + # TODO(justinsb): This a roundabout way to map the state + openstack_format = VMHelper.compile_info(vm_rec) + state = openstack_format['state'] + + instance_info = driver.InstanceInfo(name, state) + instance_infos.append(instance_info) + return instance_infos + + def revert_resize(self, instance): + vm_ref = VMHelper.lookup(self._session, instance.name) + self._start(instance, vm_ref) + + def finish_resize(self, instance, disk_info): + vdi_uuid = self.link_disks(instance, disk_info['base_copy'], + disk_info['cow']) + vm_ref = self._create_vm(instance, vdi_uuid) + self.resize_instance(instance, vdi_uuid) + self._spawn(instance, vm_ref) + + def _start(self, instance, vm_ref=None): + """Power on a VM instance""" + if not vm_ref: + vm_ref = VMHelper.lookup(self._session, instance.name) + if vm_ref is None: + raise exception(_('Attempted to power on non-existent instance' + ' bad instance id %s') % instance.id) + LOG.debug(_("Starting instance %s"), instance.name) + self._session.call_xenapi('VM.start', vm_ref, False, False) + + def _create_disk(self, instance): + user = AuthManager().get_user(instance.user_id) + project = AuthManager().get_project(instance.project_id) + disk_image_type = VMHelper.determine_disk_image_type(instance) + vdi_uuid = VMHelper.fetch_image(self._session, instance.id, + instance.image_id, user, project, disk_image_type) + return vdi_uuid + + def spawn(self, instance, network_info=None): + vdi_uuid = self._create_disk(instance) + vm_ref = self._create_vm(instance, vdi_uuid, network_info) + self._spawn(instance, vm_ref) + + def spawn_rescue(self, instance): + """Spawn a rescue instance.""" + self.spawn(instance) + + def _create_vm(self, instance, vdi_uuid, network_info=None): + """Create VM instance.""" + instance_name = instance.name + vm_ref = VMHelper.lookup(self._session, instance_name) + if vm_ref is not None: raise exception.Duplicate(_('Attempted to create' - ' non-unique name %s') % instance.name) + ' non-unique name %s') % instance_name) + #ensure enough free memory is available if not VMHelper.ensure_free_mem(self._session, instance): - name = instance['name'] - LOG.exception(_('instance %(name)s: not enough free memory') - % locals()) - db.instance_set_state(context.get_admin_context(), - instance['id'], - power_state.SHUTDOWN) - return - bridge = db.network_get_by_instance(context.get_admin_context(), - instance['id'])['bridge'] - network_ref = \ - NetworkHelper.find_network_with_bridge(self._session, bridge) + LOG.exception(_('instance %(instance_name)s: not enough free ' + 'memory') % locals()) + db.instance_set_state(context.get_admin_context(), + instance['id'], + power_state.SHUTDOWN) + return user = AuthManager().get_user(instance.user_id) project = AuthManager().get_project(instance.project_id) - #if kernel is not present we must download a raw disk - if instance.kernel_id: - disk_image_type = ImageType.DISK - else: - disk_image_type = ImageType.DISK_RAW - vdi_uuid = VMHelper.fetch_image(self._session, instance.id, - instance.image_id, user, project, disk_image_type) + + # Are we building from a pre-existing disk? vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', vdi_uuid) - #Have a look at the VDI and see if it has a PV kernel - pv_kernel = False - if not instance.kernel_id: - pv_kernel = VMHelper.lookup_image(self._session, instance.id, - vdi_ref) + + disk_image_type = VMHelper.determine_disk_image_type(instance) + kernel = None if instance.kernel_id: kernel = VMHelper.fetch_image(self._session, instance.id, instance.kernel_id, user, project, ImageType.KERNEL_RAMDISK) + ramdisk = None if instance.ramdisk_id: ramdisk = VMHelper.fetch_image(self._session, instance.id, instance.ramdisk_id, user, project, ImageType.KERNEL_RAMDISK) - vm_ref = VMHelper.create_vm(self._session, - instance, kernel, ramdisk, pv_kernel) - VMHelper.create_vbd(self._session, vm_ref, vdi_ref, 0, True) - if network_ref: - VMHelper.create_vif(self._session, vm_ref, - network_ref, instance.mac_address) + use_pv_kernel = VMHelper.determine_is_pv(self._session, instance.id, + vdi_ref, disk_image_type, instance.os_type) + vm_ref = VMHelper.create_vm(self._session, instance, kernel, ramdisk, + use_pv_kernel) + + VMHelper.create_vbd(session=self._session, vm_ref=vm_ref, + vdi_ref=vdi_ref, userdevice=0, bootable=True) + + # TODO(tr3buchet) - check to make sure we have network info, otherwise + # create it now. This goes away once nova-multi-nic hits. + if network_info is None: + network_info = self._get_network_info(instance) + + # Alter the image before VM start for, e.g. network injection + if FLAGS.xenapi_inject_image: + VMHelper.preconfigure_instance(self._session, instance, + vdi_ref, network_info) + + self.create_vifs(vm_ref, network_info) + self.inject_network_info(instance, vm_ref, network_info) + return vm_ref + + def _spawn(self, instance, vm_ref): + """Spawn a new instance.""" LOG.debug(_('Starting VM %s...'), vm_ref) - self._session.call_xenapi('VM.start', vm_ref, False, False) + self._start(instance, vm_ref) instance_name = instance.name LOG.info(_('Spawning VM %(instance_name)s created %(vm_ref)s.') - % locals()) - + % locals()) + + def _inject_files(): + injected_files = instance.injected_files + if injected_files: + # Check if this is a JSON-encoded string and convert if needed. + if isinstance(injected_files, basestring): + try: + injected_files = json.loads(injected_files) + except ValueError: + LOG.exception( + _("Invalid value for injected_files: '%s'") + % injected_files) + injected_files = [] + # Inject any files, if specified + for path, contents in instance.injected_files: + LOG.debug(_("Injecting file path: '%s'") % path) + self.inject_file(instance, path, contents) # NOTE(armando): Do we really need to do this in virt? + # NOTE(tr3buchet): not sure but wherever we do it, we need to call + # reset_network afterwards timer = utils.LoopingCall(f=None) def _wait_for_boot(): try: - state = self.get_info(instance['name'])['state'] + state = self.get_info(instance_name)['state'] db.instance_set_state(context.get_admin_context(), instance['id'], state) if state == power_state.RUNNING: - LOG.debug(_('Instance %s: booted'), instance['name']) + LOG.debug(_('Instance %s: booted'), instance_name) timer.stop() + _inject_files() + return True except Exception, exc: LOG.warn(exc) LOG.exception(_('instance %s: failed to boot'), - instance['name']) + instance_name) db.instance_set_state(context.get_admin_context(), instance['id'], power_state.SHUTDOWN) timer.stop() + return False timer.f = _wait_for_boot + + # call to reset network to configure network from xenstore + self.reset_network(instance, vm_ref) + return timer.start(interval=0.5, now=True) def _get_vm_opaque_ref(self, instance_or_vm): - """Refactored out the common code of many methods that receive either + """ + Refactored out the common code of many methods that receive either a vm name or a vm instance, and want a vm instance in return. """ - vm = None - try: - if instance_or_vm.startswith("OpaqueRef:"): - # Got passed an opaque ref; return it + # if instance_or_vm is a string it must be opaque ref or instance name + if isinstance(instance_or_vm, basestring): + obj = None + try: + # check for opaque ref + obj = self._session.get_xenapi().VM.get_uuid(instance_or_vm) return instance_or_vm - else: - # Must be the instance name + except self.XenAPI.Failure: + # wasn't an opaque ref, can be an instance name instance_name = instance_or_vm - except (AttributeError, KeyError): - # Note the the KeyError will only happen with fakes.py - # Not a string; must be an ID or a vm instance - if isinstance(instance_or_vm, (int, long)): - ctx = context.get_admin_context() - try: - instance_obj = db.instance_get(ctx, instance_or_vm) - instance_name = instance_obj.name - except exception.NotFound: - # The unit tests screw this up, as they use an integer for - # the vm name. I'd fix that up, but that's a matter for - # another bug report. So for now, just try with the passed - # value - instance_name = instance_or_vm - else: - instance_name = instance_or_vm.name - vm = VMHelper.lookup(self._session, instance_name) - if vm is None: + + # if instance_or_vm is an int/long it must be instance id + elif isinstance(instance_or_vm, (int, long)): + ctx = context.get_admin_context() + instance_obj = db.instance_get(ctx, instance_or_vm) + instance_name = instance_obj.name + else: + instance_name = instance_or_vm.name + vm_ref = VMHelper.lookup(self._session, instance_name) + if vm_ref is None: raise exception.NotFound( _('Instance not present %s') % instance_name) - return vm + return vm_ref + + def _acquire_bootlock(self, vm): + """Prevent an instance from booting.""" + self._session.call_xenapi( + "VM.set_blocked_operations", + vm, + {"start": ""}) + + def _release_bootlock(self, vm): + """Allow an instance to boot.""" + self._session.call_xenapi( + "VM.remove_from_blocked_operations", + vm, + "start") def snapshot(self, instance, image_id): - """ Create snapshot from a running VM instance + """Create snapshot from a running VM instance. :param instance: instance to be snapshotted :param image_id: id of image to upload to @@ -194,8 +299,22 @@ class VMOps(object): 3. Push-to-glance: Once coalesced, we call a plugin on the XenServer that will bundle the VHDs together and then push the bundle into Glance. + """ + template_vm_ref = None + try: + template_vm_ref, template_vdi_uuids = self._get_snapshot(instance) + # call plugin to ship snapshot off to glance + VMHelper.upload_image( + self._session, instance, template_vdi_uuids, image_id) + finally: + if template_vm_ref: + self._destroy(instance, template_vm_ref, + shutdown=False, destroy_kernel_ramdisk=False) + + logging.debug(_("Finished snapshot and upload for VM %s"), instance) + def _get_snapshot(self, instance): #TODO(sirp): Add quiesce and VSS locking support when Windows support # is added @@ -206,33 +325,116 @@ class VMOps(object): try: template_vm_ref, template_vdi_uuids = VMHelper.create_snapshot( self._session, instance.id, vm_ref, label) + return template_vm_ref, template_vdi_uuids except self.XenAPI.Failure, exc: logging.error(_("Unable to Snapshot %(vm_ref)s: %(exc)s") % locals()) return + def migrate_disk_and_power_off(self, instance, dest): + """Copies a VHD from one host machine to another. + + :param instance: the instance that owns the VHD in question. + :param dest: the destination host machine. + :param disk_type: values are 'primary' or 'cow'. + + """ + vm_ref = VMHelper.lookup(self._session, instance.name) + + # The primary VDI becomes the COW after the snapshot, and we can + # identify it via the VBD. The base copy is the parent_uuid returned + # from the snapshot creation + + base_copy_uuid = cow_uuid = None + template_vdi_uuids = template_vm_ref = None try: - # call plugin to ship snapshot off to glance - VMHelper.upload_image( - self._session, instance.id, template_vdi_uuids, image_id) + # transfer the base copy + template_vm_ref, template_vdi_uuids = self._get_snapshot(instance) + base_copy_uuid = template_vdi_uuids['image'] + vdi_ref, vm_vdi_rec = \ + VMHelper.get_vdi_for_vm_safely(self._session, vm_ref) + cow_uuid = vm_vdi_rec['uuid'] + + params = {'host': dest, + 'vdi_uuid': base_copy_uuid, + 'instance_id': instance.id, + 'sr_path': VMHelper.get_sr_path(self._session)} + + task = self._session.async_call_plugin('migration', 'transfer_vhd', + {'params': pickle.dumps(params)}) + self._session.wait_for_task(task, instance.id) + + # Now power down the instance and transfer the COW VHD + self._shutdown(instance, vm_ref, hard=False) + + params = {'host': dest, + 'vdi_uuid': cow_uuid, + 'instance_id': instance.id, + 'sr_path': VMHelper.get_sr_path(self._session), } + + task = self._session.async_call_plugin('migration', 'transfer_vhd', + {'params': pickle.dumps(params)}) + self._session.wait_for_task(task, instance.id) + finally: - self._destroy(instance, template_vm_ref, shutdown=False) + if template_vm_ref: + self._destroy(instance, template_vm_ref, + shutdown=False, destroy_kernel_ramdisk=False) - logging.debug(_("Finished snapshot and upload for VM %s"), instance) + # TODO(mdietz): we could also consider renaming these to something + # sensible so we don't need to blindly pass around dictionaries + return {'base_copy': base_copy_uuid, 'cow': cow_uuid} + + def link_disks(self, instance, base_copy_uuid, cow_uuid): + """Links the base copy VHD to the COW via the XAPI plugin.""" + vm_ref = VMHelper.lookup(self._session, instance.name) + new_base_copy_uuid = str(uuid.uuid4()) + new_cow_uuid = str(uuid.uuid4()) + params = {'instance_id': instance.id, + 'old_base_copy_uuid': base_copy_uuid, + 'old_cow_uuid': cow_uuid, + 'new_base_copy_uuid': new_base_copy_uuid, + 'new_cow_uuid': new_cow_uuid, + 'sr_path': VMHelper.get_sr_path(self._session), } + + task = self._session.async_call_plugin('migration', + 'move_vhds_into_sr', {'params': pickle.dumps(params)}) + self._session.wait_for_task(task, instance.id) + + # Now we rescan the SR so we find the VHDs + VMHelper.scan_default_sr(self._session) + + return new_cow_uuid + + def resize_instance(self, instance, vdi_uuid): + """Resize a running instance by changing it's RAM and disk size.""" + #TODO(mdietz): this will need to be adjusted for swap later + #The new disk size must be in bytes + + new_disk_size = str(instance.local_gb * 1024 * 1024 * 1024) + instance_name = instance.name + instance_local_gb = instance.local_gb + LOG.debug(_("Resizing VDI %(vdi_uuid)s for instance %(instance_name)s." + " Expanding to %(instance_local_gb)d GB") % locals()) + vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', vdi_uuid) + self._session.call_xenapi('VDI.resize_online', vdi_ref, new_disk_size) + LOG.debug(_("Resize instance %s complete") % (instance.name)) def reboot(self, instance): - """Reboot VM instance""" - vm = self._get_vm_opaque_ref(instance) - task = self._session.call_xenapi('Async.VM.clean_reboot', vm) - self._session.wait_for_task(instance.id, task) + """Reboot VM instance.""" + vm_ref = self._get_vm_opaque_ref(instance) + task = self._session.call_xenapi('Async.VM.clean_reboot', vm_ref) + self._session.wait_for_task(task, instance.id) def set_admin_password(self, instance, new_pass): - """Set the root/admin password on the VM instance. This is done via - an agent running on the VM. Communication between nova and the agent - is done via writing xenstore records. Since communication is done over - the XenAPI RPC calls, we need to encrypt the password. We're using a - simple Diffie-Hellman class instead of the more advanced one in - M2Crypto for compatibility with the agent code. + """Set the root/admin password on the VM instance. + + This is done via an agent running on the VM. Communication between nova + and the agent is done via writing xenstore records. Since communication + is done over the XenAPI RPC calls, we need to encrypt the password. + We're using a simple Diffie-Hellman class instead of the more advanced + one in M2Crypto for compatibility with the agent code. + """ # Need to uniquely identify this request. transaction_id = str(uuid.uuid4()) @@ -264,142 +466,426 @@ class VMOps(object): raise RuntimeError(resp_dict['message']) return resp_dict['message'] - def _shutdown(self, instance, vm): - """Shutdown an instance """ + def inject_file(self, instance, path, contents): + """Write a file to the VM instance. + + The path to which it is to be written and the contents of the file + need to be supplied; both will be base64-encoded to prevent errors + with non-ASCII characters being transmitted. If the agent does not + support file injection, or the user has disabled it, a + NotImplementedError will be raised. + + """ + # Files/paths must be base64-encoded for transmission to agent + b64_path = base64.b64encode(path) + b64_contents = base64.b64encode(contents) + + # Need to uniquely identify this request. + transaction_id = str(uuid.uuid4()) + args = {'id': transaction_id, 'b64_path': b64_path, + 'b64_contents': b64_contents} + # If the agent doesn't support file injection, a NotImplementedError + # will be raised with the appropriate message. + resp = self._make_agent_call('inject_file', instance, '', args) + resp_dict = json.loads(resp) + if resp_dict['returncode'] != '0': + # There was some other sort of error; the message will contain + # a description of the error. + raise RuntimeError(resp_dict['message']) + return resp_dict['message'] + + def _shutdown(self, instance, vm_ref, hard=True): + """Shutdown an instance.""" state = self.get_info(instance['name'])['state'] if state == power_state.SHUTDOWN: - LOG.warn(_("VM %(vm)s already halted, skipping shutdown...") % - locals()) + instance_name = instance.name + LOG.warn(_("VM %(instance_name)s already halted," + "skipping shutdown...") % locals()) return + instance_id = instance.id + LOG.debug(_("Shutting down VM for Instance %(instance_id)s") + % locals()) try: - task = self._session.call_xenapi('Async.VM.hard_shutdown', vm) - self._session.wait_for_task(instance.id, task) + task = None + if hard: + task = self._session.call_xenapi("Async.VM.hard_shutdown", + vm_ref) + else: + task = self._session.call_xenapi("Async.VM.clean_shutdown", + vm_ref) + self._session.wait_for_task(task, instance.id) except self.XenAPI.Failure, exc: LOG.exception(exc) - def _destroy_vdis(self, instance, vm): - """Destroys all VDIs associated with a VM """ - vdis = VMHelper.lookup_vm_vdis(self._session, vm) + def _shutdown_rescue(self, rescue_vm_ref): + """Shutdown a rescue instance.""" + self._session.call_xenapi("Async.VM.hard_shutdown", rescue_vm_ref) - if not vdis: + def _destroy_vdis(self, instance, vm_ref): + """Destroys all VDIs associated with a VM.""" + instance_id = instance.id + LOG.debug(_("Destroying VDIs for Instance %(instance_id)s") + % locals()) + vdi_refs = VMHelper.lookup_vm_vdis(self._session, vm_ref) + + if not vdi_refs: return - for vdi in vdis: + for vdi_ref in vdi_refs: try: - task = self._session.call_xenapi('Async.VDI.destroy', vdi) - self._session.wait_for_task(instance.id, task) + task = self._session.call_xenapi('Async.VDI.destroy', vdi_ref) + self._session.wait_for_task(task, instance.id) except self.XenAPI.Failure, exc: LOG.exception(exc) - def _destroy_vm(self, instance, vm): - """Destroys a VM record """ + def _destroy_rescue_vdis(self, rescue_vm_ref): + """Destroys all VDIs associated with a rescued VM.""" + vdi_refs = VMHelper.lookup_vm_vdis(self._session, rescue_vm_ref) + for vdi_ref in vdi_refs: + try: + self._session.call_xenapi("Async.VDI.destroy", vdi_ref) + except self.XenAPI.Failure: + continue + + def _destroy_rescue_vbds(self, rescue_vm_ref): + """Destroys all VBDs tied to a rescue VM.""" + vbd_refs = self._session.get_xenapi().VM.get_VBDs(rescue_vm_ref) + for vbd_ref in vbd_refs: + vbd_rec = self._session.get_xenapi().VBD.get_record(vbd_ref) + if vbd_rec.get("userdevice", None) == "1": # VBD is always 1 + VMHelper.unplug_vbd(self._session, vbd_ref) + VMHelper.destroy_vbd(self._session, vbd_ref) + + def _destroy_kernel_ramdisk(self, instance, vm_ref): + """Three situations can occur: + + 1. We have neither a ramdisk nor a kernel, in which case we are a + RAW image and can omit this step + + 2. We have one or the other, in which case, we should flag as an + error + + 3. We have both, in which case we safely remove both the kernel + and the ramdisk. + + """ + instance_id = instance.id + if not instance.kernel_id and not instance.ramdisk_id: + # 1. No kernel or ramdisk + LOG.debug(_("Instance %(instance_id)s using RAW or VHD, " + "skipping kernel and ramdisk deletion") % locals()) + return + + if not (instance.kernel_id and instance.ramdisk_id): + # 2. We only have kernel xor ramdisk + raise exception.NotFound( + _("Instance %(instance_id)s has a kernel or ramdisk but not " + "both" % locals())) + + # 3. We have both kernel and ramdisk + (kernel, ramdisk) = VMHelper.lookup_kernel_ramdisk(self._session, + vm_ref) + + LOG.debug(_("Removing kernel/ramdisk files")) + + args = {'kernel-file': kernel, 'ramdisk-file': ramdisk} + task = self._session.async_call_plugin( + 'glance', 'remove_kernel_ramdisk', args) + self._session.wait_for_task(task, instance.id) + + LOG.debug(_("kernel/ramdisk files removed")) + + def _destroy_vm(self, instance, vm_ref): + """Destroys a VM record.""" + instance_id = instance.id try: - kernel = None - ramdisk = None - if instance.kernel_id or instance.ramdisk_id: - (kernel, ramdisk) = VMHelper.lookup_kernel_ramdisk( - self._session, vm) - task1 = self._session.call_xenapi('Async.VM.destroy', vm) - LOG.debug(_("Removing kernel/ramdisk files")) - fn = "remove_kernel_ramdisk" - args = {} - if kernel: - args['kernel-file'] = kernel - if ramdisk: - args['ramdisk-file'] = ramdisk - task2 = self._session.async_call_plugin('glance', fn, args) - self._session.wait_for_task(instance.id, task1) - self._session.wait_for_task(instance.id, task2) - LOG.debug(_("kernel/ramdisk files removed")) + task = self._session.call_xenapi('Async.VM.destroy', vm_ref) + self._session.wait_for_task(task, instance_id) except self.XenAPI.Failure, exc: LOG.exception(exc) + LOG.debug(_("Instance %(instance_id)s VM destroyed") % locals()) + + def _destroy_rescue_instance(self, rescue_vm_ref): + """Destroy a rescue instance.""" + self._destroy_rescue_vbds(rescue_vm_ref) + self._shutdown_rescue(rescue_vm_ref) + self._destroy_rescue_vdis(rescue_vm_ref) + + self._session.call_xenapi("Async.VM.destroy", rescue_vm_ref) + def destroy(self, instance): - """ - Destroy VM instance + """Destroy VM instance. This is the method exposed by xenapi_conn.destroy(). The rest of the destroy_* methods are internal. - """ - vm = VMHelper.lookup(self._session, instance.name) - return self._destroy(instance, vm, shutdown=True) - def _destroy(self, instance, vm, shutdown=True): """ - Destroys VM instance by performing: + instance_id = instance.id + LOG.info(_("Destroying VM for Instance %(instance_id)s") % locals()) + vm_ref = VMHelper.lookup(self._session, instance.name) + return self._destroy(instance, vm_ref, shutdown=True) + + def _destroy(self, instance, vm_ref, shutdown=True, + destroy_kernel_ramdisk=True): + """Destroys VM instance by performing: + + 1. A shutdown if requested. + 2. Destroying associated VDIs. + 3. Destroying kernel and ramdisk files (if necessary). + 4. Destroying that actual VM record. - 1. A shutdown if requested - 2. Destroying associated VDIs - 3. Destroying that actual VM record """ - if vm is None: - # Don't complain, just return. This lets us clean up instances - # that have already disappeared from the underlying platform. + if vm_ref is None: + LOG.warning(_("VM is not present, skipping destroy...")) return if shutdown: - self._shutdown(instance, vm) + self._shutdown(instance, vm_ref) - self._destroy_vdis(instance, vm) - self._destroy_vm(instance, vm) + self._destroy_vdis(instance, vm_ref) + if destroy_kernel_ramdisk: + self._destroy_kernel_ramdisk(instance, vm_ref) + self._destroy_vm(instance, vm_ref) def _wait_with_callback(self, instance_id, task, callback): ret = None try: - ret = self._session.wait_for_task(instance_id, task) + ret = self._session.wait_for_task(task, instance_id) except self.XenAPI.Failure, exc: LOG.exception(exc) callback(ret) def pause(self, instance, callback): - """Pause VM instance""" - vm = self._get_vm_opaque_ref(instance) - task = self._session.call_xenapi('Async.VM.pause', vm) + """Pause VM instance.""" + vm_ref = self._get_vm_opaque_ref(instance) + task = self._session.call_xenapi('Async.VM.pause', vm_ref) self._wait_with_callback(instance.id, task, callback) def unpause(self, instance, callback): - """Unpause VM instance""" - vm = self._get_vm_opaque_ref(instance) - task = self._session.call_xenapi('Async.VM.unpause', vm) + """Unpause VM instance.""" + vm_ref = self._get_vm_opaque_ref(instance) + task = self._session.call_xenapi('Async.VM.unpause', vm_ref) self._wait_with_callback(instance.id, task, callback) def suspend(self, instance, callback): - """suspend the specified instance""" - vm = self._get_vm_opaque_ref(instance) - task = self._session.call_xenapi('Async.VM.suspend', vm) + """Suspend the specified instance.""" + vm_ref = self._get_vm_opaque_ref(instance) + task = self._session.call_xenapi('Async.VM.suspend', vm_ref) self._wait_with_callback(instance.id, task, callback) def resume(self, instance, callback): - """resume the specified instance""" - vm = self._get_vm_opaque_ref(instance) - task = self._session.call_xenapi('Async.VM.resume', vm, False, True) + """Resume the specified instance.""" + vm_ref = self._get_vm_opaque_ref(instance) + task = self._session.call_xenapi('Async.VM.resume', vm_ref, False, + True) self._wait_with_callback(instance.id, task, callback) + def rescue(self, instance, callback): + """Rescue the specified instance. + + - shutdown the instance VM. + - set 'bootlock' to prevent the instance from starting in rescue. + - spawn a rescue VM (the vm name-label will be instance-N-rescue). + + """ + rescue_vm_ref = VMHelper.lookup(self._session, + "%s-rescue" % instance.name) + if rescue_vm_ref: + raise RuntimeError(_( + "Instance is already in Rescue Mode: %s" % instance.name)) + + vm_ref = VMHelper.lookup(self._session, instance.name) + self._shutdown(instance, vm_ref) + self._acquire_bootlock(vm_ref) + instance._rescue = True + self.spawn_rescue(instance) + rescue_vm_ref = VMHelper.lookup(self._session, instance.name) + + vbd_ref = self._session.get_xenapi().VM.get_VBDs(vm_ref)[0] + vdi_ref = self._session.get_xenapi().VBD.get_record(vbd_ref)["VDI"] + rescue_vbd_ref = VMHelper.create_vbd(self._session, rescue_vm_ref, + vdi_ref, 1, False) + + self._session.call_xenapi("Async.VBD.plug", rescue_vbd_ref) + + def unrescue(self, instance, callback): + """Unrescue the specified instance. + + - unplug the instance VM's disk from the rescue VM. + - teardown the rescue VM. + - release the bootlock to allow the instance VM to start. + + """ + rescue_vm_ref = VMHelper.lookup(self._session, + "%s-rescue" % instance.name) + + if not rescue_vm_ref: + raise exception.NotFound(_( + "Instance is not in Rescue Mode: %s" % instance.name)) + + original_vm_ref = VMHelper.lookup(self._session, instance.name) + instance._rescue = False + + self._destroy_rescue_instance(rescue_vm_ref) + self._release_bootlock(original_vm_ref) + self._start(instance, original_vm_ref) + + def poll_rescued_instances(self, timeout): + """Look for expirable rescued instances. + + - forcibly exit rescue mode for any instances that have been + in rescue mode for >= the provided timeout + + """ + last_ran = self.poll_rescue_last_ran + if not last_ran: + # We need a base time to start tracking. + self.poll_rescue_last_ran = utils.utcnow() + return + + if not utils.is_older_than(last_ran, timeout): + # Do not run. Let's bail. + return + + # Update the time tracker and proceed. + self.poll_rescue_last_ran = utils.utcnow() + + rescue_vms = [] + for instance in self.list_instances(): + if instance.endswith("-rescue"): + rescue_vms.append(dict(name=instance, + vm_ref=VMHelper.lookup(self._session, + instance))) + + for vm in rescue_vms: + rescue_name = vm["name"] + rescue_vm_ref = vm["vm_ref"] + + self._destroy_rescue_instance(rescue_vm_ref) + + original_name = vm["name"].split("-rescue", 1)[0] + original_vm_ref = VMHelper.lookup(self._session, original_name) + + self._release_bootlock(original_vm_ref) + self._session.call_xenapi("VM.start", original_vm_ref, False, + False) + def get_info(self, instance): - """Return data about VM instance""" - vm = self._get_vm_opaque_ref(instance) - rec = self._session.get_xenapi().VM.get_record(vm) - return VMHelper.compile_info(rec) + """Return data about VM instance.""" + vm_ref = self._get_vm_opaque_ref(instance) + vm_rec = self._session.get_xenapi().VM.get_record(vm_ref) + return VMHelper.compile_info(vm_rec) def get_diagnostics(self, instance): - """Return data about VM diagnostics""" - vm = self._get_vm_opaque_ref(instance) - rec = self._session.get_xenapi().VM.get_record(vm) - return VMHelper.compile_diagnostics(self._session, rec) + """Return data about VM diagnostics.""" + vm_ref = self._get_vm_opaque_ref(instance) + vm_rec = self._session.get_xenapi().VM.get_record(vm_ref) + return VMHelper.compile_diagnostics(self._session, vm_rec) def get_console_output(self, instance): - """Return snapshot of console""" + """Return snapshot of console.""" # TODO: implement this to fix pylint! return 'FAKE CONSOLE OUTPUT of instance' def get_ajax_console(self, instance): - """Return link to instance's ajax console""" + """Return link to instance's ajax console.""" # TODO: implement this! return 'http://fakeajaxconsole/fake_url' + # TODO(tr3buchet) - remove this function after nova multi-nic + def _get_network_info(self, instance): + """Creates network info list for instance.""" + admin_context = context.get_admin_context() + IPs = db.fixed_ip_get_all_by_instance(admin_context, + instance['id']) + networks = db.network_get_all_by_instance(admin_context, + instance['id']) + flavor = db.instance_type_get_by_name(admin_context, + instance['instance_type']) + network_info = [] + for network in networks: + network_IPs = [ip for ip in IPs if ip.network_id == network.id] + + def ip_dict(ip): + return { + "ip": ip.address, + "netmask": network["netmask"], + "enabled": "1"} + + def ip6_dict(ip6): + return { + "ip": utils.to_global_ipv6(network['cidr_v6'], + instance['mac_address']), + "netmask": network['netmask_v6'], + "gateway": network['gateway_v6'], + "enabled": "1"} + + info = { + 'label': network['label'], + 'gateway': network['gateway'], + 'broadcast': network['broadcast'], + 'mac': instance.mac_address, + 'rxtx_cap': flavor['rxtx_cap'], + 'dns': [network['dns']], + 'ips': [ip_dict(ip) for ip in network_IPs]} + if network['cidr_v6']: + info['ip6s'] = [ip6_dict(ip) for ip in network_IPs] + network_info.append((network, info)) + return network_info + + def inject_network_info(self, instance, vm_ref, network_info): + """ + Generate the network info and make calls to place it into the + xenstore and the xenstore param list. + """ + logging.debug(_("injecting network info to xs for vm: |%s|"), vm_ref) + + # this function raises if vm_ref is not a vm_opaque_ref + self._session.get_xenapi().VM.get_record(vm_ref) + + for (network, info) in network_info: + location = 'vm-data/networking/%s' % info['mac'].replace(':', '') + self.write_to_param_xenstore(vm_ref, {location: info}) + try: + # TODO(tr3buchet): fix function call after refactor + #self.write_to_xenstore(vm_ref, location, info) + self._make_plugin_call('xenstore.py', 'write_record', instance, + location, {'value': json.dumps(info)}, + vm_ref) + except KeyError: + # catch KeyError for domid if instance isn't running + pass + + def create_vifs(self, vm_ref, network_info): + """Creates vifs for an instance.""" + logging.debug(_("creating vif(s) for vm: |%s|"), vm_ref) + + # this function raises if vm_ref is not a vm_opaque_ref + self._session.get_xenapi().VM.get_record(vm_ref) + + for device, (network, info) in enumerate(network_info): + mac_address = info['mac'] + bridge = network['bridge'] + rxtx_cap = info.pop('rxtx_cap') + network_ref = \ + NetworkHelper.find_network_with_bridge(self._session, + bridge) + VMHelper.create_vif(self._session, vm_ref, network_ref, + mac_address, device, rxtx_cap) + + def reset_network(self, instance, vm_ref): + """Creates uuid arg to pass to make_agent_call and calls it.""" + args = {'id': str(uuid.uuid4())} + # TODO(tr3buchet): fix function call after refactor + #resp = self._make_agent_call('resetnetwork', instance, '', args) + resp = self._make_plugin_call('agent', 'resetnetwork', instance, '', + args, vm_ref) + def list_from_xenstore(self, vm, path): - """Runs the xenstore-ls command to get a listing of all records + """ + Runs the xenstore-ls command to get a listing of all records from 'path' downward. Returns a dict with the sub-paths as keys, and the value stored in those paths as values. If nothing is found at that path, returns None. @@ -408,7 +894,8 @@ class VMOps(object): return json.loads(ret) def read_from_xenstore(self, vm, path): - """Returns the value stored in the xenstore record for the given VM + """ + Returns the value stored in the xenstore record for the given VM at the specified location. A XenAPIPlugin.PluginError will be raised if any error is encountered in the read process. """ @@ -424,7 +911,8 @@ class VMOps(object): return ret def write_to_xenstore(self, vm, path, value): - """Writes the passed value to the xenstore record for the given VM + """ + Writes the passed value to the xenstore record for the given VM at the specified location. A XenAPIPlugin.PluginError will be raised if any error is encountered in the write process. """ @@ -432,33 +920,36 @@ class VMOps(object): {'value': json.dumps(value)}) def clear_xenstore(self, vm, path): - """Deletes the VM's xenstore record for the specified path. + """ + Deletes the VM's xenstore record for the specified path. If there is no such record, the request is ignored. """ self._make_xenstore_call('delete_record', vm, path) - def _make_xenstore_call(self, method, vm, path, addl_args={}): + def _make_xenstore_call(self, method, vm, path, addl_args=None): """Handles calls to the xenstore xenapi plugin.""" return self._make_plugin_call('xenstore.py', method=method, vm=vm, path=path, addl_args=addl_args) - def _make_agent_call(self, method, vm, path, addl_args={}): + def _make_agent_call(self, method, vm, path, addl_args=None): """Abstracts out the interaction with the agent xenapi plugin.""" return self._make_plugin_call('agent', method=method, vm=vm, path=path, addl_args=addl_args) - def _make_plugin_call(self, plugin, method, vm, path, addl_args={}): - """Abstracts out the process of calling a method of a xenapi plugin. + def _make_plugin_call(self, plugin, method, vm, path, addl_args=None, + vm_ref=None): + """ + Abstracts out the process of calling a method of a xenapi plugin. Any errors raised by the plugin will in turn raise a RuntimeError here. """ instance_id = vm.id - vm = self._get_vm_opaque_ref(vm) - rec = self._session.get_xenapi().VM.get_record(vm) - args = {'dom_id': rec['domid'], 'path': path} - args.update(addl_args) + vm_ref = vm_ref or self._get_vm_opaque_ref(vm) + vm_rec = self._session.get_xenapi().VM.get_record(vm_ref) + args = {'dom_id': vm_rec['domid'], 'path': path} + args.update(addl_args or {}) try: task = self._session.async_call_plugin(plugin, method, args) - ret = self._session.wait_for_task(instance_id, task) + ret = self._session.wait_for_task(task, instance_id) except self.XenAPI.Failure, e: ret = None err_trace = e.details[-1] @@ -467,13 +958,19 @@ class VMOps(object): if 'TIMEOUT:' in err_msg: LOG.error(_('TIMEOUT: The call to %(method)s timed out. ' 'VM id=%(instance_id)s; args=%(strargs)s') % locals()) + elif 'NOT IMPLEMENTED:' in err_msg: + LOG.error(_('NOT IMPLEMENTED: The call to %(method)s is not' + ' supported by the agent. VM id=%(instance_id)s;' + ' args=%(strargs)s') % locals()) + raise NotImplementedError(err_msg) else: LOG.error(_('The call to %(method)s returned an error: %(e)s. ' 'VM id=%(instance_id)s; args=%(strargs)s') % locals()) return ret def add_to_xenstore(self, vm, path, key, value): - """Adds the passed key/value pair to the xenstore record for + """ + Adds the passed key/value pair to the xenstore record for the given VM at the specified location. A XenAPIPlugin.PluginError will be raised if any error is encountered in the write process. """ @@ -486,7 +983,8 @@ class VMOps(object): self.write_to_xenstore(vm, path, current) def remove_from_xenstore(self, vm, path, key_or_keys): - """Takes either a single key or a list of keys and removes + """ + Takes either a single key or a list of keys and removes them from the xenstoreirecord data for the given VM. If the key doesn't exist, the request is ignored. """ @@ -513,7 +1011,8 @@ class VMOps(object): ###### names to distinguish them. (dabo) ######################################################################## def read_partial_from_param_xenstore(self, instance_or_vm, key_prefix): - """Returns a dict of all the keys in the xenstore parameter record + """ + Returns a dict of all the keys in the xenstore parameter record for the given instance that begin with the key_prefix. """ data = self.read_from_param_xenstore(instance_or_vm) @@ -524,14 +1023,15 @@ class VMOps(object): return data def read_from_param_xenstore(self, instance_or_vm, keys=None): - """Returns the xenstore parameter record data for the specified VM + """ + Returns the xenstore parameter record data for the specified VM instance as a dict. Accepts an optional key or list of keys; if a value for 'keys' is passed, the returned dict is filtered to only return the values for those keys. """ - vm = self._get_vm_opaque_ref(instance_or_vm) + vm_ref = self._get_vm_opaque_ref(instance_or_vm) data = self._session.call_xenapi_request('VM.get_xenstore_data', - (vm, )) + (vm_ref,)) ret = {} if keys is None: keys = data.keys() @@ -546,17 +1046,20 @@ class VMOps(object): return ret def add_to_param_xenstore(self, instance_or_vm, key, val): - """Takes a key/value pair and adds it to the xenstore parameter + """ + Takes a key/value pair and adds it to the xenstore parameter record for the given vm instance. If the key exists in xenstore, - it is overwritten""" - vm = self._get_vm_opaque_ref(instance_or_vm) + it is overwritten + """ + vm_ref = self._get_vm_opaque_ref(instance_or_vm) self.remove_from_param_xenstore(instance_or_vm, key) jsonval = json.dumps(val) self._session.call_xenapi_request('VM.add_to_xenstore_data', - (vm, key, jsonval)) + (vm_ref, key, jsonval)) def write_to_param_xenstore(self, instance_or_vm, mapping): - """Takes a dict and writes each key/value pair to the xenstore + """ + Takes a dict and writes each key/value pair to the xenstore parameter record for the given vm instance. Any existing data for those keys is overwritten. """ @@ -564,18 +1067,19 @@ class VMOps(object): self.add_to_param_xenstore(instance_or_vm, k, v) def remove_from_param_xenstore(self, instance_or_vm, key_or_keys): - """Takes either a single key or a list of keys and removes + """ + Takes either a single key or a list of keys and removes them from the xenstore parameter record data for the given VM. If the key doesn't exist, the request is ignored. """ - vm = self._get_vm_opaque_ref(instance_or_vm) + vm_ref = self._get_vm_opaque_ref(instance_or_vm) if isinstance(key_or_keys, basestring): keys = [key_or_keys] else: keys = key_or_keys for key in keys: self._session.call_xenapi_request('VM.remove_from_xenstore_data', - (vm, key)) + (vm_ref, key)) def clear_param_xenstore(self, instance_or_vm): """Removes all data from the xenstore parameter record for this VM.""" @@ -590,7 +1094,8 @@ def _runproc(cmd): class SimpleDH(object): - """This class wraps all the functionality needed to implement + """ + This class wraps all the functionality needed to implement basic Diffie-Hellman-Merkle key exchange in Python. It features intelligent defaults for the prime and base numbers needed for the calculation, while allowing you to supply your own. It requires that @@ -599,7 +1104,8 @@ class SimpleDH(object): is not available, a RuntimeError will be raised. """ def __init__(self, prime=None, base=None, secret=None): - """You can specify the values for prime and base if you wish; + """ + You can specify the values for prime and base if you wish; otherwise, reasonable default values will be used. """ if prime is None: diff --git a/nova/virt/xenapi/volume_utils.py b/nova/virt/xenapi/volume_utils.py index d5ebd29d5..72284ac02 100644 --- a/nova/virt/xenapi/volume_utils.py +++ b/nova/virt/xenapi/volume_utils.py @@ -117,16 +117,16 @@ class VolumeHelper(HelperBase): def introduce_vdi(cls, session, sr_ref): """Introduce VDI in the host""" try: - vdis = session.get_xenapi().SR.get_VDIs(sr_ref) + vdi_refs = session.get_xenapi().SR.get_VDIs(sr_ref) except cls.XenAPI.Failure, exc: LOG.exception(exc) raise StorageError(_('Unable to introduce VDI on SR %s') % sr_ref) try: - vdi_rec = session.get_xenapi().VDI.get_record(vdis[0]) + vdi_rec = session.get_xenapi().VDI.get_record(vdi_refs[0]) except cls.XenAPI.Failure, exc: LOG.exception(exc) raise StorageError(_('Unable to get record' - ' of VDI %s on') % vdis[0]) + ' of VDI %s on') % vdi_refs[0]) else: try: return session.get_xenapi().VDI.introduce( diff --git a/nova/virt/xenapi/volumeops.py b/nova/virt/xenapi/volumeops.py index d89a6f995..757ecf5ad 100644 --- a/nova/virt/xenapi/volumeops.py +++ b/nova/virt/xenapi/volumeops.py @@ -83,7 +83,7 @@ class VolumeOps(object): try: task = self._session.call_xenapi('Async.VBD.plug', vbd_ref) - self._session.wait_for_task(vol_rec['deviceNumber'], task) + self._session.wait_for_task(task, vol_rec['deviceNumber']) except self.XenAPI.Failure, exc: LOG.exception(exc) VolumeHelper.destroy_iscsi_storage(self._session, diff --git a/nova/virt/xenapi_conn.py b/nova/virt/xenapi_conn.py index a0b0499b8..99fd35c61 100644 --- a/nova/virt/xenapi_conn.py +++ b/nova/virt/xenapi_conn.py @@ -49,6 +49,12 @@ reactor thread if the VM.get_by_name_label or VM.get_record calls block. address for the nova-volume host :target_port: iSCSI Target Port, 3260 Default :iqn_prefix: IQN Prefix, e.g. 'iqn.2010-10.org.openstack' + +**Variable Naming Scheme** + +- suffix "_ref" for opaque references +- suffix "_uuid" for UUIDs +- suffix "_rec" for record objects """ import sys @@ -63,6 +69,7 @@ from nova import db from nova import utils from nova import flags from nova import log as logging +from nova.virt import driver from nova.virt.xenapi.vmops import VMOps from nova.virt.xenapi.volumeops import VolumeOps @@ -100,6 +107,22 @@ flags.DEFINE_integer('xenapi_vhd_coalesce_max_attempts', 5, 'Max number of times to poll for VHD to coalesce.' ' Used only if connection_type=xenapi.') +flags.DEFINE_bool('xenapi_inject_image', + True, + 'Specifies whether an attempt to inject network/key' + ' data into the disk image should be made.' + ' Used only if connection_type=xenapi.') +flags.DEFINE_string('xenapi_agent_path', + 'usr/sbin/xe-update-networking', + 'Specifies the path in which the xenapi guest agent' + ' should be located. If the agent is present,' + ' network configuration is not injected into the image' + ' Used only if connection_type=xenapi.' + ' and xenapi_inject_image=True') + +flags.DEFINE_string('xenapi_sr_base_path', '/var/run/sr-mount', + 'Base path to the storage repository') + flags.DEFINE_string('target_host', None, 'iSCSI Target Host') @@ -133,10 +156,11 @@ def get_connection(_): return XenAPIConnection(url, username, password) -class XenAPIConnection(object): +class XenAPIConnection(driver.ComputeDriver): """A connection to XenServer or Xen Cloud Platform""" def __init__(self, url, user, pw): + super(XenAPIConnection, self).__init__() session = XenAPISession(url, user, pw) self._vmops = VMOps(session) self._volumeops = VolumeOps(session) @@ -152,10 +176,21 @@ class XenAPIConnection(object): """List VM instances""" return self._vmops.list_instances() + def list_instances_detail(self): + return self._vmops.list_instances_detail() + def spawn(self, instance): """Create VM instance""" self._vmops.spawn(instance) + def revert_resize(self, instance): + """Reverts a resize, powering back on the instance""" + self._vmops.revert_resize(instance) + + def finish_resize(self, instance, disk_info): + """Completes a resize, turning on the migrated instance""" + self._vmops.finish_resize(instance, disk_info) + def snapshot(self, instance, image_id): """ Create snapshot from a running VM instance """ self._vmops.snapshot(instance, image_id) @@ -168,6 +203,12 @@ class XenAPIConnection(object): """Set the root/admin password on the VM instance""" self._vmops.set_admin_password(instance, new_pass) + def inject_file(self, instance, b64_path, b64_contents): + """Create a file on the VM instance. The file path and contents + should be base64-encoded. + """ + self._vmops.inject_file(instance, b64_path, b64_contents) + def destroy(self, instance): """Destroy VM instance""" self._vmops.destroy(instance) @@ -180,6 +221,11 @@ class XenAPIConnection(object): """Unpause paused VM instance""" self._vmops.unpause(instance, callback) + def migrate_disk_and_power_off(self, instance, dest): + """Transfers the VHD of a running instance to another host, then shuts + off the instance copies over the COW disk""" + return self._vmops.migrate_disk_and_power_off(instance, dest) + def suspend(self, instance, callback): """suspend the specified instance""" self._vmops.suspend(instance, callback) @@ -188,6 +234,26 @@ class XenAPIConnection(object): """resume the specified instance""" self._vmops.resume(instance, callback) + def rescue(self, instance, callback): + """Rescue the specified instance""" + self._vmops.rescue(instance, callback) + + def unrescue(self, instance, callback): + """Unrescue the specified instance""" + self._vmops.unrescue(instance, callback) + + def poll_rescued_instances(self, timeout): + """Poll for rescued instances""" + self._vmops.poll_rescued_instances(timeout) + + def reset_network(self, instance): + """reset networking for specified instance""" + self._vmops.reset_network(instance) + + def inject_network_info(self, instance): + """inject network info for specified instance""" + self._vmops.inject_network_info(instance) + def get_info(self, instance_id): """Return data about VM instance""" return self._vmops.get_info(instance_id) @@ -204,6 +270,10 @@ class XenAPIConnection(object): """Return link to instance's ajax console""" return self._vmops.get_ajax_console(instance) + def get_host_ip_addr(self): + xs_url = urlparse.urlparse(FLAGS.xenapi_connection_url) + return xs_url.netloc + def attach_volume(self, instance_name, device_path, mountpoint): """Attach volume storage to VM instance""" return self._volumeops.attach_volume(instance_name, @@ -220,6 +290,27 @@ class XenAPIConnection(object): 'username': FLAGS.xenapi_connection_username, 'password': FLAGS.xenapi_connection_password} + def update_available_resource(self, ctxt, host): + """This method is supported only by libvirt.""" + return + + def compare_cpu(self, xml): + """This method is supported only by libvirt.""" + raise NotImplementedError('This method is supported only by libvirt.') + + def ensure_filtering_rules_for_instance(self, instance_ref): + """This method is supported only libvirt.""" + return + + def live_migration(self, context, instance_ref, dest, + post_method, recover_method): + """This method is supported only by libvirt.""" + return + + def unfilter_instance(self, instance_ref): + """This method is supported only by libvirt.""" + raise NotImplementedError('This method is supported only by libvirt.') + class XenAPISession(object): """The session to invoke XenAPI SDK calls""" @@ -263,7 +354,7 @@ class XenAPISession(object): self._session.xenapi.Async.host.call_plugin, self.get_xenapi_host(), plugin, fn, args) - def wait_for_task(self, id, task): + def wait_for_task(self, task, id=None): """Return the result of the given task. The task is polled until it completes. Not re-entrant.""" done = event.Event() @@ -290,10 +381,11 @@ class XenAPISession(object): try: name = self._session.xenapi.task.get_name_label(task) status = self._session.xenapi.task.get_status(task) - action = dict( - instance_id=int(id), - action=name[0:255], # Ensure action is never > 255 - error=None) + if id: + action = dict( + instance_id=int(id), + action=name[0:255], # Ensure action is never > 255 + error=None) if status == "pending": return elif status == "success": @@ -307,7 +399,9 @@ class XenAPISession(object): LOG.warn(_("Task [%(name)s] %(task)s status:" " %(status)s %(error_info)s") % locals()) done.send_exception(self.XenAPI.Failure(error_info)) - db.instance_action_create(context.get_admin_context(), action) + + if id: + db.instance_action_create(context.get_admin_context(), action) except self.XenAPI.Failure, exc: LOG.warn(exc) done.send_exception(*sys.exc_info()) |
