x/gantt
Code Issues Proposed changes

Merge trunk

Browse Source
This commit is contained in:
Soren Hansen
2011-03-14 14:21:44 +01:00
parent a063712e3c fba940590c
commit 87d1bb4661
29 changed files with 1418 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
bin/nova-api
View File

@@ -36,49 +36,15 @@ gettext.install('nova', unicode=1)
from nova import flags
from nova import log as logging
from nova import service
from nova import utils
from nova import version
from nova import wsgi
LOG = logging.getLogger('nova.api')
FLAGS = flags.FLAGS
flags.DEFINE_string('ec2_listen', "0.0.0.0",
'IP address for EC2 API to listen')
flags.DEFINE_integer('ec2_listen_port', 8773, 'port for ec2 api to listen')
flags.DEFINE_string('osapi_listen', "0.0.0.0",
'IP address for OpenStack API to listen')
flags.DEFINE_integer('osapi_listen_port', 8774, 'port for os api to listen')
flags.DEFINE_flag(flags.HelpFlag())
flags.DEFINE_flag(flags.HelpshortFlag())
flags.DEFINE_flag(flags.HelpXMLFlag())
API_ENDPOINTS = ['ec2', 'osapi']
def run_app(paste_config_file):
LOG.debug(_("Using paste.deploy config at: %s"), paste_config_file)
apps = []
for api in API_ENDPOINTS:
config = wsgi.load_paste_configuration(paste_config_file, api)
if config is None:
LOG.debug(_("No paste configuration for app: %s"), api)
continue
LOG.debug(_("App Config: %(api)s\n%(config)r") % locals())
LOG.info(_("Running %s API"), api)
app = wsgi.load_paste_app(paste_config_file, api)
apps.append((app, getattr(FLAGS, "%s_listen_port" % api),
getattr(FLAGS, "%s_listen" % api)))
if len(apps) == 0:
LOG.error(_("No known API applications configured in %s."),
paste_config_file)
return
server = wsgi.Server()
for app in apps:
server.start(*app)
server.wait()
if __name__ == '__main__':
utils.default_flagfile()
@@ -90,8 +56,6 @@ if __name__ == '__main__':
for flag in FLAGS:
flag_get = FLAGS.get(flag, None)
LOG.debug("%(flag)s : %(flag_get)s" % locals())
conf = wsgi.paste_config_file('nova-api.conf')
if conf:
run_app(conf)
else:
LOG.error(_("No paste configuration found for: %s"), 'nova-api.conf')
service = service.serve_wsgi(service.ApiService)
service.wait()

171
bin/nova-manage
View File

@@ -55,6 +55,8 @@
import datetime
import gettext
import glob
import json
import os
import re
import sys
@@ -81,7 +83,7 @@ from nova import log as logging
from nova import quota
from nova import rpc
from nova import utils
from nova.api.ec2.cloud import ec2_id_to_id
from nova.api.ec2 import ec2utils
from nova.auth import manager
from nova.cloudpipe import pipelib
from nova.compute import instance_types
@@ -94,6 +96,7 @@ flags.DECLARE('network_size', 'nova.network.manager')
flags.DECLARE('vlan_start', 'nova.network.manager')
flags.DECLARE('vpn_start', 'nova.network.manager')
flags.DECLARE('fixed_range_v6', 'nova.network.manager')
flags.DECLARE('images_path', 'nova.image.local')
flags.DEFINE_flag(flags.HelpFlag())
flags.DEFINE_flag(flags.HelpshortFlag())
flags.DEFINE_flag(flags.HelpXMLFlag())
@@ -104,7 +107,7 @@ def param2id(object_id):
args: [object_id], e.g. 'vol-0000000a' or 'volume-0000000a' or '10'
"""
if '-' in object_id:
return ec2_id_to_id(object_id)
return ec2utils.ec2_id_to_id(object_id)
else:
return int(object_id)
@@ -273,7 +276,7 @@ def _db_error(caught_exception):
print caught_exception
print _("The above error may show that the database has not "
"been created.\nPlease create a database using "
"nova-manage sync db before running this command.")
"'nova-manage db sync' before running this command.")
exit(1)
@@ -434,6 +437,8 @@ class ProjectCommands(object):
"been created.\nPlease create a database by running a "
"nova-api server on this host.")
AccountCommands = ProjectCommands
class FixedIpCommands(object):
"""Class for managing fixed ip."""
@@ -545,6 +550,15 @@ class NetworkCommands(object):
network.dhcp_start,
network.dns)
def delete(self, fixed_range):
"""Deletes a network"""
network = db.network_get_by_cidr(context.get_admin_context(), \
fixed_range)
if network.project_id is not None:
raise ValueError(_('Network must be disassociated from project %s'
' before delete' % network.project_id))
db.network_delete_safe(context.get_admin_context(), network.id)
class ServiceCommands(object):
"""Enable and disable running services"""
@@ -735,8 +749,158 @@ class InstanceTypeCommands(object):
self._print_instance_types(name, inst_types)
class ImageCommands(object):
"""Methods for dealing with a cloud in an odd state"""
def __init__(self, *args, **kwargs):
self.image_service = utils.import_object(FLAGS.image_service)
def _register(self, image_type, disk_format, container_format,
path, owner, name=None, is_public='T',
architecture='x86_64', kernel_id=None, ramdisk_id=None):
meta = {'is_public': True,
'name': name,
'disk_format': disk_format,
'container_format': container_format,
'properties': {'image_state': 'available',
'owner': owner,
'type': image_type,
'architecture': architecture,
'image_location': 'local',
'is_public': (is_public == 'T')}}
print image_type, meta
if kernel_id:
meta['properties']['kernel_id'] = int(kernel_id)
if ramdisk_id:
meta['properties']['ramdisk_id'] = int(ramdisk_id)
elevated = context.get_admin_context()
try:
with open(path) as ifile:
image = self.image_service.create(elevated, meta, ifile)
new = image['id']
print _("Image registered to %(new)s (%(new)08x).") % locals()
return new
except Exception as exc:
print _("Failed to register %(path)s: %(exc)s") % locals()
def all_register(self, image, kernel, ramdisk, owner, name=None,
is_public='T', architecture='x86_64'):
"""Uploads an image, kernel, and ramdisk into the image_service
arguments: image kernel ramdisk owner [name] [is_public='T']
[architecture='x86_64']"""
kernel_id = self.kernel_register(kernel, owner, None,
is_public, architecture)
ramdisk_id = self.ramdisk_register(ramdisk, owner, None,
is_public, architecture)
self.image_register(image, owner, name, is_public,
architecture, kernel_id, ramdisk_id)
def image_register(self, path, owner, name=None, is_public='T',
architecture='x86_64', kernel_id=None, ramdisk_id=None,
disk_format='ami', container_format='ami'):
"""Uploads an image into the image_service
arguments: path owner [name] [is_public='T'] [architecture='x86_64']
[kernel_id=None] [ramdisk_id=None]
[disk_format='ami'] [container_format='ami']"""
return self._register('machine', disk_format, container_format, path,
owner, name, is_public, architecture,
kernel_id, ramdisk_id)
def kernel_register(self, path, owner, name=None, is_public='T',
architecture='x86_64'):
"""Uploads a kernel into the image_service
arguments: path owner [name] [is_public='T'] [architecture='x86_64']
"""
return self._register('kernel', 'aki', 'aki', path, owner, name,
is_public, architecture)
def ramdisk_register(self, path, owner, name=None, is_public='T',
architecture='x86_64'):
"""Uploads a ramdisk into the image_service
arguments: path owner [name] [is_public='T'] [architecture='x86_64']
"""
return self._register('ramdisk', 'ari', 'ari', path, owner, name,
is_public, architecture)
def _lookup(self, old_image_id):
try:
internal_id = ec2utils.ec2_id_to_id(old_image_id)
image = self.image_service.show(context, internal_id)
except exception.NotFound:
image = self.image_service.show_by_name(context, old_image_id)
return image['id']
def _old_to_new(self, old):
mapping = {'machine': 'ami',
'kernel': 'aki',
'ramdisk': 'ari'}
container_format = mapping[old['type']]
disk_format = container_format
new = {'disk_format': disk_format,
'container_format': container_format,
'is_public': True,
'name': old['imageId'],
'properties': {'image_state': old['imageState'],
'owner': old['imageOwnerId'],
'architecture': old['architecture'],
'type': old['type'],
'image_location': old['imageLocation'],
'is_public': old['isPublic']}}
if old.get('kernelId'):
new['properties']['kernel_id'] = self._lookup(old['kernelId'])
if old.get('ramdiskId'):
new['properties']['ramdisk_id'] = self._lookup(old['ramdiskId'])
return new
def _convert_images(self, images):
elevated = context.get_admin_context()
for image_path, image_metadata in images.iteritems():
meta = self._old_to_new(image_metadata)
old = meta['name']
try:
with open(image_path) as ifile:
image = self.image_service.create(elevated, meta, ifile)
new = image['id']
print _("Image %(old)s converted to " \
"%(new)s (%(new)08x).") % locals()
except Exception as exc:
print _("Failed to convert %(old)s: %(exc)s") % locals()
def convert(self, directory):
"""Uploads old objectstore images in directory to new service
arguments: directory"""
machine_images = {}
other_images = {}
directory = os.path.abspath(directory)
# NOTE(vish): If we're importing from the images path dir, attempt
# to move the files out of the way before importing
# so we aren't writing to the same directory. This
# may fail if the dir was a mointpoint.
if (FLAGS.image_service == 'nova.image.local.LocalImageService'
and directory == os.path.abspath(FLAGS.images_path)):
new_dir = "%s_bak" % directory
os.move(directory, new_dir)
os.mkdir(directory)
directory = new_dir
for fn in glob.glob("%s/*/info.json" % directory):
try:
image_path = os.path.join(fn.rpartition('/')[0], 'image')
with open(fn) as metadata_file:
image_metadata = json.load(metadata_file)
if image_metadata['type'] == 'machine':
machine_images[image_path] = image_metadata
else:
other_images[image_path] = image_metadata
except Exception as exc:
print _("Failed to load %(fn)s.") % locals()
# NOTE(vish): do kernels and ramdisks first so images
self._convert_images(other_images)
self._convert_images(machine_images)
CATEGORIES = [
('user', UserCommands),
('account', AccountCommands),
('project', ProjectCommands),
('role', RoleCommands),
('shell', ShellCommands),
@@ -749,6 +913,7 @@ CATEGORIES = [
('db', DbCommands),
('volume', VolumeCommands),
('instance_type', InstanceTypeCommands),
('image', ImageCommands),
('flavor', InstanceTypeCommands)]

3
doc/ext/nova_autodoc.py
View File

@@ -8,5 +8,6 @@ from nova import utils
def setup(app):
rootdir = os.path.abspath(app.srcdir + '/..')
print "**Autodocumenting from %s" % rootdir
rv = utils.execute('cd %s && ./generate_autodoc_index.sh' % rootdir)
os.chdir(rootdir)
rv = utils.execute('./generate_autodoc_index.sh')
print rv[0]

30
doc/source/man/novamanage.rst
View File

@@ -173,7 +173,10 @@ Nova Floating IPs
``nova-manage floating create <host> <ip_range>``
Creates floating IP addresses for the named host by the given range.
floating delete <ip_range> Deletes floating IP addresses in the range given.
``nova-manage floating delete <ip_range>``
Deletes floating IP addresses in the range given.
``nova-manage floating list``
@@ -193,7 +196,7 @@ Nova Flavor
``nova-manage flavor create <name> <memory> <vCPU> <local_storage> <flavorID> <(optional) swap> <(optional) RXTX Quota> <(optional) RXTX Cap>``
creates a flavor with the following positional arguments:
* memory (expressed in megabytes)
* memory (expressed in megabytes)
* vcpu(s) (integer)
* local storage (expressed in gigabytes)
* flavorid (unique integer)
@@ -209,12 +212,33 @@ Nova Flavor
Purges the flavor with the name <name>. This removes this flavor from the database.
Nova Instance_type
~~~~~~~~~~~~~~~~~~
The instance_type command is provided as an alias for the flavor command. All the same subcommands and arguments from nova-manage flavor can be used.
Nova Images
~~~~~~~~~~~
``nova-manage image image_register <path> <owner>``
Registers an image with the image service.
``nova-manage image kernel_register <path> <owner>``
Registers a kernel with the image service.
``nova-manage image ramdisk_register <path> <owner>``
Registers a ramdisk with the image service.
``nova-manage image all_register <image_path> <kernel_path> <ramdisk_path> <owner>``
Registers an image kernel and ramdisk with the image service.
``nova-manage image convert <directory>``
Converts all images in directory from the old (Bexar) format to the new format.
FILES
========

23
doc/source/runnova/nova.manage.rst
View File

@@ -182,6 +182,29 @@ Nova Floating IPs
Displays a list of all floating IP addresses.
Nova Images
~~~~~~~~~~~
``nova-manage image image_register <path> <owner>``
Registers an image with the image service.
``nova-manage image kernel_register <path> <owner>``
Registers a kernel with the image service.
``nova-manage image ramdisk_register <path> <owner>``
Registers a ramdisk with the image service.
``nova-manage image all_register <image_path> <kernel_path> <ramdisk_path> <owner>``
Registers an image kernel and ramdisk with the image service.
``nova-manage image convert <directory>``
Converts all images in directory from the old (Bexar) format to the new format.
Concept: Flags
--------------

4
nova/exception.py
View File

@@ -88,6 +88,10 @@ class InvalidInputException(Error):
pass
class InvalidContentType(Error):
pass
class TimeoutException(Error):
pass

1
nova/fakerabbit.py
View File

@@ -48,7 +48,6 @@ class Exchange(object):
nm = self.name
LOG.debug(_('(%(nm)s) publish (key: %(routing_key)s)'
' %(message)s') % locals())
routing_key = routing_key.split('.')[0]
if routing_key in self._routes:
for f in self._routes[routing_key]:
LOG.debug(_('Publishing to route %s'), f)

8
nova/flags.py
View File

@@ -321,6 +321,8 @@ DEFINE_integer('auth_token_ttl', 3600, 'Seconds for auth tokens to linger')
DEFINE_string('state_path', os.path.join(os.path.dirname(__file__), '../'),
"Top-level directory for maintaining nova's state")
DEFINE_string('lock_path', os.path.join(os.path.dirname(__file__), '../'),
"Directory for lock files")
DEFINE_string('logdir', None, 'output to a per-service log file in named '
'directory')
@@ -346,7 +348,7 @@ DEFINE_string('scheduler_manager', 'nova.scheduler.manager.SchedulerManager',
'Manager for scheduler')
# The service to use for image search and retrieval
DEFINE_string('image_service', 'nova.image.s3.S3ImageService',
DEFINE_string('image_service', 'nova.image.local.LocalImageService',
'The service to use for retrieving and searching for images.')
DEFINE_string('host', socket.gethostname(),
@@ -354,3 +356,7 @@ DEFINE_string('host', socket.gethostname(),
DEFINE_string('node_availability_zone', 'nova',
'availability zone of this node')
DEFINE_string('zone_name', 'nova', 'name of this zone')
DEFINE_string('zone_capabilities', 'kypervisor:xenserver;os:linux',
'Key/Value tags which represent capabilities of this zone')

5
nova/log.py
View File

@@ -266,7 +266,10 @@ class NovaRootLogger(NovaLogger):
def handle_exception(type, value, tb):
logging.root.critical(str(value), exc_info=(type, value, tb))
extra = {}
if FLAGS.verbose:
extra['exc_info'] = (type, value, tb)
logging.root.critical(str(value), **extra)
def reset():

2
nova/rpc.py
View File

@@ -123,7 +123,7 @@ class Consumer(messaging.Consumer):
LOG.error(_("Reconnected to queue"))
self.failed_connection = False
# NOTE(vish): This is catching all errors because we really don't
# exceptions to be logged 10 times a second if some
# want exceptions to be logged 10 times a second if some
# persistent failure occurs.
except Exception: # pylint: disable-msg=W0703
if not self.failed_connection:

49
nova/scheduler/api.py Normal file
View File

@@ -0,0 +1,49 @@
# Copyright (c) 2011 Openstack, LLC.
# 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.
"""
Handles all requests relating to schedulers.
"""
from nova import flags
from nova import log as logging
from nova import rpc
FLAGS = flags.FLAGS
LOG = logging.getLogger('nova.scheduler.api')
class API(object):
"""API for interacting with the scheduler."""
def _call_scheduler(self, method, context, params=None):
"""Generic handler for RPC calls to the scheduler.
:param params: Optional dictionary of arguments to be passed to the
scheduler worker
:retval: Result returned by scheduler worker
"""
if not params:
params = {}
queue = FLAGS.scheduler_topic
kwargs = {'method': method, 'args': params}
return rpc.call(context, queue, kwargs)
def get_zone_list(self, context):
items = self._call_scheduler('get_zone_list', context)
for item in items:
item['api_url'] = item['api_url'].replace('\\/', '/')
return items

10
nova/scheduler/manager.py
View File

@@ -29,6 +29,7 @@ from nova import log as logging
from nova import manager
from nova import rpc
from nova import utils
from nova.scheduler import zone_manager
LOG = logging.getLogger('nova.scheduler.manager')
FLAGS = flags.FLAGS
@@ -43,12 +44,21 @@ class SchedulerManager(manager.Manager):
if not scheduler_driver:
scheduler_driver = FLAGS.scheduler_driver
self.driver = utils.import_object(scheduler_driver)
self.zone_manager = zone_manager.ZoneManager()
super(SchedulerManager, self).__init__(*args, **kwargs)
def __getattr__(self, key):
"""Converts all method calls to use the schedule method"""
return functools.partial(self._schedule, key)
def periodic_tasks(self, context=None):
"""Poll child zones periodically to get status."""
self.zone_manager.ping(context)
def get_zone_list(self, context=None):
"""Get a list of zones from the ZoneManager."""
return self.zone_manager.get_zone_list()
def _schedule(self, method, context, topic, *args, **kwargs):
"""Tries to call schedule_* method on the driver to retrieve host.

