Merge lp:~zedshaw/nova/generic-msg-queue-layer into lp:~hudson-openstack/nova/trunk

[Zed A. Shaw]

This change creates a minimalist API abstraction for the nova/rpc.py code so that it's possible to use other queue mechanisms besides Rabbit and/or AMQP, and even use other drivers for AMQP rather than Rabbit. The change is intended to give the least amount of interference with the rest of the code, fixes several bugs in the tests, and works with the current branch. I also have a small demo driver+server for using 0MQ which I'll submit after this patch is merged.

[Brian Waldon (bcwaldon)]

Looks like you may have requested the wrong Vish to review this. You probably want https://launchpad.net/~vishvananda

[Lorin Hochstein (lorinh)]

It looks like there are some artifacts in the source files left over from resolving merge conflicts:

nova/tests/test_adminapi.py

266: +<<<<<<< TREE
274: +=======
281: +>>>>>>> MERGE-SOURCE

review: Needs Fixing

[Vish Ishaya (vishvananda)]

Looks pretty good. Something weird happened in merge of test_cloud. I think you accidentally deleted a test or two. Any possible replacement for the deleted tests in test_service, or were they just too dependent on amqp to be useful?

review: Needs Fixing

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 12:36:34AM -0000, Lorin Hochstein wrote:
> Review: Needs Fixing
> It looks like there are some artifacts in the source files left over from resolving merge conflicts:

No, that's not in my source, it's from the bzr merge resolution. Since
it takes people days to review things there ends up being conflicts by
the time it's ready. I'll try it again.

--
Zed A. Shaw
http://zedshaw.com/

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 01:58:50AM -0000, Vish Ishaya wrote:
> Review: Needs Fixing
> Looks pretty good. Something weird happened in merge of test_cloud. I think you accidentally deleted a test or two. Any possible replacement for the deleted tests in test_service, or were they just too dependent on amqp to be useful?

We found the tests actually didn't test anything in service, they just
tested some things in amqp, but then used mock objects which meant they
didn't test anything.

--
Zed A. Shaw
http://zedshaw.com/

[Lorin Hochstein (lorinh)]

> On Fri, Jul 29, 2011 at 12:36:34AM -0000, Lorin Hochstein wrote:
> > Review: Needs Fixing
> > It looks like there are some artifacts in the source files left over from
> resolving merge conflicts:
>
> No, that's not in my source, it's from the bzr merge resolution. Since
> it takes people days to review things there ends up being conflicts by
> the time it's ready. I'll try it again.
>

Whoops, sorry about that.

review: Abstain

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 03:16:28AM -0000, Lorin Hochstein wrote:
> Whoops, sorry about that.

No problem, all the tools do a bad job of showing what's really going
on. I'm doing the merge and retest now, should have it committed soon.

--
Zed A. Shaw
http://zedshaw.com/

[Vish Ishaya (vishvananda)]

nice zed!

Looking forward to the alternative implementation. Just one minor question. Why use load_module instead of using the existing import_object? It looks like it does the same thing (with the exception that import_object will also let you use a class instead of a module for your backend)

review: Needs Information

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 06:43:24AM -0000, Vish Ishaya wrote:
> Review: Needs Information nice zed!
>
> Looking forward to the alternative implementation. Just one minor
> question. Why use load_module instead of using the existing
> import_object? It looks like it does the same thing (with the
> exception that import_object will also let you use a class instead of
> a module for your backend)

