diff --git a/Authors b/Authors index 1a07946b..e8207000 100644 --- a/Authors +++ b/Authors @@ -105,3 +105,4 @@ Yoshiaki Tamura Youcef Laribi Yuriy Taraday Zhixue Wu +Zed Shaw diff --git a/nova/rpc/__init__.py b/nova/rpc/__init__.py new file mode 100644 index 00000000..bdf7f705 --- /dev/null +++ b/nova/rpc/__init__.py @@ -0,0 +1,66 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# 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. + + +from nova.utils import import_object +from nova.rpc.common import RemoteError, LOG +from nova import flags + +FLAGS = flags.FLAGS +flags.DEFINE_string('rpc_backend', + 'nova.rpc.amqp', + "The messaging module to use, defaults to AMQP.") + +RPCIMPL = import_object(FLAGS.rpc_backend) + + +def create_connection(new=True): + return RPCIMPL.Connection.instance(new=True) + + +def create_consumer(conn, topic, proxy, fanout=False): + if fanout: + return RPCIMPL.FanoutAdapterConsumer( + connection=conn, + topic=topic, + proxy=proxy) + else: + return RPCIMPL.TopicAdapterConsumer( + connection=conn, + topic=topic, + proxy=proxy) + + +def create_consumer_set(conn, consumers): + return RPCIMPL.ConsumerSet(connection=conn, consumer_list=consumers) + + +def call(context, topic, msg): + return RPCIMPL.call(context, topic, msg) + + +def cast(context, topic, msg): + return RPCIMPL.cast(context, topic, msg) + + +def fanout_cast(context, topic, msg): + return RPCIMPL.fanout_cast(context, topic, msg) + + +def multicall(context, topic, msg): + return RPCIMPL.multicall(context, topic, msg) diff --git a/nova/rpc.py b/nova/rpc/amqp.py similarity index 96% rename from nova/rpc.py rename to nova/rpc/amqp.py index e2771ca8..61555795 100644 --- a/nova/rpc.py +++ b/nova/rpc/amqp.py @@ -44,9 +44,7 @@ from nova import fakerabbit from nova import flags from nova import log as logging from nova import utils - - -LOG = logging.getLogger('nova.rpc') +from nova.rpc.common import RemoteError, LOG FLAGS = flags.FLAGS @@ -418,25 +416,6 @@ def msg_reply(msg_id, reply=None, failure=None): publisher.close() -class RemoteError(exception.Error): - """Signifies that a remote class has raised an exception. - - Containes a string representation of the type of the original exception, - the value of the original exception, and the traceback. These are - sent to the parent as a joined string so printing the exception - contains all of the relevent info. - - """ - - def __init__(self, exc_type, value, traceback): - self.exc_type = exc_type - self.value = value - self.traceback = traceback - super(RemoteError, self).__init__('%s %s\n%s' % (exc_type, - value, - traceback)) - - def _unpack_context(msg): """Unpack context from msg.""" context_dict = {} diff --git a/nova/rpc/common.py b/nova/rpc/common.py new file mode 100644 index 00000000..1d3065a8 --- /dev/null +++ b/nova/rpc/common.py @@ -0,0 +1,23 @@ +from nova import exception +from nova import log as logging + +LOG = logging.getLogger('nova.rpc') + + +class RemoteError(exception.Error): + """Signifies that a remote class has raised an exception. + + Containes a string representation of the type of the original exception, + the value of the original exception, and the traceback. These are + sent to the parent as a joined string so printing the exception + contains all of the relevent info. + + """ + + def __init__(self, exc_type, value, traceback): + self.exc_type = exc_type + self.value = value + self.traceback = traceback + super(RemoteError, self).__init__('%s %s\n%s' % (exc_type, + value, + traceback)) diff --git a/nova/tests/test_adminapi.py b/nova/tests/test_adminapi.py index fde26e31..06cc498a 100644 --- a/nova/tests/test_adminapi.py +++ b/nova/tests/test_adminapi.py @@ -38,7 +38,7 @@ class AdminApiTestCase(test.TestCase): super(AdminApiTestCase, self).setUp() self.flags(connection_type='fake') - self.conn = rpc.Connection.instance() + self.conn = rpc.create_connection() # set up our cloud self.api = admin.AdminController() diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py index 74d87ad3..8c1a74c7 100644 --- a/nova/tests/test_cloud.py +++ b/nova/tests/test_cloud.py @@ -49,7 +49,7 @@ class CloudTestCase(test.TestCase): self.flags(connection_type='fake', stub_network=True) - self.conn = rpc.Connection.instance() + self.conn = rpc.create_connection() # set up our cloud self.cloud = cloud.CloudController() @@ -322,22 +322,15 @@ class CloudTestCase(test.TestCase): revoke = self.cloud.revoke_security_group_ingress self.assertTrue(revoke(self.context, group_name=sec['name'], **kwargs)) - def test_revoke_security_group_ingress_by_id(self): - kwargs = {'project_id': self.context.project_id, 'name': 'test'} - sec = db.security_group_create(self.context, kwargs) - authz = self.cloud.authorize_security_group_ingress - kwargs = {'to_port': '999', 'from_port': '999', 'ip_protocol': 'tcp'} - authz(self.context, group_id=sec['id'], **kwargs) - revoke = self.cloud.revoke_security_group_ingress - self.assertTrue(revoke(self.context, group_id=sec['id'], **kwargs)) - - def test_authorize_security_group_ingress_by_id(self): + def test_authorize_revoke_security_group_ingress_by_id(self): sec = db.security_group_create(self.context, {'project_id': self.context.project_id, 'name': 'test'}) authz = self.cloud.authorize_security_group_ingress kwargs = {'to_port': '999', 'from_port': '999', 'ip_protocol': 'tcp'} - self.assertTrue(authz(self.context, group_id=sec['id'], **kwargs)) + authz(self.context, group_id=sec['id'], **kwargs) + revoke = self.cloud.revoke_security_group_ingress + self.assertTrue(revoke(self.context, group_id=sec['id'], **kwargs)) def test_authorize_security_group_ingress_missing_protocol_params(self): sec = db.security_group_create(self.context, @@ -957,21 +950,6 @@ class CloudTestCase(test.TestCase): self._wait_for_running(ec2_instance_id) return ec2_instance_id - def test_rescue_unrescue_instance(self): - instance_id = self._run_instance( - image_id='ami-1', - instance_type=FLAGS.default_instance_type, - max_count=1) - self.cloud.rescue_instance(context=self.context, - instance_id=instance_id) - # NOTE(vish): This currently does no validation, it simply makes sure - # that the code path doesn't throw an exception. - self.cloud.unrescue_instance(context=self.context, - instance_id=instance_id) - # TODO(soren): We need this until we can stop polling in the rpc code - # for unit tests. - self.cloud.terminate_instances(self.context, [instance_id]) - def test_console_output(self): instance_id = self._run_instance( image_id='ami-1', diff --git a/nova/tests/test_compute.py b/nova/tests/test_compute.py index ac0fd7d5..860cdedd 100644 --- a/nova/tests/test_compute.py +++ b/nova/tests/test_compute.py @@ -410,7 +410,7 @@ class ComputeTestCase(test.TestCase): def fake(*args, **kwargs): pass - self.stubs.Set(self.compute.driver, 'finish_resize', fake) + self.stubs.Set(self.compute.driver, 'finish_migration', fake) self.stubs.Set(self.compute.network_api, 'get_instance_nw_info', fake) context = self.context.elevated() instance_id = self._create_instance() @@ -521,8 +521,8 @@ class ComputeTestCase(test.TestCase): def fake(*args, **kwargs): pass - self.stubs.Set(self.compute.driver, 'finish_resize', fake) - self.stubs.Set(self.compute.driver, 'revert_resize', fake) + self.stubs.Set(self.compute.driver, 'finish_migration', fake) + self.stubs.Set(self.compute.driver, 'revert_migration', fake) self.stubs.Set(self.compute.network_api, 'get_instance_nw_info', fake) self.compute.run_instance(self.context, instance_id) diff --git a/nova/tests/test_libvirt.py b/nova/tests/test_libvirt.py index 49c591af..c5c3151d 100644 --- a/nova/tests/test_libvirt.py +++ b/nova/tests/test_libvirt.py @@ -217,7 +217,7 @@ class LibvirtConnTestCase(test.TestCase): 'mac_address': 'fake', 'ip_address': 'fake', 'dhcp_server': 'fake', - 'extra_params': 'fake' + 'extra_params': 'fake', } # Creating mocks diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py index ffd748ef..2d243617 100644 --- a/nova/tests/test_rpc.py +++ b/nova/tests/test_rpc.py @@ -33,11 +33,12 @@ LOG = logging.getLogger('nova.tests.rpc') class RpcTestCase(test.TestCase): def setUp(self): super(RpcTestCase, self).setUp() - self.conn = rpc.Connection.instance(True) + self.conn = rpc.create_connection(True) self.receiver = TestReceiver() - self.consumer = rpc.TopicAdapterConsumer(connection=self.conn, - topic='test', - proxy=self.receiver) + self.consumer = rpc.create_consumer(self.conn, + 'test', + self.receiver, + False) self.consumer.attach_to_eventlet() self.context = context.get_admin_context() @@ -129,6 +130,8 @@ class RpcTestCase(test.TestCase): """Calls echo in the passed queue""" LOG.debug(_("Nested received %(queue)s, %(value)s") % locals()) + # TODO: so, it will replay the context and use the same REQID? + # that's bizarre. ret = rpc.call(context, queue, {"method": "echo", @@ -137,10 +140,11 @@ class RpcTestCase(test.TestCase): return value nested = Nested() - conn = rpc.Connection.instance(True) - consumer = rpc.TopicAdapterConsumer(connection=conn, - topic='nested', - proxy=nested) + conn = rpc.create_connection(True) + consumer = rpc.create_consumer(conn, + 'nested', + nested, + False) consumer.attach_to_eventlet() value = 42 result = rpc.call(self.context, @@ -149,47 +153,6 @@ class RpcTestCase(test.TestCase): "value": value}}) self.assertEqual(value, result) - def test_connectionpool_single(self): - """Test that ConnectionPool recycles a single connection.""" - conn1 = rpc.ConnectionPool.get() - rpc.ConnectionPool.put(conn1) - conn2 = rpc.ConnectionPool.get() - rpc.ConnectionPool.put(conn2) - self.assertEqual(conn1, conn2) - - def test_connectionpool_double(self): - """Test that ConnectionPool returns and reuses separate connections. - - When called consecutively we should get separate connections and upon - returning them those connections should be reused for future calls - before generating a new connection. - - """ - conn1 = rpc.ConnectionPool.get() - conn2 = rpc.ConnectionPool.get() - - self.assertNotEqual(conn1, conn2) - rpc.ConnectionPool.put(conn1) - rpc.ConnectionPool.put(conn2) - - conn3 = rpc.ConnectionPool.get() - conn4 = rpc.ConnectionPool.get() - self.assertEqual(conn1, conn3) - self.assertEqual(conn2, conn4) - - def test_connectionpool_limit(self): - """Test connection pool limit and connection uniqueness.""" - max_size = FLAGS.rpc_conn_pool_size - conns = [] - - for i in xrange(max_size): - conns.append(rpc.ConnectionPool.get()) - - self.assertFalse(rpc.ConnectionPool.free_items) - self.assertEqual(rpc.ConnectionPool.current_size, - rpc.ConnectionPool.max_size) - self.assertEqual(len(set(conns)), max_size) - class TestReceiver(object): """Simple Proxy class so the consumer has methods to call. diff --git a/nova/tests/test_rpc_amqp.py b/nova/tests/test_rpc_amqp.py new file mode 100644 index 00000000..d29f7ae3 --- /dev/null +++ b/nova/tests/test_rpc_amqp.py @@ -0,0 +1,68 @@ +from nova import context +from nova import flags +from nova import log as logging +from nova import rpc +from nova.rpc import amqp +from nova import test + + +FLAGS = flags.FLAGS +LOG = logging.getLogger('nova.tests.rpc') + + +class RpcAMQPTestCase(test.TestCase): + def setUp(self): + super(RpcAMQPTestCase, self).setUp() + self.conn = rpc.create_connection(True) + self.receiver = TestReceiver() + self.consumer = rpc.create_consumer(self.conn, + 'test', + self.receiver, + False) + self.consumer.attach_to_eventlet() + self.context = context.get_admin_context() + + def test_connectionpool_single(self): + """Test that ConnectionPool recycles a single connection.""" + conn1 = amqp.ConnectionPool.get() + amqp.ConnectionPool.put(conn1) + conn2 = amqp.ConnectionPool.get() + amqp.ConnectionPool.put(conn2) + self.assertEqual(conn1, conn2) + + +class TestReceiver(object): + """Simple Proxy class so the consumer has methods to call. + + Uses static methods because we aren't actually storing any state. + + """ + + @staticmethod + def echo(context, value): + """Simply returns whatever value is sent in.""" + LOG.debug(_("Received %s"), value) + return value + + @staticmethod + def context(context, value): + """Returns dictionary version of context.""" + LOG.debug(_("Received %s"), context) + return context.to_dict() + + @staticmethod + def echo_three_times(context, value): + context.reply(value) + context.reply(value + 1) + context.reply(value + 2) + + @staticmethod + def echo_three_times_yield(context, value): + yield value + yield value + 1 + yield value + 2 + + @staticmethod + def fail(context, value): + """Raises an exception with the value sent in.""" + raise Exception(value) diff --git a/nova/tests/test_xenapi.py b/nova/tests/test_xenapi.py index 4c51b1b2..ab51f7d1 100644 --- a/nova/tests/test_xenapi.py +++ b/nova/tests/test_xenapi.py @@ -786,8 +786,15 @@ class XenAPIMigrateInstance(test.TestCase): conn = xenapi_conn.get_connection(False) conn.migrate_disk_and_power_off(instance, '127.0.0.1') - def test_finish_resize(self): + def test_finish_migrate(self): instance = db.instance_create(self.context, self.values) + self.called = False + + def fake_vdi_resize(*args, **kwargs): + self.called = True + + self.stubs.Set(stubs.FakeSessionForMigrationTests, + "VDI_resize_online", fake_vdi_resize) stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests) stubs.stubout_loopingcall_start(self.stubs) conn = xenapi_conn.get_connection(False) @@ -805,8 +812,70 @@ class XenAPIMigrateInstance(test.TestCase): 'label': 'fake', 'mac': 'DE:AD:BE:EF:00:00', 'rxtx_cap': 3})] - conn.finish_resize(instance, dict(base_copy='hurr', cow='durr'), - network_info) + conn.finish_migration(instance, dict(base_copy='hurr', cow='durr'), + network_info, resize_instance=True) + self.assertEqual(self.called, True) + + def test_finish_migrate_no_local_storage(self): + tiny_type_id = \ + instance_types.get_instance_type_by_name('m1.tiny')['id'] + self.values.update({'instance_type_id': tiny_type_id, 'local_gb': 0}) + instance = db.instance_create(self.context, self.values) + + def fake_vdi_resize(*args, **kwargs): + raise Exception("This shouldn't be called") + + self.stubs.Set(stubs.FakeSessionForMigrationTests, + "VDI_resize_online", fake_vdi_resize) + stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests) + stubs.stubout_loopingcall_start(self.stubs) + conn = xenapi_conn.get_connection(False) + network_info = [({'bridge': 'fa0', 'id': 0, 'injected': False}, + {'broadcast': '192.168.0.255', + 'dns': ['192.168.0.1'], + 'gateway': '192.168.0.1', + 'gateway6': 'dead:beef::1', + 'ip6s': [{'enabled': '1', + 'ip': 'dead:beef::dcad:beff:feef:0', + 'netmask': '64'}], + 'ips': [{'enabled': '1', + 'ip': '192.168.0.100', + 'netmask': '255.255.255.0'}], + 'label': 'fake', + 'mac': 'DE:AD:BE:EF:00:00', + 'rxtx_cap': 3})] + conn.finish_migration(instance, dict(base_copy='hurr', cow='durr'), + network_info, resize_instance=True) + + def test_finish_migrate_no_resize_vdi(self): + instance = db.instance_create(self.context, self.values) + + def fake_vdi_resize(*args, **kwargs): + raise Exception("This shouldn't be called") + + self.stubs.Set(stubs.FakeSessionForMigrationTests, + "VDI_resize_online", fake_vdi_resize) + stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests) + stubs.stubout_loopingcall_start(self.stubs) + conn = xenapi_conn.get_connection(False) + network_info = [({'bridge': 'fa0', 'id': 0, 'injected': False}, + {'broadcast': '192.168.0.255', + 'dns': ['192.168.0.1'], + 'gateway': '192.168.0.1', + 'gateway6': 'dead:beef::1', + 'ip6s': [{'enabled': '1', + 'ip': 'dead:beef::dcad:beff:feef:0', + 'netmask': '64'}], + 'ips': [{'enabled': '1', + 'ip': '192.168.0.100', + 'netmask': '255.255.255.0'}], + 'label': 'fake', + 'mac': 'DE:AD:BE:EF:00:00', + 'rxtx_cap': 3})] + + # Resize instance would be determined by the compute call + conn.finish_migration(instance, dict(base_copy='hurr', cow='durr'), + network_info, resize_instance=False) class XenAPIDetermineDiskImageTestCase(test.TestCase):