143
nova/scheduler/zone_manager.py Normal file
View File

@@ -0,0 +1,143 @@
# Copyright (c) 2011 Openstack, LLC.
# 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.
"""
ZoneManager oversees all communications with child Zones.
"""
import novaclient
import thread
import traceback
from datetime import datetime
from eventlet import greenpool
from nova import db
from nova import flags
from nova import log as logging
FLAGS = flags.FLAGS
flags.DEFINE_integer('zone_db_check_interval', 60,
'Seconds between getting fresh zone info from db.')
flags.DEFINE_integer('zone_failures_to_offline', 3,
'Number of consecutive errors before marking zone offline')
class ZoneState(object):
"""Holds the state of all connected child zones."""
def __init__(self):
self.is_active = True
self.name = None
self.capabilities = None
self.attempt = 0
self.last_seen = datetime.min
self.last_exception = None
self.last_exception_time = None
def update_credentials(self, zone):
"""Update zone credentials from db"""
self.zone_id = zone.id
self.api_url = zone.api_url
self.username = zone.username
self.password = zone.password
def update_metadata(self, zone_metadata):
"""Update zone metadata after successful communications with
child zone."""
self.last_seen = datetime.now()
self.attempt = 0
self.name = zone_metadata["name"]
self.capabilities = zone_metadata["capabilities"]
self.is_active = True
def to_dict(self):
return dict(name=self.name, capabilities=self.capabilities,
is_active=self.is_active, api_url=self.api_url,
id=self.zone_id)
def log_error(self, exception):
"""Something went wrong. Check to see if zone should be
marked as offline."""
self.last_exception = exception
self.last_exception_time = datetime.now()
api_url = self.api_url
logging.warning(_("'%(exception)s' error talking to "
"zone %(api_url)s") % locals())
max_errors = FLAGS.zone_failures_to_offline
self.attempt += 1
if self.attempt >= max_errors:
self.is_active = False
logging.error(_("No answer from zone %(api_url)s "
"after %(max_errors)d "
"attempts. Marking inactive.") % locals())
def _call_novaclient(zone):
"""Call novaclient. Broken out for testing purposes."""
client = novaclient.OpenStack(zone.username, zone.password, zone.api_url)
return client.zones.info()._info
def _poll_zone(zone):
"""Eventlet worker to poll a zone."""
logging.debug(_("Polling zone: %s") % zone.api_url)
try:
zone.update_metadata(_call_novaclient(zone))
except Exception, e:
zone.log_error(traceback.format_exc())
class ZoneManager(object):
"""Keeps the zone states updated."""
def __init__(self):
self.last_zone_db_check = datetime.min
self.zone_states = {}
self.green_pool = greenpool.GreenPool()
def get_zone_list(self):
"""Return the list of zones we know about."""
return [zone.to_dict() for zone in self.zone_states.values()]
def _refresh_from_db(self, context):
"""Make our zone state map match the db."""
# Add/update existing zones ...
zones = db.zone_get_all(context)
existing = self.zone_states.keys()
db_keys = []
for zone in zones:
db_keys.append(zone.id)
if zone.id not in existing:
self.zone_states[zone.id] = ZoneState()
self.zone_states[zone.id].update_credentials(zone)
# Cleanup zones removed from db ...
keys = self.zone_states.keys() # since we're deleting
for zone_id in keys:
if zone_id not in db_keys:
del self.zone_states[zone_id]
def _poll_zones(self, context):
"""Try to connect to each child zone and get update."""
self.green_pool.imap(_poll_zone, self.zone_states.values())
def ping(self, context=None):
"""Ping should be called periodically to update zone status."""
diff = datetime.now() - self.last_zone_db_check
if diff.seconds >= FLAGS.zone_db_check_interval:
logging.debug(_("Updating zone cache from db."))
self.last_zone_db_check = datetime.now()
self._refresh_from_db(context)
self._poll_zones(context)