I needed just a module loader, not an object-in-a-module loader. The
way the nova.rpc code is used only at the module level in most of the
code (rpc.call, rpc.cast, etc.) meant I could refactor the code to use
an OOP style without changing tons of code everywhere. It could be
changed to be generic (since python sort of doesn't care), and use an
object or module, but right now that'd be too much change for one patch.

--
Zed A. Shaw
http://zedshaw.com/

[Vish Ishaya (vishvananda)]

> It could be
> changed to be generic (since python sort of doesn't care), and use an
> object or module, but right now that'd be too much change for one patch.

that is exactly what utils.import_object does. It attempts to import it as a module or object, if that fails, it attempts to import it as a callable and call it.

network/manager.py uses a module:
97:flags.DEFINE_string('network_driver', 'nova.network.linux_net'
314: self.driver = utils.import_object(network_driver)

compute/manager.py uses a method:
65:flags.DEFINE_string('compute_driver', 'nova.virt.connection.get_connection'
134: utils.import_object(compute_driver),

volume/manager.py uses a class:
59:flags.DEFINE_string('volume_driver', 'nova.volume.driver.ISCSIDriver',
71: self.driver = utils.import_object(volume_driver)

Gotta love python! I think you can just replace your load_module with import_object with no issues.

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 07:07:25AM -0000, Vish Ishaya wrote:
> > It could be
> > changed to be generic (since python sort of doesn't care), and use an
> > object or module, but right now that'd be too much change for one patch.
>
> that is exactly what utils.import_object does. It attempts to import it as a module or object, if that fails, it attempts to import it as a callable and call it.

So then it's not really import_object, it's more "import stuff
magically"? :-)

--
Zed A. Shaw
http://zedshaw.com/

[Vish Ishaya (vishvananda)]

> So then it's not really import_object, it's more "import stuff
> magically"? :-)

I never claimed it had a good name. :p

[Jason Kölker (jason-koelker)]

Any reason not to rename rpc_backends to rpc and move rpc.py into __init__.py or rpc/api.py and then import it in __init__.py to keep the layout the same as the rest of the codebase?

Other than that, is bueno. Tests pass and I was able to boot an instance. I'm glad you squared this away, its been annoying me that it wasn't pluggable.

review: Needs Information

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 03:27:51PM -0000, Jason Kölker wrote:
> Review: Needs Information
> Any reason not to rename rpc_backends to rpc and move rpc.py into __init__.py or rpc/api.py and then import it in __init__.py to keep the layout the same as the rest of the codebase?

I believe I tried that but the imports didn't work right without the
abstration code being in an explicit nova/rpc.py module.

--
Zed A. Shaw
http://zedshaw.com/

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 03:01:51PM -0000, Vish Ishaya wrote:
>
> > So then it's not really import_object, it's more "import stuff
> > magically"? :-)
>
> I never claimed it had a good name. :p

Ok, this is fixed, it now just uses the nova.util.import_object.

FYI, the tests are broken if you have to make a new virtualenv.

--
Zed A. Shaw
http://zedshaw.com/

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 03:27:51PM -0000, Jason Kölker wrote:
> Review: Needs Information
> Any reason not to rename rpc_backends to rpc and move rpc.py into __init__.py or rpc/api.py and then import it in __init__.py to keep the layout the same as the rest of the codebase?

I just fixed the latest thing Vish brought up, and then looking at how
I'd have to change the code to fit this request I'm thinking it makes it
more complicated.

I'm going to ask that, if the reviewers want this request, that they
vote on it so that I'm not chasing my tail implementing something that's
not really necessary.

After this patch is in, it'll be easier to refactor.

--
Zed A. Shaw
http://zedshaw.com/

[Trey Morris (tr3buchet)]

"...keep the layout the same as the rest of the codebase?"
+1

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 06:07:26PM -0000, Trey Morris wrote:
> "...keep the layout the same as the rest of the codebase?"
> +1

If that's the case, point me to another part of the codebase I should
model this after. Specific files that do this same design of "loadable
modules are inside a directory with __init__.py as the abstraction."

--
Zed A. Shaw
http://zedshaw.com/

[Lorin Hochstein (lorinh)]

> On Fri, Jul 29, 2011 at 03:01:51PM -0000, Vish Ishaya wrote:
> FYI, the tests are broken if you have to make a new virtualenv.
>

Logged this as https://bugs.launchpad.net/nova/+bug/818210 (assuming it's the same problem you're encountering).

[Jason Kölker (jason-koelker)]

> On Fri, Jul 29, 2011 at 06:07:26PM -0000, Trey Morris wrote:
> > "...keep the layout the same as the rest of the codebase?"
> > +1
>
> If that's the case, point me to another part of the codebase I should
> model this after. Specific files that do this same design of "loadable
> modules are inside a directory with __init__.py as the abstraction."

The ipv6 module is the simplest, but the db uses the same pattern as well. Its a trivial change:

$ bzr diff
=== renamed directory 'nova/rpc_backends' => 'nova/rpc'
=== modified file 'nova/rpc/__init__.py'
--- nova/rpc_backends/__init__.py 2011-07-26 23:29:50 +0000
+++ nova/rpc/__init__.py 2011-07-29 18:58:38 +0000
@@ -0,0 +1,2 @@
+from nova.rpc.api import *
+from nova.rpc.common import RemoteError, LOG

=== modified file 'nova/rpc/amqp.py'
--- nova/rpc_backends/amqp.py 2011-07-26 23:29:50 +0000
+++ nova/rpc/amqp.py 2011-07-29 18:57:42 +0000
@@ -44,7 +44,7 @@
 from nova import flags
 from nova import log as logging
 from nova import utils
-from nova.rpc_backends.common import RemoteError, LOG
+from nova.rpc.common import RemoteError, LOG

 FLAGS = flags.FLAGS

=== renamed file 'nova/rpc.py' => 'nova/rpc/api.py'
--- nova/rpc.py 2011-07-29 17:33:58 +0000
+++ nova/rpc/api.py 2011-07-29 18:58:11 +0000
@@ -18,12 +18,11 @@

 from nova.utils import import_object
-from nova.rpc_backends.common import RemoteError, LOG
 from nova import flags

 FLAGS = flags.FLAGS
 flags.DEFINE_string('rpc_backend',
- 'nova.rpc_backends.amqp',
+ 'nova.rpc.amqp',
                     "The messaging module to use, defaults to AMQP.")

 RPCIMPL = import_object(FLAGS.rpc_backend)

=== modified file 'nova/tests/test_rpc_amqp.py'
--- nova/tests/test_rpc_amqp.py 2011-07-26 23:29:50 +0000
+++ nova/tests/test_rpc_amqp.py 2011-07-29 19:02:57 +0000
@@ -2,7 +2,7 @@
 from nova import flags
 from nova import log as logging
 from nova import rpc
-from nova.rpc_backends import amqp
+from nova.rpc import amqp
 from nova import test

[Jason Kölker (jason-koelker)]

> > On Fri, Jul 29, 2011 at 06:07:26PM -0000, Trey Morris wrote:
> > > "...keep the layout the same as the rest of the codebase?"
> > > +1
> >
> > If that's the case, point me to another part of the codebase I should
> > model this after. Specific files that do this same design of "loadable
> > modules are inside a directory with __init__.py as the abstraction."
>
> The ipv6 module is the simplest, but the db uses the same pattern as well. Its
> a trivial change:
>

Ah, I see you already did that, excellent!

review: Approve

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 06:16:02PM -0000, Zed A. Shaw wrote:
> On Fri, Jul 29, 2011 at 06:07:26PM -0000, Trey Morris wrote:
> > "...keep the layout the same as the rest of the codebase?"
> > +1

Alright, it's now matching the "rest of the codebase", based on the
description since I can't find other modules doing this with a similar
layout.

It's also remerged, and new request to review put in.

--
Zed A. Shaw
http://zedshaw.com/

[Trey Morris (tr3buchet)]

koelker: haha way to pay attention!

zed: i see you just added yourself to Authors. welcome to the party!

review: Approve

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 07:22:35PM -0000, Jason Kölker wrote:
> > The ipv6 module is the simplest, but the db uses the same pattern as well. Its
> > a trivial change:
> >
>
> Ah, I see you already did that, excellent!

Yep, did it already. FYI, it's not trivial to move these modules around
because people are trying to edit them *and* bzr tends to have problems
with moved files. If there's much more moving required I'll ask that it
be done on subsequent changes to the code rather than wait around for
perfection on this patch.

--
Zed A. Shaw
http://zedshaw.com/

[Zed A. Shaw (zedshaw)]

On Fri, Jul 29, 2011 at 07:25:58PM -0000, Trey Morris wrote:
> Review: Approve
> koelker: haha way to pay attention!
>
> zed: i see you just added yourself to Authors. welcome to the party!

Thanks, although it was mostly to satisfy a unit test.

--
Zed A. Shaw
http://zedshaw.com/

=== modified file 'Authors'
--- Authors	2011-07-26 00:49:36 +0000
+++ Authors	2011-07-29 19:17:40 +0000
@@ -105,3 +105,4 @@
 Youcef Laribi <[e-mail]>
 Yuriy Taraday <[e-mail]>
 Zhixue Wu <[e-mail]>
+Zed Shaw <[e-mail]>
 
=== added directory 'nova/rpc'
=== added file 'nova/rpc/__init__.py'
--- nova/rpc/__init__.py	1970-01-01 00:00:00 +0000
+++ nova/rpc/__init__.py	2011-07-29 19:17:40 +0000
@@ -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)
 
=== renamed file 'nova/rpc.py' => 'nova/rpc/amqp.py'
--- nova/rpc.py	2011-07-01 14:53:20 +0000
+++ nova/rpc/amqp.py	2011-07-29 19:17:40 +0000
@@ -44,9 +44,7 @@
 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 @@
         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 = {}
 
=== added file 'nova/rpc/common.py'
--- nova/rpc/common.py	1970-01-01 00:00:00 +0000
+++ nova/rpc/common.py	2011-07-29 19:17:40 +0000
@@ -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))
 
=== modified file 'nova/service.py'
--- nova/service.py	2011-06-24 01:31:00 +0000
+++ nova/service.py	2011-07-29 19:17:40 +0000
@@ -149,26 +149,22 @@
         if 'nova-compute' == self.binary:
             self.manager.update_available_resource(ctxt)
 
-        self.conn = rpc.Connection.instance(new=True)
+        self.conn = rpc.create_connection(new=True)
         logging.debug("Creating Consumer connection for Service %s" %
                       self.topic)
 
         # Share this same connection for these Consumers
-        consumer_all = rpc.TopicAdapterConsumer(
-                connection=self.conn,
-                topic=self.topic,
-                proxy=self)
-        consumer_node = rpc.TopicAdapterConsumer(
-                connection=self.conn,
-                topic='%s.%s' % (self.topic, self.host),
-                proxy=self)
-        fanout = rpc.FanoutAdapterConsumer(
-                connection=self.conn,
-                topic=self.topic,
-                proxy=self)
-        consumer_set = rpc.ConsumerSet(
-                connection=self.conn,
-                consumer_list=[consumer_all, consumer_node, fanout])
+        consumer_all = rpc.create_consumer(self.conn, self.topic, self,
+                                           fanout=False)
+
+        node_topic = '%s.%s' % (self.topic, self.host)
+        consumer_node = rpc.create_consumer(self.conn, node_topic, self,
+                                            fanout=False)
+
+        fanout = rpc.create_consumer(self.conn, self.topic, self, fanout=True)
+
+        consumers = [consumer_all, consumer_node, fanout]
+        consumer_set = rpc.create_consumer_set(self.conn, consumers)
 
         # Wait forever, processing these consumers
         def _wait():
 
=== modified file 'nova/test.py'
--- nova/test.py	2011-07-08 09:35:01 +0000
+++ nova/test.py	2011-07-29 19:17:40 +0000
@@ -99,9 +99,7 @@
         self.flag_overrides = {}
         self.injected = []
         self._services = []
-        self._monkey_patch_attach()
         self._original_flags = FLAGS.FlagValuesDict()
-        rpc.ConnectionPool = rpc.Pool(max_size=FLAGS.rpc_conn_pool_size)
 
     def tearDown(self):
         """Runs after each test method to tear down test environment."""
@@ -126,9 +124,6 @@
             # Reset any overriden flags
             self.reset_flags()
 
-            # Reset our monkey-patches
-            rpc.Consumer.attach_to_eventlet = self.original_attach
-
             # Stop any timers
             for x in self.injected:
                 try:
@@ -172,17 +167,6 @@
         self._services.append(svc)
         return svc
 
-    def _monkey_patch_attach(self):
-        self.original_attach = rpc.Consumer.attach_to_eventlet
-
-        def _wrapped(inner_self):
-            rv = self.original_attach(inner_self)
-            self.injected.append(rv)
-            return rv
-
-        _wrapped.func_name = self.original_attach.func_name
-        rpc.Consumer.attach_to_eventlet = _wrapped
-
     # Useful assertions
     def assertDictMatch(self, d1, d2, approx_equal=False, tolerance=0.001):
         """Assert two dicts are equivalent.
 
=== modified file 'nova/tests/test_adminapi.py'
--- nova/tests/test_adminapi.py	2011-06-27 21:48:03 +0000
+++ nova/tests/test_adminapi.py	2011-07-29 19:17:40 +0000
@@ -39,7 +39,7 @@
         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()
 
=== modified file 'nova/tests/test_cloud.py'
--- nova/tests/test_cloud.py	2011-07-27 21:39:27 +0000
+++ nova/tests/test_cloud.py	2011-07-29 19:17:40 +0000
@@ -50,7 +50,7 @@
         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()
@@ -326,23 +326,16 @@
         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)
+    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'}
         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):
-        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))
-
     def test_authorize_security_group_ingress_missing_protocol_params(self):
         sec = db.security_group_create(self.context,
                                        {'project_id': self.context.project_id,
@@ -961,21 +954,6 @@
         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',
 
=== modified file 'nova/tests/test_rpc.py'
--- nova/tests/test_rpc.py	2011-05-26 22:08:53 +0000
+++ nova/tests/test_rpc.py	2011-07-29 19:17:40 +0000
@@ -33,11 +33,12 @@
 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 @@
                 """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 @@
                 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 @@
                                               "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.
 
=== added file 'nova/tests/test_rpc_amqp.py'
--- nova/tests/test_rpc_amqp.py	1970-01-01 00:00:00 +0000
+++ nova/tests/test_rpc_amqp.py	2011-07-29 19:17:40 +0000
@@ -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)
 
=== modified file 'nova/tests/test_service.py'
--- nova/tests/test_service.py	2011-06-28 14:39:04 +0000
+++ nova/tests/test_service.py	2011-07-29 19:17:40 +0000
@@ -109,103 +109,8 @@
         #             the looping calls are created in StartService.
         app = service.Service.create(host=host, binary=binary, topic=topic)
 
-        self.mox.StubOutWithMock(service.rpc.Connection, 'instance')
-        service.rpc.Connection.instance(new=mox.IgnoreArg())
-
-        self.mox.StubOutWithMock(rpc,
-                                 'TopicAdapterConsumer',
-                                 use_mock_anything=True)
-        self.mox.StubOutWithMock(rpc,
-                                 'FanoutAdapterConsumer',
-                                 use_mock_anything=True)
-
-        self.mox.StubOutWithMock(rpc,
-                                 'ConsumerSet',
-                                 use_mock_anything=True)
-
-        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic=topic,
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.TopicAdapterConsumer)
-
-        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic='%s.%s' % (topic, host),
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.TopicAdapterConsumer)
-
-        rpc.FanoutAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic=topic,
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.FanoutAdapterConsumer)
-
-        def wait_func(self, limit=None):
-            return None
-
-        mock_cset = self.mox.CreateMock(rpc.ConsumerSet,
-                {'wait': wait_func})
-        rpc.ConsumerSet(connection=mox.IgnoreArg(),
-                        consumer_list=mox.IsA(list)).AndReturn(mock_cset)
-        wait_func(mox.IgnoreArg())
-
-        service_create = {'host': host,
-                          'binary': binary,
-                          'topic': topic,
-                          'report_count': 0,
-                          'availability_zone': 'nova'}
-        service_ref = {'host': host,
-                       'binary': binary,
-                       'report_count': 0,
-                       'id': 1}
-
-        service.db.service_get_by_args(mox.IgnoreArg(),
-                                       host,
-                                       binary).AndRaise(exception.NotFound())
-        service.db.service_create(mox.IgnoreArg(),
-                                  service_create).AndReturn(service_ref)
-        self.mox.ReplayAll()
-
-        app.start()
-        app.stop()
         self.assert_(app)
 
