449 lines
17 KiB
Python
449 lines
17 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright (c) 2010 Citrix Systems, Inc.
|
|
# Copyright 2010 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 XenServer or Xen Cloud Platform.
|
|
|
|
The concurrency model for this class is as follows:
|
|
|
|
All XenAPI calls are on a green thread (using eventlet's "tpool"
|
|
thread pool). They are remote calls, and so may hang for the usual
|
|
reasons.
|
|
|
|
All long-running XenAPI calls (VM.start, VM.reboot, etc) are called async
|
|
(using XenAPI.VM.async_start etc). These return a task, which can then be
|
|
polled for completion.
|
|
|
|
This combination of techniques means that we don't block the main thread at
|
|
all, and at the same time we don't hold lots of threads waiting for
|
|
long-running operations.
|
|
|
|
FIXME: get_info currently doesn't conform to these rules, and will block the
|
|
reactor thread if the VM.get_by_name_label or VM.get_record calls block.
|
|
|
|
**Related Flags**
|
|
|
|
:xenapi_connection_url: URL for connection to XenServer/Xen Cloud Platform.
|
|
:xenapi_connection_username: Username for connection to XenServer/Xen Cloud
|
|
Platform (default: root).
|
|
:xenapi_connection_password: Password for connection to XenServer/Xen Cloud
|
|
Platform.
|
|
:xenapi_task_poll_interval: The interval (seconds) used for polling of
|
|
remote tasks (Async.VM.start, etc)
|
|
(default: 0.5).
|
|
:target_host: the iSCSI Target Host IP address, i.e. the IP
|
|
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
|
|
import urlparse
|
|
import xmlrpclib
|
|
|
|
from eventlet import event
|
|
from eventlet import tpool
|
|
from eventlet import timeout
|
|
|
|
from nova import context
|
|
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
|
|
|
|
|
|
LOG = logging.getLogger("nova.virt.xenapi")
|
|
|
|
|
|
FLAGS = flags.FLAGS
|
|
|
|
flags.DEFINE_string('xenapi_connection_url',
|
|
None,
|
|
'URL for connection to XenServer/Xen Cloud Platform.'
|
|
' Required if connection_type=xenapi.')
|
|
flags.DEFINE_string('xenapi_connection_username',
|
|
'root',
|
|
'Username for connection to XenServer/Xen Cloud Platform.'
|
|
' Used only if connection_type=xenapi.')
|
|
flags.DEFINE_string('xenapi_connection_password',
|
|
None,
|
|
'Password for connection to XenServer/Xen Cloud Platform.'
|
|
' Used only if connection_type=xenapi.')
|
|
flags.DEFINE_float('xenapi_task_poll_interval',
|
|
0.5,
|
|
'The interval used for polling of remote tasks '
|
|
'(Async.VM.start, etc). Used only if '
|
|
'connection_type=xenapi.')
|
|
flags.DEFINE_string('xenapi_image_service',
|
|
'glance',
|
|
'Where to get VM images: glance or objectstore.')
|
|
flags.DEFINE_float('xenapi_vhd_coalesce_poll_interval',
|
|
5.0,
|
|
'The interval used for polling of coalescing vhds.'
|
|
' Used only if connection_type=xenapi.')
|
|
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')
|
|
flags.DEFINE_string('target_port',
|
|
'3260',
|
|
'iSCSI Target Port, 3260 Default')
|
|
flags.DEFINE_string('iqn_prefix',
|
|
'iqn.2010-10.org.openstack',
|
|
'IQN Prefix')
|
|
# NOTE(sirp): This is a work-around for a bug in Ubuntu Maverick, when we pull
|
|
# support for it, we should remove this
|
|
flags.DEFINE_bool('xenapi_remap_vbd_dev', False,
|
|
'Used to enable the remapping of VBD dev '
|
|
'(Works around an issue in Ubuntu Maverick)')
|
|
flags.DEFINE_string('xenapi_remap_vbd_dev_prefix', 'sd',
|
|
'Specify prefix to remap VBD dev to '
|
|
'(ex. /dev/xvdb -> /dev/sdb)')
|
|
flags.DEFINE_integer('xenapi_login_timeout',
|
|
10,
|
|
'Timeout in seconds for XenAPI login.')
|
|
|
|
|
|
def get_connection(_):
|
|
"""Note that XenAPI doesn't have a read-only connection mode, so
|
|
the read_only parameter is ignored."""
|
|
url = FLAGS.xenapi_connection_url
|
|
username = FLAGS.xenapi_connection_username
|
|
password = FLAGS.xenapi_connection_password
|
|
if not url or password is None:
|
|
raise Exception(_('Must specify xenapi_connection_url, '
|
|
'xenapi_connection_username (optionally), and '
|
|
'xenapi_connection_password to use '
|
|
'connection_type=xenapi'))
|
|
return XenAPIConnection(url, username, password)
|
|
|
|
|
|
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)
|
|
|
|
def init_host(self, host):
|
|
#FIXME(armando): implement this
|
|
#NOTE(armando): would we need a method
|
|
#to call when shutting down the host?
|
|
#e.g. to do session logout?
|
|
pass
|
|
|
|
def list_instances(self):
|
|
"""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)
|
|
|
|
def reboot(self, instance):
|
|
"""Reboot VM instance"""
|
|
self._vmops.reboot(instance)
|
|
|
|
def set_admin_password(self, instance, new_pass):
|
|
"""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)
|
|
|
|
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 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)
|
|
|
|
def resume(self, instance, callback):
|
|
"""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)
|
|
|
|
def get_diagnostics(self, instance):
|
|
"""Return data about VM diagnostics"""
|
|
return self._vmops.get_diagnostics(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 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,
|
|
device_path,
|
|
mountpoint)
|
|
|
|
def detach_volume(self, instance_name, mountpoint):
|
|
"""Detach volume storage to VM instance"""
|
|
return self._volumeops.detach_volume(instance_name, mountpoint)
|
|
|
|
def get_console_pool_info(self, console_type):
|
|
xs_url = urlparse.urlparse(FLAGS.xenapi_connection_url)
|
|
return {'address': xs_url.netloc,
|
|
'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"""
|
|
|
|
def __init__(self, url, user, pw):
|
|
self.XenAPI = self.get_imported_xenapi()
|
|
self._session = self._create_session(url)
|
|
exception = self.XenAPI.Failure(_("Unable to log in to XenAPI "
|
|
"(is the Dom0 disk full?)"))
|
|
with timeout.Timeout(FLAGS.xenapi_login_timeout, exception):
|
|
self._session.login_with_password(user, pw)
|
|
self.loop = None
|
|
|
|
def get_imported_xenapi(self):
|
|
"""Stubout point. This can be replaced with a mock xenapi module."""
|
|
return __import__('XenAPI')
|
|
|
|
def get_xenapi(self):
|
|
"""Return the xenapi object"""
|
|
return self._session.xenapi
|
|
|
|
def get_xenapi_host(self):
|
|
"""Return the xenapi host"""
|
|
return self._session.xenapi.session.get_this_host(self._session.handle)
|
|
|
|
def call_xenapi(self, method, *args):
|
|
"""Call the specified XenAPI method on a background thread."""
|
|
f = self._session.xenapi
|
|
for m in method.split('.'):
|
|
f = f.__getattr__(m)
|
|
return tpool.execute(f, *args)
|
|
|
|
def call_xenapi_request(self, method, *args):
|
|
"""Some interactions with dom0, such as interacting with xenstore's
|
|
param record, require using the xenapi_request method of the session
|
|
object. This wraps that call on a background thread.
|
|
"""
|
|
f = self._session.xenapi_request
|
|
return tpool.execute(f, method, *args)
|
|
|
|
def async_call_plugin(self, plugin, fn, args):
|
|
"""Call Async.host.call_plugin on a background thread."""
|
|
return tpool.execute(self._unwrap_plugin_exceptions,
|
|
self._session.xenapi.Async.host.call_plugin,
|
|
self.get_xenapi_host(), plugin, fn, args)
|
|
|
|
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()
|
|
self.loop = utils.LoopingCall(self._poll_task, id, task, done)
|
|
self.loop.start(FLAGS.xenapi_task_poll_interval, now=True)
|
|
rv = done.wait()
|
|
self.loop.stop()
|
|
return rv
|
|
|
|
def _stop_loop(self):
|
|
"""Stop polling for task to finish."""
|
|
#NOTE(sandy-walsh) Had to break this call out to support unit tests.
|
|
if self.loop:
|
|
self.loop.stop()
|
|
|
|
def _create_session(self, url):
|
|
"""Stubout point. This can be replaced with a mock session."""
|
|
return self.XenAPI.Session(url)
|
|
|
|
def _poll_task(self, id, task, done):
|
|
"""Poll the given XenAPI task, and fire the given action if we
|
|
get a result.
|
|
"""
|
|
try:
|
|
name = self._session.xenapi.task.get_name_label(task)
|
|
status = self._session.xenapi.task.get_status(task)
|
|
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":
|
|
result = self._session.xenapi.task.get_result(task)
|
|
LOG.info(_("Task [%(name)s] %(task)s status:"
|
|
" success %(result)s") % locals())
|
|
done.send(_parse_xmlrpc_value(result))
|
|
else:
|
|
error_info = self._session.xenapi.task.get_error_info(task)
|
|
action["error"] = str(error_info)
|
|
LOG.warn(_("Task [%(name)s] %(task)s status:"
|
|
" %(status)s %(error_info)s") % locals())
|
|
done.send_exception(self.XenAPI.Failure(error_info))
|
|
|
|
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())
|
|
self._stop_loop()
|
|
|
|
def _unwrap_plugin_exceptions(self, func, *args, **kwargs):
|
|
"""Parse exception details"""
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except self.XenAPI.Failure, exc:
|
|
LOG.debug(_("Got exception: %s"), exc)
|
|
if (len(exc.details) == 4 and
|
|
exc.details[0] == 'XENAPI_PLUGIN_EXCEPTION' and
|
|
exc.details[2] == 'Failure'):
|
|
params = None
|
|
try:
|
|
params = eval(exc.details[3])
|
|
except:
|
|
raise exc
|
|
raise self.XenAPI.Failure(params)
|
|
else:
|
|
raise
|
|
except xmlrpclib.ProtocolError, exc:
|
|
LOG.debug(_("Got exception: %s"), exc)
|
|
raise
|
|
|
|
|
|
def _parse_xmlrpc_value(val):
|
|
"""Parse the given value as if it were an XML-RPC value. This is
|
|
sometimes used as the format for the task.result field."""
|
|
if not val:
|
|
return val
|
|
x = xmlrpclib.loads(
|
|
'<?xml version="1.0"?><methodResponse><params><param>' +
|
|
val +
|
|
'</param></params></methodResponse>')
|
|
return x[0][0]
|