88
nova/service.py
View File

@@ -2,6 +2,7 @@
# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# Copyright 2011 Justin Santa Barbara
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@@ -39,6 +40,7 @@ from nova import flags
from nova import rpc
from nova import utils
from nova import version
from nova import wsgi
FLAGS = flags.FLAGS
@@ -48,6 +50,14 @@ flags.DEFINE_integer('report_interval', 10,
flags.DEFINE_integer('periodic_interval', 60,
'seconds between running periodic tasks',
lower_bound=1)
flags.DEFINE_string('ec2_listen', "0.0.0.0",
'IP address for EC2 API to listen')
flags.DEFINE_integer('ec2_listen_port', 8773, 'port for ec2 api to listen')
flags.DEFINE_string('osapi_listen', "0.0.0.0",
'IP address for OpenStack API to listen')
flags.DEFINE_integer('osapi_listen_port', 8774, 'port for os api to listen')
flags.DEFINE_string('api_paste_config', "api-paste.ini",
'File name for the paste.deploy config for nova-api')
class Service(object):
@@ -210,6 +220,41 @@ class Service(object):
logging.exception(_("model server went away"))
class WsgiService(object):
"""Base class for WSGI based services.
For each api you define, you must also define these flags:
:<api>_listen: The address on which to listen
:<api>_listen_port: The port on which to listen
"""
def __init__(self, conf, apis):
self.conf = conf
self.apis = apis
self.wsgi_app = None
def start(self):
self.wsgi_app = _run_wsgi(self.conf, self.apis)
def wait(self):
self.wsgi_app.wait()
class ApiService(WsgiService):
"""Class for our nova-api service"""
@classmethod
def create(cls, conf=None):
if not conf:
conf = wsgi.paste_config_file(FLAGS.api_paste_config)
if not conf:
message = (_("No paste configuration found for: %s"),
FLAGS.api_paste_config)
raise exception.Error(message)
api_endpoints = ['ec2', 'osapi']
service = cls(conf, api_endpoints)
return service
def serve(*services):
try:
if not services:
@@ -239,3 +284,46 @@ def serve(*services):
def wait():
while True:
greenthread.sleep(5)
def serve_wsgi(cls, conf=None):
try:
service = cls.create(conf)
except Exception:
logging.exception('in WsgiService.create()')
raise
finally:
# After we've loaded up all our dynamic bits, check
# whether we should print help
flags.DEFINE_flag(flags.HelpFlag())
flags.DEFINE_flag(flags.HelpshortFlag())
flags.DEFINE_flag(flags.HelpXMLFlag())
FLAGS.ParseNewFlags()
service.start()
return service
def _run_wsgi(paste_config_file, apis):
logging.debug(_("Using paste.deploy config at: %s"), paste_config_file)
apps = []
for api in apis:
config = wsgi.load_paste_configuration(paste_config_file, api)
if config is None:
logging.debug(_("No paste configuration for app: %s"), api)
continue
logging.debug(_("App Config: %(api)s\n%(config)r") % locals())
logging.info(_("Running %s API"), api)
app = wsgi.load_paste_app(paste_config_file, api)
apps.append((app, getattr(FLAGS, "%s_listen_port" % api),
getattr(FLAGS, "%s_listen" % api)))
if len(apps) == 0:
logging.error(_("No known API applications configured in %s."),
paste_config_file)
return
server = wsgi.Server()
for app in apps:
server.start(*app)
return server

1
nova/tests/fake_flags.py
View File

@@ -32,6 +32,7 @@ flags.DECLARE('fake_network', 'nova.network.manager')
FLAGS.network_size = 8
FLAGS.num_networks = 2
FLAGS.fake_network = True
FLAGS.image_service = 'nova.image.local.LocalImageService'
flags.DECLARE('num_shelves', 'nova.volume.driver')
flags.DECLARE('blades_per_shelf', 'nova.volume.driver')
flags.DECLARE('iscsi_num_targets', 'nova.volume.driver')

30
nova/tests/test_cloud.py
View File

@@ -38,6 +38,8 @@ from nova import test
from nova.auth import manager
from nova.compute import power_state
from nova.api.ec2 import cloud
from nova.api.ec2 import ec2utils
from nova.image import local
from nova.objectstore import image
@@ -76,6 +78,12 @@ class CloudTestCase(test.TestCase):
project=self.project)
host = self.network.get_network_host(self.context.elevated())
def fake_show(meh, context, id):
return {'id': 1, 'properties': {'kernel_id': 1, 'ramdisk_id': 1}}
self.stubs.Set(local.LocalImageService, 'show', fake_show)
self.stubs.Set(local.LocalImageService, 'show_by_name', fake_show)
def tearDown(self):
network_ref = db.project_get_network(self.context,
self.project.id)
@@ -122,7 +130,7 @@ class CloudTestCase(test.TestCase):
self.cloud.allocate_address(self.context)
inst = db.instance_create(self.context, {'host': self.compute.host})
fixed = self.network.allocate_fixed_ip(self.context, inst['id'])
ec2_id = cloud.id_to_ec2_id(inst['id'])
ec2_id = ec2utils.id_to_ec2_id(inst['id'])
self.cloud.associate_address(self.context,
instance_id=ec2_id,
public_ip=address)
@@ -158,12 +166,12 @@ class CloudTestCase(test.TestCase):
vol2 = db.volume_create(self.context, {})
result = self.cloud.describe_volumes(self.context)
self.assertEqual(len(result['volumeSet']), 2)
volume_id = cloud.id_to_ec2_id(vol2['id'], 'vol-%08x')
volume_id = ec2utils.id_to_ec2_id(vol2['id'], 'vol-%08x')
result = self.cloud.describe_volumes(self.context,
volume_id=[volume_id])
self.assertEqual(len(result['volumeSet']), 1)
self.assertEqual(
cloud.ec2_id_to_id(result['volumeSet'][0]['volumeId']),
ec2utils.ec2_id_to_id(result['volumeSet'][0]['volumeId']),
vol2['id'])
db.volume_destroy(self.context, vol1['id'])
db.volume_destroy(self.context, vol2['id'])
@@ -188,8 +196,10 @@ class CloudTestCase(test.TestCase):
def test_describe_instances(self):
"""Makes sure describe_instances works and filters results."""
inst1 = db.instance_create(self.context, {'reservation_id': 'a',
'image_id': 1,
'host': 'host1'})
inst2 = db.instance_create(self.context, {'reservation_id': 'a',
'image_id': 1,
'host': 'host2'})
comp1 = db.service_create(self.context, {'host': 'host1',
'availability_zone': 'zone1',
@@ -200,7 +210,7 @@ class CloudTestCase(test.TestCase):
result = self.cloud.describe_instances(self.context)
result = result['reservationSet'][0]
self.assertEqual(len(result['instancesSet']), 2)
instance_id = cloud.id_to_ec2_id(inst2['id'])
instance_id = ec2utils.id_to_ec2_id(inst2['id'])
result = self.cloud.describe_instances(self.context,
instance_id=[instance_id])
result = result['reservationSet'][0]
@@ -215,10 +225,9 @@ class CloudTestCase(test.TestCase):
db.service_destroy(self.context, comp2['id'])
def test_console_output(self):
image_id = FLAGS.default_image
instance_type = FLAGS.default_instance_type
max_count = 1
kwargs = {'image_id': image_id,
kwargs = {'image_id': 'ami-1',
'instance_type': instance_type,
'max_count': max_count}
rv = self.cloud.run_instances(self.context, **kwargs)
@@ -234,8 +243,7 @@ class CloudTestCase(test.TestCase):
greenthread.sleep(0.3)
def test_ajax_console(self):
image_id = FLAGS.default_image
kwargs = {'image_id': image_id}
kwargs = {'image_id': 'ami-1'}
rv = self.cloud.run_instances(self.context, **kwargs)
instance_id = rv['instancesSet'][0]['instanceId']
greenthread.sleep(0.3)
@@ -347,7 +355,7 @@ class CloudTestCase(test.TestCase):
def test_update_of_instance_display_fields(self):
inst = db.instance_create(self.context, {})
ec2_id = cloud.id_to_ec2_id(inst['id'])
ec2_id = ec2utils.id_to_ec2_id(inst['id'])
self.cloud.update_instance(self.context, ec2_id,
display_name='c00l 1m4g3')
inst = db.instance_get(self.context, inst['id'])
@@ -365,7 +373,7 @@ class CloudTestCase(test.TestCase):
def test_update_of_volume_display_fields(self):
vol = db.volume_create(self.context, {})
self.cloud.update_volume(self.context,
cloud.id_to_ec2_id(vol['id'], 'vol-%08x'),
ec2utils.id_to_ec2_id(vol['id'], 'vol-%08x'),
display_name='c00l v0lum3')
vol = db.volume_get(self.context, vol['id'])
self.assertEqual('c00l v0lum3', vol['display_name'])
@@ -374,7 +382,7 @@ class CloudTestCase(test.TestCase):
def test_update_of_volume_wont_update_private_fields(self):
vol = db.volume_create(self.context, {})
self.cloud.update_volume(self.context,
cloud.id_to_ec2_id(vol['id'], 'vol-%08x'),
ec2utils.id_to_ec2_id(vol['id'], 'vol-%08x'),
mountpoint='/not/here')
vol = db.volume_get(self.context, vol['id'])
self.assertEqual(None, vol['mountpoint'])

33
nova/tests/test_compute.py
View File

@@ -31,7 +31,7 @@ from nova import test
from nova import utils
from nova.auth import manager
from nova.compute import instance_types
from nova.image import local
LOG = logging.getLogger('nova.tests.compute')
FLAGS = flags.FLAGS
@@ -52,15 +52,20 @@ class ComputeTestCase(test.TestCase):
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.context = context.RequestContext('fake', 'fake', False)
def fake_show(meh, context, id):
return {'id': 1, 'properties': {'kernel_id': 1, 'ramdisk_id': 1}}
self.stubs.Set(local.LocalImageService, 'show', fake_show)
def tearDown(self):
self.manager.delete_user(self.user)
self.manager.delete_project(self.project)
super(ComputeTestCase, self).tearDown()
def _create_instance(self):
def _create_instance(self, params={}):
"""Create a test instance"""
inst = {}
inst['image_id'] = 'ami-test'
inst['image_id'] = 1
inst['reservation_id'] = 'r-fakeres'
inst['launch_time'] = '10'
inst['user_id'] = self.user.id
@@ -68,6 +73,7 @@ class ComputeTestCase(test.TestCase):
inst['instance_type'] = 'm1.tiny'
inst['mac_address'] = utils.generate_mac()
inst['ami_launch_index'] = 0
inst.update(params)
return db.instance_create(self.context, inst)['id']
def _create_group(self):
@@ -268,9 +274,30 @@ class ComputeTestCase(test.TestCase):
self.compute.terminate_instance(self.context, instance_id)
def test_resize_instance(self):
"""Ensure instance can be migrated/resized"""
instance_id = self._create_instance()
context = self.context.elevated()
self.compute.run_instance(self.context, instance_id)
db.instance_update(self.context, instance_id, {'host': 'foo'})
self.compute.prep_resize(context, instance_id)
migration_ref = db.migration_get_by_instance_and_status(context,
instance_id, 'pre-migrating')
self.compute.resize_instance(context, instance_id,
migration_ref['id'])
self.compute.terminate_instance(context, instance_id)
def test_get_by_flavor_id(self):
type = instance_types.get_by_flavor_id(1)
self.assertEqual(type, 'm1.tiny')
def test_resize_same_source_fails(self):
"""Ensure instance fails to migrate when source and destination are
the same host"""
instance_id = self._create_instance()
self.compute.run_instance(self.context, instance_id)
self.assertRaises(exception.Error, self.compute.prep_resize,
self.context, instance_id)
self.compute.terminate_instance(self.context, instance_id)
type = instance_types.get_by_flavor_id("1")
self.assertEqual(type, 'm1.tiny')

2
nova/tests/test_console.py
View File

@@ -57,7 +57,7 @@ class ConsoleTestCase(test.TestCase):
inst = {}
#inst['host'] = self.host
#inst['name'] = 'instance-1234'
inst['image_id'] = 'ami-test'
inst['image_id'] = 1
inst['reservation_id'] = 'r-fakeres'
inst['launch_time'] = '10'
inst['user_id'] = self.user.id

6
nova/tests/test_direct.py
View File

@@ -59,6 +59,7 @@ class DirectTestCase(test.TestCase):
req.headers['X-OpenStack-User'] = 'user1'
req.headers['X-OpenStack-Project'] = 'proj1'
resp = req.get_response(self.auth_router)
self.assertEqual(resp.status_int, 200)
data = json.loads(resp.body)
self.assertEqual(data['user'], 'user1')
self.assertEqual(data['project'], 'proj1')
@@ -69,6 +70,7 @@ class DirectTestCase(test.TestCase):
req.method = 'POST'
req.body = 'json=%s' % json.dumps({'data': 'foo'})
resp = req.get_response(self.router)
self.assertEqual(resp.status_int, 200)
resp_parsed = json.loads(resp.body)
self.assertEqual(resp_parsed['data'], 'foo')
@@ -78,6 +80,7 @@ class DirectTestCase(test.TestCase):
req.method = 'POST'
req.body = 'data=foo'
resp = req.get_response(self.router)
self.assertEqual(resp.status_int, 200)
resp_parsed = json.loads(resp.body)
self.assertEqual(resp_parsed['data'], 'foo')
@@ -90,8 +93,7 @@ class DirectTestCase(test.TestCase):
class DirectCloudTestCase(test_cloud.CloudTestCase):
def setUp(self):
super(DirectCloudTestCase, self).setUp()
compute_handle = compute.API(image_service=self.cloud.image_service,
network_api=self.cloud.network_api,
compute_handle = compute.API(network_api=self.cloud.network_api,
volume_api=self.cloud.volume_api)
direct.register_service('compute', compute_handle)
self.router = direct.JsonParamsMiddleware(direct.Router())

48
nova/tests/test_misc.py
View File

@@ -14,10 +14,12 @@
# License for the specific language governing permissions and limitations
# under the License.
import errno
import os
import select
from nova import test
from nova.utils import parse_mailmap, str_dict_replace
from nova.utils import parse_mailmap, str_dict_replace, synchronized
class ProjectTestCase(test.TestCase):
@@ -55,3 +57,47 @@ class ProjectTestCase(test.TestCase):
'%r not listed in Authors' % missing)
finally:
tree.unlock()
class LockTestCase(test.TestCase):
def test_synchronized_wrapped_function_metadata(self):
@synchronized('whatever')
def foo():
"""Bar"""
pass
self.assertEquals(foo.__doc__, 'Bar', "Wrapped function's docstring "
"got lost")
self.assertEquals(foo.__name__, 'foo', "Wrapped function's name "
"got mangled")
def test_synchronized(self):
rpipe1, wpipe1 = os.pipe()
rpipe2, wpipe2 = os.pipe()
@synchronized('testlock')
def f(rpipe, wpipe):
try:
os.write(wpipe, "foo")
except OSError, e:
self.assertEquals(e.errno, errno.EPIPE)
return
rfds, _, __ = select.select([rpipe], [], [], 1)
self.assertEquals(len(rfds), 0, "The other process, which was"
" supposed to be locked, "
"wrote on its end of the "
"pipe")
os.close(rpipe)
pid = os.fork()
if pid > 0:
os.close(wpipe1)
os.close(rpipe2)
f(rpipe1, wpipe2)
else:
os.close(rpipe1)
os.close(wpipe2)
f(rpipe2, wpipe1)
os._exit(0)

157
nova/tests/test_network.py
View File

@@ -36,6 +36,147 @@ FLAGS = flags.FLAGS
LOG = logging.getLogger('nova.tests.network')
class IptablesManagerTestCase(test.TestCase):
sample_filter = ['#Generated by iptables-save on Fri Feb 18 15:17:05 2011',
'*filter',
':INPUT ACCEPT [2223527:305688874]',
':FORWARD ACCEPT [0:0]',
':OUTPUT ACCEPT [2172501:140856656]',
':nova-compute-FORWARD - [0:0]',
':nova-compute-INPUT - [0:0]',
':nova-compute-local - [0:0]',
':nova-compute-OUTPUT - [0:0]',
':nova-filter-top - [0:0]',
'-A FORWARD -j nova-filter-top ',
'-A OUTPUT -j nova-filter-top ',
'-A nova-filter-top -j nova-compute-local ',
'-A INPUT -j nova-compute-INPUT ',
'-A OUTPUT -j nova-compute-OUTPUT ',
'-A FORWARD -j nova-compute-FORWARD ',
'-A INPUT -i virbr0 -p udp -m udp --dport 53 -j ACCEPT ',
'-A INPUT -i virbr0 -p tcp -m tcp --dport 53 -j ACCEPT ',
'-A INPUT -i virbr0 -p udp -m udp --dport 67 -j ACCEPT ',
'-A INPUT -i virbr0 -p tcp -m tcp --dport 67 -j ACCEPT ',
'-A FORWARD -s 192.168.122.0/24 -i virbr0 -j ACCEPT ',
'-A FORWARD -i virbr0 -o virbr0 -j ACCEPT ',
'-A FORWARD -o virbr0 -j REJECT --reject-with '
'icmp-port-unreachable ',
'-A FORWARD -i virbr0 -j REJECT --reject-with '
'icmp-port-unreachable ',
'COMMIT',
'# Completed on Fri Feb 18 15:17:05 2011']
sample_nat = ['# Generated by iptables-save on Fri Feb 18 15:17:05 2011',
'*nat',
':PREROUTING ACCEPT [3936:762355]',
':INPUT ACCEPT [2447:225266]',
':OUTPUT ACCEPT [63491:4191863]',
':POSTROUTING ACCEPT [63112:4108641]',
':nova-compute-OUTPUT - [0:0]',
':nova-compute-floating-ip-snat - [0:0]',
':nova-compute-SNATTING - [0:0]',
':nova-compute-PREROUTING - [0:0]',
':nova-compute-POSTROUTING - [0:0]',
':nova-postrouting-bottom - [0:0]',
'-A PREROUTING -j nova-compute-PREROUTING ',
'-A OUTPUT -j nova-compute-OUTPUT ',
'-A POSTROUTING -j nova-compute-POSTROUTING ',
'-A POSTROUTING -j nova-postrouting-bottom ',
'-A nova-postrouting-bottom -j nova-compute-SNATTING ',
'-A nova-compute-SNATTING -j nova-compute-floating-ip-snat ',
'COMMIT',
'# Completed on Fri Feb 18 15:17:05 2011']
def setUp(self):
super(IptablesManagerTestCase, self).setUp()
self.manager = linux_net.IptablesManager()
def test_filter_rules_are_wrapped(self):
current_lines = self.sample_filter
table = self.manager.ipv4['filter']
table.add_rule('FORWARD', '-s 1.2.3.4/5 -j DROP')
new_lines = self.manager._modify_rules(current_lines, table)
self.assertTrue('-A run_tests.py-FORWARD '
'-s 1.2.3.4/5 -j DROP' in new_lines)
table.remove_rule('FORWARD', '-s 1.2.3.4/5 -j DROP')
new_lines = self.manager._modify_rules(current_lines, table)
self.assertTrue('-A run_tests.py-FORWARD '
'-s 1.2.3.4/5 -j DROP' not in new_lines)
def test_nat_rules(self):
current_lines = self.sample_nat
new_lines = self.manager._modify_rules(current_lines,
self.manager.ipv4['nat'])
for line in [':nova-compute-OUTPUT - [0:0]',
':nova-compute-floating-ip-snat - [0:0]',
':nova-compute-SNATTING - [0:0]',
':nova-compute-PREROUTING - [0:0]',
':nova-compute-POSTROUTING - [0:0]']:
self.assertTrue(line in new_lines, "One of nova-compute's chains "
"went missing.")
seen_lines = set()
for line in new_lines:
line = line.strip()
self.assertTrue(line not in seen_lines,
"Duplicate line: %s" % line)
seen_lines.add(line)
last_postrouting_line = ''
for line in new_lines:
if line.startswith('-A POSTROUTING'):
last_postrouting_line = line
self.assertTrue('-j nova-postrouting-bottom' in last_postrouting_line,
"Last POSTROUTING rule does not jump to "
"nova-postouting-bottom: %s" % last_postrouting_line)
for chain in ['POSTROUTING', 'PREROUTING', 'OUTPUT']:
self.assertTrue('-A %s -j run_tests.py-%s' \
% (chain, chain) in new_lines,
"Built-in chain %s not wrapped" % (chain,))
def test_filter_rules(self):
current_lines = self.sample_filter
new_lines = self.manager._modify_rules(current_lines,
self.manager.ipv4['filter'])
for line in [':nova-compute-FORWARD - [0:0]',
':nova-compute-INPUT - [0:0]',
':nova-compute-local - [0:0]',
':nova-compute-OUTPUT - [0:0]']:
self.assertTrue(line in new_lines, "One of nova-compute's chains"
" went missing.")
seen_lines = set()
for line in new_lines:
line = line.strip()
self.assertTrue(line not in seen_lines,
"Duplicate line: %s" % line)
seen_lines.add(line)
for chain in ['FORWARD', 'OUTPUT']:
for line in new_lines:
if line.startswith('-A %s' % chain):
self.assertTrue('-j nova-filter-top' in line,
"First %s rule does not "
"jump to nova-filter-top" % chain)
break
self.assertTrue('-A nova-filter-top '
'-j run_tests.py-local' in new_lines,
"nova-filter-top does not jump to wrapped local chain")
for chain in ['INPUT', 'OUTPUT', 'FORWARD']:
self.assertTrue('-A %s -j run_tests.py-%s' \
% (chain, chain) in new_lines,
"Built-in chain %s not wrapped" % (chain,))
class NetworkTestCase(test.TestCase):
"""Test cases for network code"""
def setUp(self):
@@ -370,13 +511,13 @@ def lease_ip(private_ip):
private_ip)
instance_ref = db.fixed_ip_get_instance(context.get_admin_context(),
private_ip)
cmd = "%s add %s %s fake" % (binpath('nova-dhcpbridge'),
instance_ref['mac_address'],
private_ip)
cmd = (binpath('nova-dhcpbridge'), 'add',
instance_ref['mac_address'],
private_ip, 'fake')
env = {'DNSMASQ_INTERFACE': network_ref['bridge'],
'TESTING': '1',
'FLAGFILE': FLAGS.dhcpbridge_flagfile}
(out, err) = utils.execute(cmd, addl_env=env)
(out, err) = utils.execute(*cmd, addl_env=env)
LOG.debug("ISSUE_IP: %s, %s ", out, err)
@@ -386,11 +527,11 @@ def release_ip(private_ip):
private_ip)
instance_ref = db.fixed_ip_get_instance(context.get_admin_context(),
private_ip)
cmd = "%s del %s %s fake" % (binpath('nova-dhcpbridge'),
instance_ref['mac_address'],
private_ip)
cmd = (binpath('nova-dhcpbridge'), 'del',
instance_ref['mac_address'],
private_ip, 'fake')
env = {'DNSMASQ_INTERFACE': network_ref['bridge'],
'TESTING': '1',
'FLAGFILE': FLAGS.dhcpbridge_flagfile}
(out, err) = utils.execute(cmd, addl_env=env)
(out, err) = utils.execute(*cmd, addl_env=env)
LOG.debug("RELEASE_IP: %s, %s ", out, err)

4
nova/tests/test_scheduler.py
View File

@@ -155,7 +155,7 @@ class SimpleDriverTestCase(test.TestCase):
def _create_instance(self, **kwargs):
"""Create a test instance"""
inst = {}
inst['image_id'] = 'ami-test'
inst['image_id'] = 1
inst['reservation_id'] = 'r-fakeres'
inst['user_id'] = self.user.id
inst['project_id'] = self.project.id
@@ -169,8 +169,6 @@ class SimpleDriverTestCase(test.TestCase):
def _create_volume(self):
"""Create a test volume"""
vol = {}
vol['image_id'] = 'ami-test'
vol['reservation_id'] = 'r-fakeres'
vol['size'] = 1
vol['availability_zone'] = 'test'
return db.volume_create(self.context, vol)['id']

132
nova/tests/test_virt.py
View File

@@ -14,6 +14,10 @@
# License for the specific language governing permissions and limitations
# under the License.
import re
import os
import eventlet
from xml.etree.ElementTree import fromstring as xml_to_tree
from xml.dom.minidom import parseString as xml_to_dom
@@ -30,6 +34,70 @@ FLAGS = flags.FLAGS
flags.DECLARE('instances_path', 'nova.compute.manager')
def _concurrency(wait, done, target):
wait.wait()
done.send()
class CacheConcurrencyTestCase(test.TestCase):
def setUp(self):
super(CacheConcurrencyTestCase, self).setUp()
def fake_exists(fname):
basedir = os.path.join(FLAGS.instances_path, '_base')
if fname == basedir:
return True
return False
def fake_execute(*args, **kwargs):
pass
self.stubs.Set(os.path, 'exists', fake_exists)
self.stubs.Set(utils, 'execute', fake_execute)
def test_same_fname_concurrency(self):
"""Ensures that the same fname cache runs at a sequentially"""
conn = libvirt_conn.LibvirtConnection
wait1 = eventlet.event.Event()
done1 = eventlet.event.Event()
eventlet.spawn(conn._cache_image, _concurrency,
'target', 'fname', False, wait1, done1)
wait2 = eventlet.event.Event()
done2 = eventlet.event.Event()
eventlet.spawn(conn._cache_image, _concurrency,
'target', 'fname', False, wait2, done2)
wait2.send()
eventlet.sleep(0)
try:
self.assertFalse(done2.ready())
self.assertTrue('fname' in conn._image_sems)
finally:
wait1.send()
done1.wait()
eventlet.sleep(0)
self.assertTrue(done2.ready())
self.assertFalse('fname' in conn._image_sems)
def test_different_fname_concurrency(self):
"""Ensures that two different fname caches are concurrent"""
conn = libvirt_conn.LibvirtConnection
wait1 = eventlet.event.Event()
done1 = eventlet.event.Event()
eventlet.spawn(conn._cache_image, _concurrency,
'target', 'fname2', False, wait1, done1)
wait2 = eventlet.event.Event()
done2 = eventlet.event.Event()
eventlet.spawn(conn._cache_image, _concurrency,
'target', 'fname1', False, wait2, done2)
wait2.send()
eventlet.sleep(0)
try:
self.assertTrue(done2.ready())
finally:
wait1.send()
eventlet.sleep(0)
class LibvirtConnTestCase(test.TestCase):
def setUp(self):
super(LibvirtConnTestCase, self).setUp()
@@ -234,16 +302,22 @@ class IptablesFirewallTestCase(test.TestCase):
self.manager.delete_user(self.user)
super(IptablesFirewallTestCase, self).tearDown()
in_rules = [
in_nat_rules = [
'# Generated by iptables-save v1.4.10 on Sat Feb 19 00:03:19 2011',
'*nat',
':PREROUTING ACCEPT [1170:189210]',
':INPUT ACCEPT [844:71028]',
':OUTPUT ACCEPT [5149:405186]',
':POSTROUTING ACCEPT [5063:386098]',
]
in_filter_rules = [
'# Generated by iptables-save v1.4.4 on Mon Dec 6 11:54:13 2010',
'*filter',
':INPUT ACCEPT [969615:281627771]',
':FORWARD ACCEPT [0:0]',
':OUTPUT ACCEPT [915599:63811649]',
':nova-block-ipv4 - [0:0]',
'-A INPUT -i virbr0 -p udp -m udp --dport 53 -j ACCEPT ',
'-A INPUT -i virbr0 -p tcp -m tcp --dport 53 -j ACCEPT ',
'-A INPUT -i virbr0 -p udp -m udp --dport 67 -j ACCEPT ',
'-A INPUT -i virbr0 -p tcp -m tcp --dport 67 -j ACCEPT ',
'-A FORWARD -d 192.168.122.0/24 -o virbr0 -m state --state RELATED'
',ESTABLISHED -j ACCEPT ',
@@ -255,7 +329,7 @@ class IptablesFirewallTestCase(test.TestCase):
'# Completed on Mon Dec 6 11:54:13 2010',
]
in6_rules = [
in6_filter_rules = [
'# Generated by ip6tables-save v1.4.4 on Tue Jan 18 23:47:56 2011',
'*filter',
':INPUT ACCEPT [349155:75810423]',
@@ -315,23 +389,34 @@ class IptablesFirewallTestCase(test.TestCase):
instance_ref = db.instance_get(admin_ctxt, instance_ref['id'])
# self.fw.add_instance(instance_ref)
def fake_iptables_execute(cmd, process_input=None):
if cmd == 'sudo ip6tables-save -t filter':
return '\n'.join(self.in6_rules), None
if cmd == 'sudo iptables-save -t filter':
return '\n'.join(self.in_rules), None
if cmd == 'sudo iptables-restore':
self.out_rules = process_input.split('\n')
def fake_iptables_execute(*cmd, **kwargs):
process_input = kwargs.get('process_input', None)
if cmd == ('sudo', 'ip6tables-save', '-t', 'filter'):
return '\n'.join(self.in6_filter_rules), None
if cmd == ('sudo', 'iptables-save', '-t', 'filter'):
return '\n'.join(self.in_filter_rules), None
if cmd == ('sudo', 'iptables-save', '-t', 'nat'):
return '\n'.join(self.in_nat_rules), None
if cmd == ('sudo', 'iptables-restore'):
lines = process_input.split('\n')
if '*filter' in lines:
self.out_rules = lines
return '', ''
if cmd == 'sudo ip6tables-restore':
self.out6_rules = process_input.split('\n')
if cmd == ('sudo', 'ip6tables-restore'):
lines = process_input.split('\n')
if '*filter' in lines:
self.out6_rules = lines
return '', ''
self.fw.execute = fake_iptables_execute
print cmd, kwargs
from nova.network import linux_net
linux_net.iptables_manager.execute = fake_iptables_execute
self.fw.prepare_instance_filter(instance_ref)
self.fw.apply_instance_filter(instance_ref)
in_rules = filter(lambda l: not l.startswith('#'), self.in_rules)
in_rules = filter(lambda l: not l.startswith('#'),
self.in_filter_rules)
for rule in in_rules:
if not 'nova' in rule:
self.assertTrue(rule in self.out_rules,
@@ -354,17 +439,18 @@ class IptablesFirewallTestCase(test.TestCase):
self.assertTrue(security_group_chain,
"The security group chain wasn't added")
self.assertTrue('-A %s -p icmp -s 192.168.11.0/24 -j ACCEPT' % \
security_group_chain in self.out_rules,
regex = re.compile('-A .* -p icmp -s 192.168.11.0/24 -j ACCEPT')
self.assertTrue(len(filter(regex.match, self.out_rules)) > 0,
"ICMP acceptance rule wasn't added")
self.assertTrue('-A %s -p icmp -s 192.168.11.0/24 -m icmp --icmp-type '
'8 -j ACCEPT' % security_group_chain in self.out_rules,
regex = re.compile('-A .* -p icmp -s 192.168.11.0/24 -m icmp '
'--icmp-type 8 -j ACCEPT')
self.assertTrue(len(filter(regex.match, self.out_rules)) > 0,
"ICMP Echo Request acceptance rule wasn't added")
self.assertTrue('-A %s -p tcp -s 192.168.10.0/24 -m multiport '
'--dports 80:81 -j ACCEPT' % security_group_chain \
in self.out_rules,
regex = re.compile('-A .* -p tcp -s 192.168.10.0/24 -m multiport '
'--dports 80:81 -j ACCEPT')
self.assertTrue(len(filter(regex.match, self.out_rules)) > 0,
"TCP port 80/81 acceptance rule wasn't added")
db.instance_destroy(admin_ctxt, instance_ref['id'])

2
nova/tests/test_volume.py
View File

@@ -99,7 +99,7 @@ class VolumeTestCase(test.TestCase):
def test_run_attach_detach_volume(self):
"""Make sure volume can be attached and detached from instance."""
inst = {}
inst['image_id'] = 'ami-test'
inst['image_id'] = 1
inst['reservation_id'] = 'r-fakeres'
inst['launch_time'] = '10'
inst['user_id'] = 'fake'

148
nova/tests/test_xenapi.py
View File

@@ -18,6 +18,7 @@
Test suite for XenAPI
"""
import functools
import stubout
from nova import db
@@ -41,6 +42,21 @@ from nova.tests.glance import stubs as glance_stubs
FLAGS = flags.FLAGS
def stub_vm_utils_with_vdi_attached_here(function, should_return=True):
"""
vm_utils.with_vdi_attached_here needs to be stubbed out because it
calls down to the filesystem to attach a vdi. This provides a
decorator to handle that.
"""
@functools.wraps(function)
def decorated_function(self, *args, **kwargs):
orig_with_vdi_attached_here = vm_utils.with_vdi_attached_here
vm_utils.with_vdi_attached_here = lambda *x: should_return
function(self, *args, **kwargs)
vm_utils.with_vdi_attached_here = orig_with_vdi_attached_here
return decorated_function
class XenAPIVolumeTestCase(test.TestCase):
"""
Unit tests for Volume operations
@@ -62,6 +78,7 @@ class XenAPIVolumeTestCase(test.TestCase):
'ramdisk_id': 3,
'instance_type': 'm1.large',
'mac_address': 'aa:bb:cc:dd:ee:ff',
'os_type': 'linux'
}
def _create_volume(self, size='0'):
@@ -219,7 +236,7 @@ class XenAPIVMTestCase(test.TestCase):
check()
def check_vm_record(self, conn):
def create_vm_record(self, conn, os_type):
instances = conn.list_instances()
self.assertEquals(instances, [1])
@@ -231,28 +248,63 @@ class XenAPIVMTestCase(test.TestCase):
in xenapi_fake.get_all_records('VM').iteritems()
if not rec['is_control_domain']]
vm = vms[0]
self.vm_info = vm_info
self.vm = vm
def check_vm_record(self, conn):
# Check that m1.large above turned into the right thing.
instance_type = db.instance_type_get_by_name(conn, 'm1.large')
mem_kib = long(instance_type['memory_mb']) << 10
mem_bytes = str(mem_kib << 10)
vcpus = instance_type['vcpus']
self.assertEquals(vm_info['max_mem'], mem_kib)
self.assertEquals(vm_info['mem'], mem_kib)
self.assertEquals(vm['memory_static_max'], mem_bytes)
self.assertEquals(vm['memory_dynamic_max'], mem_bytes)
self.assertEquals(vm['memory_dynamic_min'], mem_bytes)
self.assertEquals(vm['VCPUs_max'], str(vcpus))
self.assertEquals(vm['VCPUs_at_startup'], str(vcpus))
self.assertEquals(self.vm_info['max_mem'], mem_kib)
self.assertEquals(self.vm_info['mem'], mem_kib)
self.assertEquals(self.vm['memory_static_max'], mem_bytes)
self.assertEquals(self.vm['memory_dynamic_max'], mem_bytes)
self.assertEquals(self.vm['memory_dynamic_min'], mem_bytes)
self.assertEquals(self.vm['VCPUs_max'], str(vcpus))
self.assertEquals(self.vm['VCPUs_at_startup'], str(vcpus))
# Check that the VM is running according to Nova
self.assertEquals(vm_info['state'], power_state.RUNNING)
self.assertEquals(self.vm_info['state'], power_state.RUNNING)
# Check that the VM is running according to XenAPI.
self.assertEquals(vm['power_state'], 'Running')
self.assertEquals(self.vm['power_state'], 'Running')
def check_vm_params_for_windows(self):
self.assertEquals(self.vm['platform']['nx'], 'true')
self.assertEquals(self.vm['HVM_boot_params'], {'order': 'dc'})
self.assertEquals(self.vm['HVM_boot_policy'], 'BIOS order')
# check that these are not set
self.assertEquals(self.vm['PV_args'], '')
self.assertEquals(self.vm['PV_bootloader'], '')
self.assertEquals(self.vm['PV_kernel'], '')
self.assertEquals(self.vm['PV_ramdisk'], '')
def check_vm_params_for_linux(self):
self.assertEquals(self.vm['platform']['nx'], 'false')
self.assertEquals(self.vm['PV_args'], 'clocksource=jiffies')
self.assertEquals(self.vm['PV_bootloader'], 'pygrub')
# check that these are not set
self.assertEquals(self.vm['PV_kernel'], '')
self.assertEquals(self.vm['PV_ramdisk'], '')
self.assertEquals(self.vm['HVM_boot_params'], {})
self.assertEquals(self.vm['HVM_boot_policy'], '')
def check_vm_params_for_linux_with_external_kernel(self):
self.assertEquals(self.vm['platform']['nx'], 'false')
self.assertEquals(self.vm['PV_args'], 'root=/dev/xvda1')
self.assertNotEquals(self.vm['PV_kernel'], '')
self.assertNotEquals(self.vm['PV_ramdisk'], '')
# check that these are not set
self.assertEquals(self.vm['HVM_boot_params'], {})
self.assertEquals(self.vm['HVM_boot_policy'], '')
def _test_spawn(self, image_id, kernel_id, ramdisk_id,
instance_type="m1.large"):
instance_type="m1.large", os_type="linux"):
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
values = {'name': 1,
'id': 1,
@@ -263,10 +315,12 @@ class XenAPIVMTestCase(test.TestCase):
'ramdisk_id': ramdisk_id,
'instance_type': instance_type,
'mac_address': 'aa:bb:cc:dd:ee:ff',
'os_type': os_type
}
conn = xenapi_conn.get_connection(False)
instance = db.instance_create(values)
conn.spawn(instance)
self.create_vm_record(conn, os_type)
self.check_vm_record(conn)
def test_spawn_not_enough_memory(self):
@@ -283,24 +337,37 @@ class XenAPIVMTestCase(test.TestCase):
FLAGS.xenapi_image_service = 'objectstore'
self._test_spawn(1, 2, 3)
@stub_vm_utils_with_vdi_attached_here
def test_spawn_raw_glance(self):
FLAGS.xenapi_image_service = 'glance'
self._test_spawn(glance_stubs.FakeGlance.IMAGE_RAW, None, None)
self.check_vm_params_for_linux()
def test_spawn_vhd_glance(self):
def test_spawn_vhd_glance_linux(self):
FLAGS.xenapi_image_service = 'glance'
self._test_spawn(glance_stubs.FakeGlance.IMAGE_VHD, None, None)
self._test_spawn(glance_stubs.FakeGlance.IMAGE_VHD, None, None,
os_type="linux")
self.check_vm_params_for_linux()
def test_spawn_vhd_glance_windows(self):
FLAGS.xenapi_image_service = 'glance'
self._test_spawn(glance_stubs.FakeGlance.IMAGE_VHD, None, None,
os_type="windows")
self.check_vm_params_for_windows()
def test_spawn_glance(self):
FLAGS.xenapi_image_service = 'glance'
self._test_spawn(glance_stubs.FakeGlance.IMAGE_MACHINE,
glance_stubs.FakeGlance.IMAGE_KERNEL,
glance_stubs.FakeGlance.IMAGE_RAMDISK)
self.check_vm_params_for_linux_with_external_kernel()
def tearDown(self):
super(XenAPIVMTestCase, self).tearDown()
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
self.vm_info = None
self.vm = None
self.stubs.UnsetAll()
def _create_instance(self):
@@ -314,7 +381,8 @@ class XenAPIVMTestCase(test.TestCase):
'kernel_id': 2,
'ramdisk_id': 3,
'instance_type': 'm1.large',
'mac_address': 'aa:bb:cc:dd:ee:ff'}
'mac_address': 'aa:bb:cc:dd:ee:ff',
'os_type': 'linux'}
instance = db.instance_create(values)
self.conn.spawn(instance)
return instance
@@ -346,6 +414,57 @@ class XenAPIDiffieHellmanTestCase(test.TestCase):
super(XenAPIDiffieHellmanTestCase, self).tearDown()
class XenAPIMigrateInstance(test.TestCase):
"""
Unit test for verifying migration-related actions
"""
def setUp(self):
super(XenAPIMigrateInstance, self).setUp()
self.stubs = stubout.StubOutForTesting()
FLAGS.target_host = '127.0.0.1'
FLAGS.xenapi_connection_url = 'test_url'
FLAGS.xenapi_connection_password = 'test_pass'
db_fakes.stub_out_db_instance_api(self.stubs)
stubs.stub_out_get_target(self.stubs)
xenapi_fake.reset()
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake',
admin=True)
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.values = {'name': 1, 'id': 1,
'project_id': self.project.id,
'user_id': self.user.id,
'image_id': 1,
'kernel_id': None,
'ramdisk_id': None,
'instance_type': 'm1.large',
'mac_address': 'aa:bb:cc:dd:ee:ff',
'os_type': 'linux'
}
stubs.stub_out_migration_methods(self.stubs)
glance_stubs.stubout_glance_client(self.stubs,
glance_stubs.FakeGlance)
def tearDown(self):
super(XenAPIMigrateInstance, self).tearDown()
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
self.stubs.UnsetAll()
def test_migrate_disk_and_power_off(self):
instance = db.instance_create(self.values)
stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests)
conn = xenapi_conn.get_connection(False)
conn.migrate_disk_and_power_off(instance, '127.0.0.1')
def test_finish_resize(self):
instance = db.instance_create(self.values)
stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests)
conn = xenapi_conn.get_connection(False)
conn.finish_resize(instance, dict(base_copy='hurr', cow='durr'))
class XenAPIDetermineDiskImageTestCase(test.TestCase):
"""
Unit tests for code that detects the ImageType
@@ -360,6 +479,7 @@ class XenAPIDetermineDiskImageTestCase(test.TestCase):
self.fake_instance = FakeInstance()
self.fake_instance.id = 42
self.fake_instance.os_type = 'linux'
def assert_disk_type(self, disk_type):
dt = vm_utils.VMHelper.determine_disk_image_type(

172
nova/tests/test_zones.py Normal file
View File

@@ -0,0 +1,172 @@
# Copyright 2010 United States Government as represented by the
# 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.
"""
Tests For ZoneManager
"""
import datetime
import mox
import novaclient
from nova import context
from nova import db
from nova import flags
from nova import service
from nova import test
from nova import rpc
from nova import utils
from nova.auth import manager as auth_manager
from nova.scheduler import zone_manager
FLAGS = flags.FLAGS
class FakeZone:
"""Represents a fake zone from the db"""
def __init__(self, *args, **kwargs):
for k, v in kwargs.iteritems():
setattr(self, k, v)
def exploding_novaclient(zone):
"""Used when we want to simulate a novaclient call failing."""
raise Exception("kaboom")
class ZoneManagerTestCase(test.TestCase):
"""Test case for zone manager"""
def test_ping(self):
zm = zone_manager.ZoneManager()
self.mox.StubOutWithMock(zm, '_refresh_from_db')
self.mox.StubOutWithMock(zm, '_poll_zones')
zm._refresh_from_db(mox.IgnoreArg())
zm._poll_zones(mox.IgnoreArg())
self.mox.ReplayAll()
zm.ping(None)
self.mox.VerifyAll()
def test_refresh_from_db_new(self):
zm = zone_manager.ZoneManager()
self.mox.StubOutWithMock(db, 'zone_get_all')
db.zone_get_all(mox.IgnoreArg()).AndReturn([
FakeZone(id=1, api_url='http://foo.com', username='user1',
password='pass1'),
])
self.assertEquals(len(zm.zone_states), 0)
self.mox.ReplayAll()
zm._refresh_from_db(None)
self.mox.VerifyAll()
self.assertEquals(len(zm.zone_states), 1)
self.assertEquals(zm.zone_states[1].username, 'user1')
def test_refresh_from_db_replace_existing(self):
zm = zone_manager.ZoneManager()
zone_state = zone_manager.ZoneState()
zone_state.update_credentials(FakeZone(id=1, api_url='http://foo.com',
username='user1', password='pass1'))
zm.zone_states[1] = zone_state
self.mox.StubOutWithMock(db, 'zone_get_all')
db.zone_get_all(mox.IgnoreArg()).AndReturn([
FakeZone(id=1, api_url='http://foo.com', username='user2',
password='pass2'),
])
self.assertEquals(len(zm.zone_states), 1)
self.mox.ReplayAll()
zm._refresh_from_db(None)
self.mox.VerifyAll()
self.assertEquals(len(zm.zone_states), 1)
self.assertEquals(zm.zone_states[1].username, 'user2')
def test_refresh_from_db_missing(self):
zm = zone_manager.ZoneManager()
zone_state = zone_manager.ZoneState()
zone_state.update_credentials(FakeZone(id=1, api_url='http://foo.com',
username='user1', password='pass1'))
zm.zone_states[1] = zone_state
self.mox.StubOutWithMock(db, 'zone_get_all')
db.zone_get_all(mox.IgnoreArg()).AndReturn([])
self.assertEquals(len(zm.zone_states), 1)
self.mox.ReplayAll()
zm._refresh_from_db(None)
self.mox.VerifyAll()
self.assertEquals(len(zm.zone_states), 0)
def test_refresh_from_db_add_and_delete(self):
zm = zone_manager.ZoneManager()
zone_state = zone_manager.ZoneState()
zone_state.update_credentials(FakeZone(id=1, api_url='http://foo.com',
username='user1', password='pass1'))
zm.zone_states[1] = zone_state
self.mox.StubOutWithMock(db, 'zone_get_all')
db.zone_get_all(mox.IgnoreArg()).AndReturn([
FakeZone(id=2, api_url='http://foo.com', username='user2',
password='pass2'),
])
self.assertEquals(len(zm.zone_states), 1)
self.mox.ReplayAll()
zm._refresh_from_db(None)
self.mox.VerifyAll()
self.assertEquals(len(zm.zone_states), 1)
self.assertEquals(zm.zone_states[2].username, 'user2')
def test_poll_zone(self):
self.mox.StubOutWithMock(zone_manager, '_call_novaclient')
zone_manager._call_novaclient(mox.IgnoreArg()).AndReturn(
dict(name='zohan', capabilities='hairdresser'))
zone_state = zone_manager.ZoneState()
zone_state.update_credentials(FakeZone(id=2,
api_url='http://foo.com', username='user2',
password='pass2'))
zone_state.attempt = 1
self.mox.ReplayAll()
zone_manager._poll_zone(zone_state)
self.mox.VerifyAll()
self.assertEquals(zone_state.attempt, 0)
self.assertEquals(zone_state.name, 'zohan')
def test_poll_zone_fails(self):
self.stubs.Set(zone_manager, "_call_novaclient", exploding_novaclient)
zone_state = zone_manager.ZoneState()
zone_state.update_credentials(FakeZone(id=2,
api_url='http://foo.com', username='user2',
password='pass2'))
zone_state.attempt = FLAGS.zone_failures_to_offline - 1
self.mox.ReplayAll()
zone_manager._poll_zone(zone_state)
self.mox.VerifyAll()
self.assertEquals(zone_state.attempt, 3)
self.assertFalse(zone_state.is_active)
self.assertEquals(zone_state.name, None)

58
nova/tests/xenapi/stubs.py
View File

@@ -20,6 +20,7 @@ from nova.virt import xenapi_conn
from nova.virt.xenapi import fake
from nova.virt.xenapi import volume_utils
from nova.virt.xenapi import vm_utils
from nova.virt.xenapi import vmops
def stubout_instance_snapshot(stubs):
@@ -217,3 +218,60 @@ class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests):
def SR_forget(self, _1, ref):
pass
class FakeSessionForMigrationTests(fake.SessionBase):
"""Stubs out a XenAPISession for Migration tests"""
def __init__(self, uri):
super(FakeSessionForMigrationTests, self).__init__(uri)
def VDI_get_by_uuid(*args):
return 'hurr'
def VM_start(self, _1, ref, _2, _3):
vm = fake.get_record('VM', ref)
if vm['power_state'] != 'Halted':
raise fake.Failure(['VM_BAD_POWER_STATE', ref, 'Halted',
vm['power_state']])
vm['power_state'] = 'Running'
vm['is_a_template'] = False
vm['is_control_domain'] = False
def stub_out_migration_methods(stubs):
def fake_get_snapshot(self, instance):
return 'foo', 'bar'
@classmethod
def fake_get_vdi(cls, session, vm_ref):
vdi_ref = fake.create_vdi(name_label='derp', read_only=False,
sr_ref='herp', sharable=False)
vdi_rec = session.get_xenapi().VDI.get_record(vdi_ref)
return vdi_ref, {'uuid': vdi_rec['uuid'], }
def fake_shutdown(self, inst, vm, method='clean'):
pass
@classmethod
def fake_sr(cls, session, *args):
pass
@classmethod
def fake_get_sr_path(cls, *args):
return "fake"
def fake_destroy(*args, **kwargs):
pass
def fake_reset_network(*args, **kwargs):
pass
stubs.Set(vmops.VMOps, '_destroy', fake_destroy)
stubs.Set(vm_utils.VMHelper, 'scan_default_sr', fake_sr)
stubs.Set(vm_utils.VMHelper, 'scan_sr', fake_sr)
stubs.Set(vmops.VMOps, '_get_snapshot', fake_get_snapshot)
stubs.Set(vm_utils.VMHelper, 'get_vdi_for_vm_safely', fake_get_vdi)
stubs.Set(xenapi_conn.XenAPISession, 'wait_for_task', lambda x, y, z: None)
stubs.Set(vm_utils.VMHelper, 'get_sr_path', fake_get_sr_path)
stubs.Set(vmops.VMOps, 'reset_network', fake_reset_network)
stubs.Set(vmops.VMOps, '_shutdown', fake_shutdown)

107
nova/utils.py
View File

@@ -23,10 +23,14 @@ System-level utilities and helper functions.
import base64
import datetime
import functools
import inspect
import json
import lockfile
import netaddr
import os
import random
import re
import socket
import string
import struct
@@ -34,20 +38,20 @@ import sys
import time
import types
from xml.sax import saxutils
import re
import netaddr
from eventlet import event
from eventlet import greenthread
from eventlet.green import subprocess
None
from nova import exception
from nova.exception import ProcessExecutionError
from nova import flags
from nova import log as logging
LOG = logging.getLogger("nova.utils")
TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
FLAGS = flags.FLAGS
def import_class(import_str):
@@ -125,40 +129,59 @@ def fetchfile(url, target):
# c.perform()
# c.close()
# fp.close()
execute("curl --fail %s -o %s" % (url, target))
execute("curl", "--fail", url, "-o", target)
def execute(cmd, process_input=None, addl_env=None, check_exit_code=True):
LOG.debug(_("Running cmd (subprocess): %s"), cmd)
env = os.environ.copy()
if addl_env:
env.update(addl_env)
obj = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
result = None
if process_input != None:
result = obj.communicate(process_input)
else:
result = obj.communicate()
obj.stdin.close()
if obj.returncode:
LOG.debug(_("Result was %s") % obj.returncode)
if check_exit_code and obj.returncode != 0:
(stdout, stderr) = result
raise ProcessExecutionError(exit_code=obj.returncode,
stdout=stdout,
stderr=stderr,
cmd=cmd)
# NOTE(termie): this appears to be necessary to let the subprocess call
# clean something up in between calls, without it two
# execute calls in a row hangs the second one
greenthread.sleep(0)
return result
def execute(*cmd, **kwargs):
process_input = kwargs.get('process_input', None)
addl_env = kwargs.get('addl_env', None)
check_exit_code = kwargs.get('check_exit_code', 0)
stdin = kwargs.get('stdin', subprocess.PIPE)
stdout = kwargs.get('stdout', subprocess.PIPE)
stderr = kwargs.get('stderr', subprocess.PIPE)
attempts = kwargs.get('attempts', 1)
cmd = map(str, cmd)
while attempts > 0:
attempts -= 1
try:
LOG.debug(_("Running cmd (subprocess): %s"), ' '.join(cmd))
env = os.environ.copy()
if addl_env:
env.update(addl_env)
obj = subprocess.Popen(cmd, stdin=stdin,
stdout=stdout, stderr=stderr, env=env)
result = None
if process_input != None:
result = obj.communicate(process_input)
else:
result = obj.communicate()
obj.stdin.close()
if obj.returncode:
LOG.debug(_("Result was %s") % obj.returncode)
if type(check_exit_code) == types.IntType \
and obj.returncode != check_exit_code:
(stdout, stderr) = result
raise ProcessExecutionError(exit_code=obj.returncode,
stdout=stdout,
stderr=stderr,
cmd=' '.join(cmd))
# NOTE(termie): this appears to be necessary to let the subprocess
# call clean something up in between calls, without
# it two execute calls in a row hangs the second one
greenthread.sleep(0)
return result
except ProcessExecutionError:
if not attempts:
raise
else:
LOG.debug(_("%r failed. Retrying."), cmd)
greenthread.sleep(random.randint(20, 200) / 100.0)
def ssh_execute(ssh, cmd, process_input=None,
addl_env=None, check_exit_code=True):
LOG.debug(_("Running cmd (SSH): %s"), cmd)
LOG.debug(_("Running cmd (SSH): %s"), ' '.join(cmd))
if addl_env:
raise exception.Error("Environment not supported over SSH")
@@ -187,7 +210,7 @@ def ssh_execute(ssh, cmd, process_input=None,
raise exception.ProcessExecutionError(exit_code=exit_status,
stdout=stdout,
stderr=stderr,
cmd=cmd)
cmd=' '.join(cmd))
return (stdout, stderr)
@@ -220,9 +243,9 @@ def debug(arg):
return arg
def runthis(prompt, cmd, check_exit_code=True):
LOG.debug(_("Running %s"), (cmd))
rv, err = execute(cmd, check_exit_code=check_exit_code)
def runthis(prompt, *cmd, **kwargs):
LOG.debug(_("Running %s"), (" ".join(cmd)))
rv, err = execute(*cmd, **kwargs)
def generate_uid(topic, size=8):
@@ -254,7 +277,7 @@ def last_octet(address):
def get_my_linklocal(interface):
try:
if_str = execute("ip -f inet6 -o addr show %s" % interface)
if_str = execute("ip", "-f", "inet6", "-o", "addr", "show", interface)
condition = "\s+inet6\s+([0-9a-f:]+)/\d+\s+scope\s+link"
links = [re.search(condition, x) for x in if_str[0].split('\n')]
address = [w.group(1) for w in links if w is not None]
@@ -491,6 +514,18 @@ def loads(s):
return json.loads(s)
def synchronized(name):
def wrap(f):
@functools.wraps(f)
def inner(*args, **kwargs):
lock = lockfile.FileLock(os.path.join(FLAGS.lock_path,
'nova-%s.lock' % name))
with lock:
return f(*args, **kwargs)
return inner
return wrap
def ensure_b64_encoding(val):
"""Safety method to ensure that values expected to be base64-encoded
actually are. If they are, the value is returned unchanged. Otherwise,