-    # We're testing sort of weird behavior in how report_state decides
-    # whether it is disconnected, it looks for a variable on itself called
-    # 'model_disconnected' and report_state doesn't really do much so this
-    # these are mostly just for coverage
-    def test_report_state_no_service(self):
-        host = 'foo'
-        binary = 'bar'
-        topic = 'test'
-        service_create = {'host': host,
-                          'binary': binary,
-                          'topic': topic,
-                          'report_count': 0,
-                          'availability_zone': 'nova'}
-        service_ref = {'host': host,
-                          'binary': binary,
-                          'topic': topic,
-                          'report_count': 0,
-                          'availability_zone': 'nova',
-                          'id': 1}
-
-        service.db.service_get_by_args(mox.IgnoreArg(),
-                                      host,
-                                      binary).AndRaise(exception.NotFound())
-        service.db.service_create(mox.IgnoreArg(),
-                                  service_create).AndReturn(service_ref)
-        service.db.service_get(mox.IgnoreArg(),
-                               service_ref['id']).AndReturn(service_ref)
-        service.db.service_update(mox.IgnoreArg(), service_ref['id'],
-                                  mox.ContainsKeyValue('report_count', 1))
-
-        self.mox.ReplayAll()
-        serv = service.Service(host,
-                               binary,
-                               topic,
-                               'nova.tests.test_service.FakeManager')
-        serv.start()
-        serv.report_state()
-
     def test_report_state_newly_disconnected(self):
         host = 'foo'
         binary = 'bar'