134
nova/wsgi.py
View File

@@ -36,6 +36,7 @@ import webob.exc
from paste import deploy
from nova import exception
from nova import flags
from nova import log as logging
from nova import utils
@@ -82,6 +83,35 @@ class Server(object):
log=WritableLogger(logger))
class Request(webob.Request):
def best_match_content_type(self):
"""
Determine the most acceptable content-type based on the
query extension then the Accept header
"""
parts = self.path.rsplit(".", 1)
if len(parts) > 1:
format = parts[1]
if format in ["json", "xml"]:
return "application/{0}".format(parts[1])
ctypes = ["application/json", "application/xml"]
bm = self.accept.best_match(ctypes)
return bm or "application/json"
def get_content_type(self):
try:
ct = self.headers["Content-Type"]
assert ct in ("application/xml", "application/json")
return ct
except Exception:
raise webob.exc.HTTPBadRequest("Invalid content type")
class Application(object):
"""Base WSGI application wrapper. Subclasses need to implement __call__."""
@@ -113,7 +143,7 @@ class Application(object):
def __call__(self, environ, start_response):
r"""Subclasses will probably want to implement __call__ like this:
@webob.dec.wsgify
@webob.dec.wsgify(RequestClass=Request)
def __call__(self, req):
# Any of the following objects work as responses:
@@ -199,7 +229,7 @@ class Middleware(Application):
"""Do whatever you'd like to the response."""
return response
@webob.dec.wsgify
@webob.dec.wsgify(RequestClass=Request)
def __call__(self, req):
response = self.process_request(req)
if response:
@@ -212,7 +242,7 @@ class Debug(Middleware):
"""Helper class that can be inserted into any WSGI application chain
to get information about the request and response."""
@webob.dec.wsgify
@webob.dec.wsgify(RequestClass=Request)
def __call__(self, req):
print ("*" * 40) + " REQUEST ENVIRON"
for key, value in req.environ.items():
@@ -276,7 +306,7 @@ class Router(object):
self._router = routes.middleware.RoutesMiddleware(self._dispatch,
self.map)
@webob.dec.wsgify
@webob.dec.wsgify(RequestClass=Request)
def __call__(self, req):
"""
Route the incoming request to a controller based on self.map.
@@ -285,7 +315,7 @@ class Router(object):
return self._router
@staticmethod
@webob.dec.wsgify
@webob.dec.wsgify(RequestClass=Request)
def _dispatch(req):
"""
Called by self._router after matching the incoming request to a route
@@ -304,11 +334,11 @@ class Controller(object):
WSGI app that reads routing information supplied by RoutesMiddleware
and calls the requested action method upon itself. All action methods
must, in addition to their normal parameters, accept a 'req' argument
which is the incoming webob.Request. They raise a webob.exc exception,
which is the incoming wsgi.Request. They raise a webob.exc exception,
or return a dict which will be serialized by requested content type.
"""
@webob.dec.wsgify
@webob.dec.wsgify(RequestClass=Request)
def __call__(self, req):
"""
Call the method specified in req.environ by RoutesMiddleware.
@@ -318,32 +348,45 @@ class Controller(object):
method = getattr(self, action)
del arg_dict['controller']
del arg_dict['action']
if 'format' in arg_dict:
del arg_dict['format']
arg_dict['req'] = req
result = method(**arg_dict)
if type(result) is dict:
return self._serialize(result, req)
content_type = req.best_match_content_type()
body = self._serialize(result, content_type)
response = webob.Response()
response.headers["Content-Type"] = content_type
response.body = body
return response
else:
return result
def _serialize(self, data, request):
def _serialize(self, data, content_type):
"""
Serialize the given dict to the response type requested in request.
Serialize the given dict to the provided content_type.
Uses self._serialization_metadata if it exists, which is a dict mapping
MIME types to information needed to serialize to that type.
"""
_metadata = getattr(type(self), "_serialization_metadata", {})
serializer = Serializer(request.environ, _metadata)
return serializer.to_content_type(data)
serializer = Serializer(_metadata)
try:
return serializer.serialize(data, content_type)
except exception.InvalidContentType:
raise webob.exc.HTTPNotAcceptable()
def _deserialize(self, data, request):
def _deserialize(self, data, content_type):
"""
Deserialize the request body to the response type requested in request.
Deserialize the request body to the specefied content type.
Uses self._serialization_metadata if it exists, which is a dict mapping
MIME types to information needed to serialize to that type.
"""
_metadata = getattr(type(self), "_serialization_metadata", {})
serializer = Serializer(request.environ, _metadata)
return serializer.deserialize(data)
serializer = Serializer(_metadata)
return serializer.deserialize(data, content_type)
class Serializer(object):
@@ -351,50 +394,53 @@ class Serializer(object):
Serializes and deserializes dictionaries to certain MIME types.
"""
def __init__(self, environ, metadata=None):
def __init__(self, metadata=None):
"""
Create a serializer based on the given WSGI environment.
'metadata' is an optional dict mapping MIME types to information
needed to serialize a dictionary to that type.
"""
self.metadata = metadata or {}
req = webob.Request.blank('', environ)
suffix = req.path_info.split('.')[-1].lower()
if suffix == 'json':
self.handler = self._to_json
elif suffix == 'xml':
self.handler = self._to_xml
elif 'application/json' in req.accept:
self.handler = self._to_json
elif 'application/xml' in req.accept:
self.handler = self._to_xml
else:
# This is the default
self.handler = self._to_json
def to_content_type(self, data):
def _get_serialize_handler(self, content_type):
handlers = {
"application/json": self._to_json,
"application/xml": self._to_xml,
}
try:
return handlers[content_type]
except Exception:
raise exception.InvalidContentType()
def serialize(self, data, content_type):
"""
Serialize a dictionary into a string.
The format of the string will be decided based on the Content Type
requested in self.environ: by Accept: header, or by URL suffix.
Serialize a dictionary into a string of the specified content type.
"""
return self.handler(data)
return self._get_serialize_handler(content_type)(data)
def deserialize(self, datastring):
def deserialize(self, datastring, content_type):
"""
Deserialize a string to a dictionary.
The string must be in the format of a supported MIME type.
"""
datastring = datastring.strip()
return self.get_deserialize_handler(content_type)(datastring)
def get_deserialize_handler(self, content_type):
handlers = {
"application/json": self._from_json,
"application/xml": self._from_xml,
}
try:
is_xml = (datastring[0] == '<')
if not is_xml:
return utils.loads(datastring)
return self._from_xml(datastring)
except:
return None
return handlers[content_type]
except Exception:
raise exception.InvalidContentType(_("Invalid content type %s"
% content_type))
def _from_json(self, datastring):
return utils.loads(datastring)
def _from_xml(self, datastring):
xmldata = self.metadata.get('application/xml', {})