@@ -276,81 +181,6 @@
 
         self.assert_(not serv.model_disconnected)
 
-    def test_compute_can_update_available_resource(self):
-        """Confirm compute updates their record of compute-service table."""
-        host = 'foo'
-        binary = 'nova-compute'
-        topic = 'compute'
-
-        # Any mocks are not working without UnsetStubs() here.
-        self.mox.UnsetStubs()
-        ctxt = context.get_admin_context()
-        service_ref = db.service_create(ctxt, {'host': host,
-                                               'binary': binary,
-                                               'topic': topic})
-        serv = service.Service(host,
-                               binary,
-                               topic,
-                               'nova.compute.manager.ComputeManager')
-
-        # This testcase want to test calling update_available_resource.
-        # No need to call periodic call, then below variable must be set 0.
-        serv.report_interval = 0
-        serv.periodic_interval = 0
-
-        # Creating mocks
-        self.mox.StubOutWithMock(service.rpc.Connection, 'instance')
-        service.rpc.Connection.instance(new=mox.IgnoreArg())
-
-        self.mox.StubOutWithMock(rpc,
-                                 'TopicAdapterConsumer',
-                                 use_mock_anything=True)
-        self.mox.StubOutWithMock(rpc,
-                                 'FanoutAdapterConsumer',
-                                 use_mock_anything=True)
-
-        self.mox.StubOutWithMock(rpc,
-                                 'ConsumerSet',
-                                 use_mock_anything=True)
-
-        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic=topic,
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.TopicAdapterConsumer)
-
-        rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic='%s.%s' % (topic, host),
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.TopicAdapterConsumer)
-
-        rpc.FanoutAdapterConsumer(connection=mox.IgnoreArg(),
-                            topic=topic,
-                            proxy=mox.IsA(service.Service)).AndReturn(
-                                    rpc.FanoutAdapterConsumer)
-
-        def wait_func(self, limit=None):
-            return None
-
-        mock_cset = self.mox.CreateMock(rpc.ConsumerSet,
-                {'wait': wait_func})
-        rpc.ConsumerSet(connection=mox.IgnoreArg(),
-                        consumer_list=mox.IsA(list)).AndReturn(mock_cset)
-        wait_func(mox.IgnoreArg())
-
-        self.mox.StubOutWithMock(serv.manager.driver,
-                                 'update_available_resource')
-        serv.manager.driver.update_available_resource(mox.IgnoreArg(), host)
-
-        # Just doing start()-stop(), not confirm new db record is created,
-        # because update_available_resource() works only in
-        # libvirt environment. This testcase confirms
-        # update_available_resource() is called. Otherwise, mox complains.
-        self.mox.ReplayAll()
-        serv.start()
-        serv.stop()
-
-        db.service_destroy(ctxt, service_ref['id'])
-
 
 class TestWSGIService(test.TestCase):
 
 
=== modified file 'nova/tests/test_test.py'
--- nova/tests/test_test.py	2011-02-24 01:58:32 +0000
+++ nova/tests/test_test.py	2011-07-29 19:17:40 +0000
@@ -33,8 +33,13 @@
         self.start_service('compute')
 
     def test_rpc_consumer_isolation(self):
-        connection = rpc.Connection.instance(new=True)
-        consumer = rpc.TopicAdapterConsumer(connection, topic='compute')
-        consumer.register_callback(
-                lambda x, y: self.fail('I should never be called'))
+        class NeverCalled(object):
+
+            def __getattribute__(*args):
+                assert False, "I should never get called."
+
+        connection = rpc.create_connection(new=True)
+        proxy = NeverCalled()
+        consumer = rpc.create_consumer(connection, 'compute',
+                                       proxy, fanout=False)
         consumer.attach_to_eventlet()