diff --git a/etc/nova/nova.conf.sample b/etc/nova/nova.conf.sample
index b2607a9c2f9f..b9ea77d0c3f3 100644
--- a/etc/nova/nova.conf.sample
+++ b/etc/nova/nova.conf.sample
@@ -918,6 +918,11 @@
 ###### (StrOpt) path to s3 buckets
 # buckets_path="$state_path/buckets"
 
+######### defined in nova.rpc.common #########
+
+###### (ListOpt) Modules of exceptions that are permitted to be recreated
+# allowed_rpc_exception_modules="nova.exception"
+
 ######### defined in nova.rpc.impl_kombu #########
 
 ###### (StrOpt) SSL certification authority file (valid only if SSL enabled)
diff --git a/nova/api/ec2/cloud.py b/nova/api/ec2/cloud.py
index 2ff5b32c61c3..9346d107bdc4 100644
--- a/nova/api/ec2/cloud.py
+++ b/nova/api/ec2/cloud.py
@@ -41,7 +41,6 @@ from nova import flags
 from nova.image import s3
 from nova import log as logging
 from nova import network
-from nova.rpc import common as rpc_common
 from nova import utils
 from nova import volume
 
@@ -1253,15 +1252,8 @@ class CloudController(object):
 
     def allocate_address(self, context, **kwargs):
         LOG.audit(_("Allocate address"), context=context)
-        try:
-            public_ip = self.network_api.allocate_floating_ip(context)
-            return {'publicIp': public_ip}
-        except rpc_common.RemoteError as ex:
-            # NOTE(tr3buchet) - why does this block exist?
-            if ex.exc_type == 'NoMoreFloatingIps':
-                raise exception.NoMoreFloatingIps()
-            else:
-                raise
+        public_ip = self.network_api.allocate_floating_ip(context)
+        return {'publicIp': public_ip}
 
     def release_address(self, context, public_ip, **kwargs):
         LOG.audit(_("Release address %s"), public_ip, context=context)
diff --git a/nova/api/openstack/compute/contrib/floating_ips.py b/nova/api/openstack/compute/contrib/floating_ips.py
index 6b9e9e97c506..4a5cec8d2235 100644
--- a/nova/api/openstack/compute/contrib/floating_ips.py
+++ b/nova/api/openstack/compute/contrib/floating_ips.py
@@ -152,16 +152,12 @@ class FloatingIPController(object):
         try:
             address = self.network_api.allocate_floating_ip(context, pool)
             ip = self.network_api.get_floating_ip_by_address(context, address)
-        except rpc_common.RemoteError as ex:
-            # NOTE(tr3buchet) - why does this block exist?
-            if ex.exc_type == 'NoMoreFloatingIps':
-                if pool:
-                    msg = _("No more floating ips in pool %s.") % pool
-                else:
-                    msg = _("No more floating ips available.")
-                raise webob.exc.HTTPBadRequest(explanation=msg)
+        except exception.NoMoreFloatingIps:
+            if pool:
+                msg = _("No more floating ips in pool %s.") % pool
             else:
-                raise
+                msg = _("No more floating ips available.")
+            raise webob.exc.HTTPBadRequest(explanation=msg)
 
         return _translate_floating_ip_view(ip)
 
@@ -212,9 +208,6 @@ class FloatingIPActionController(wsgi.Controller):
         except exception.FixedIpNotFoundForInstance:
             msg = _("No fixed ips associated to instance")
             raise webob.exc.HTTPBadRequest(explanation=msg)
-        except rpc_common.RemoteError:
-            msg = _("Associate floating ip failed")
-            raise webob.exc.HTTPInternalServerError(explanation=msg)
 
         return webob.Response(status_int=202)
 
diff --git a/nova/network/api.py b/nova/network/api.py
index 805fdbc84a8a..fa9567427de7 100644
--- a/nova/network/api.py
+++ b/nova/network/api.py
@@ -213,26 +213,10 @@ class API(base.Base):
                 'rxtx_factor': instance['instance_type']['rxtx_factor'],
                 'host': instance['host'],
                 'project_id': instance['project_id']}
-        try:
-            nw_info = rpc.call(context, FLAGS.network_topic,
-                               {'method': 'get_instance_nw_info',
-                                'args': args})
-            return network_model.NetworkInfo.hydrate(nw_info)
-        # FIXME(comstud) rpc calls raise RemoteError if the remote raises
-        # an exception.  In the case here, because of a race condition,
-        # it's possible the remote will raise a InstanceNotFound when
-        # someone deletes the instance while this call is in progress.
-        #
-        # Unfortunately, we don't have access to the original exception
-        # class now.. but we do have the exception class's name.  So,
-        # we're checking it here and raising a new exception.
-        #
-        # Ultimately we need RPC to be able to serialize more things like
-        # classes.
-        except rpc_common.RemoteError as err:
-            if err.exc_type == 'InstanceNotFound':
-                raise exception.InstanceNotFound(instance_id=instance['id'])
-            raise
+        nw_info = rpc.call(context, FLAGS.network_topic,
+                           {'method': 'get_instance_nw_info',
+                            'args': args})
+        return network_model.NetworkInfo.hydrate(nw_info)
 
     def validate_networks(self, context, requested_networks):
         """validate the networks passed at the time of creating
diff --git a/nova/rpc/amqp.py b/nova/rpc/amqp.py
index 444ade480f19..df6e1b974a32 100644
--- a/nova/rpc/amqp.py
+++ b/nova/rpc/amqp.py
@@ -27,7 +27,6 @@ AMQP, but is deprecated and predates this code.
 
 import inspect
 import sys
-import traceback
 import uuid
 
 from eventlet import greenpool
@@ -141,11 +140,7 @@ def msg_reply(msg_id, connection_pool, reply=None, failure=None, ending=False):
     """
     with ConnectionContext(connection_pool) as conn:
         if failure:
-            message = str(failure[1])
-            tb = traceback.format_exception(*failure)
-            LOG.error(_("Returning exception %s to caller"), message)
-            LOG.error(tb)
-            failure = (failure[0].__name__, str(failure[1]), tb)
+            failure = rpc_common.serialize_remote_exception(failure)
 
         try:
             msg = {'result': reply, 'failure': failure}
@@ -285,7 +280,9 @@ class MulticallWaiter(object):
     def __call__(self, data):
         """The consume() callback will call this.  Store the result."""
         if data['failure']:
-            self._result = rpc_common.RemoteError(*data['failure'])
+            failure = data['failure']
+            self._result = rpc_common.deserialize_remote_exception(failure)
+
         elif data.get('ending', False):
             self._got_ending = True
         else:
diff --git a/nova/rpc/common.py b/nova/rpc/common.py
index 95c2458101e6..51bf2fd26850 100644
--- a/nova/rpc/common.py
+++ b/nova/rpc/common.py
@@ -18,11 +18,14 @@
 #    under the License.
 
 import copy
+import sys
+import traceback
 
 from nova import exception
 from nova import flags
 from nova import log as logging
 from nova.openstack.common import cfg
+from nova import utils
 
 
 LOG = logging.getLogger(__name__)
@@ -37,9 +40,14 @@ rpc_opts = [
     cfg.IntOpt('rpc_response_timeout',
                default=60,
                help='Seconds to wait for a response from call or multicall'),
+    cfg.IntOpt('allowed_rpc_exception_modules',
+               default=['nova.exception'],
+               help='Modules of exceptions that are permitted to be recreated'
+                    'upon receiving exception data from an rpc call.'),
     ]
 
 flags.FLAGS.register_opts(rpc_opts)
+FLAGS = flags.FLAGS
 
 
 class RemoteError(exception.NovaException):
@@ -158,3 +166,74 @@ def _safe_log(log_func, msg, msg_data):
         msg_data['auth_token'] = '<SANITIZED>'
 
     return log_func(msg, msg_data)
+
+
+def serialize_remote_exception(failure_info):
+    """Prepares exception data to be sent over rpc.
+
+    Failure_info should be a sys.exc_info() tuple.
+
+    """
+    tb = traceback.format_exception(*failure_info)
+    failure = failure_info[1]
+    LOG.error(_("Returning exception %s to caller"), unicode(failure))
+    LOG.error(tb)
+
+    kwargs = {}
+    if hasattr(failure, 'kwargs'):
+        kwargs = failure.kwargs
+
+    data = {
+        'class': str(failure.__class__.__name__),
+        'module': str(failure.__class__.__module__),
+        'message': unicode(failure),
+        'tb': tb,
+        'args': failure.args,
+        'kwargs': kwargs
+    }
+
+    json_data = utils.dumps(data)
+
+    return json_data
+
+
+def deserialize_remote_exception(data):
+    failure = utils.loads(str(data))
+
+    trace = failure.get('tb', [])
+    message = failure.get('message', "") + "\n" + "\n".join(trace)
+    name = failure.get('class')
+    module = failure.get('module')
+
+    # NOTE(ameade): We DO NOT want to allow just any module to be imported, in
+    # order to prevent arbitrary code execution.
+    if not module in FLAGS.allowed_rpc_exception_modules:
+        return RemoteError(name, failure.get('message'), trace)
+
+    try:
+        __import__(module)
+        mod = sys.modules[module]
+        klass = getattr(mod, name)
+        if not issubclass(klass, Exception):
+            raise TypeError("Can only deserialize Exceptions")
+
+        failure = klass(**failure.get('kwargs', {}))
+    except (AttributeError, TypeError, ImportError):
+        return RemoteError(name, failure.get('message'), trace)
+
+    ex_type = type(failure)
+    str_override = lambda self: message
+    new_ex_type = type(ex_type.__name__ + "_Remote", (ex_type,),
+                       {'__str__': str_override})
+    try:
+        # NOTE(ameade): Dynamically create a new exception type and swap it in
+        # as the new type for the exception. This only works on user defined
+        # Exceptions and not core python exceptions. This is important because
+        # we cannot necessarily change an exception message so we must override
+        # the __str__ method.
+        failure.__class__ = new_ex_type
+    except TypeError as e:
+        # NOTE(ameade): If a core exception then just add the traceback to the
+        # first exception argument.
+        failure.args = (message,) + failure.args[1:]
+    return failure
diff --git a/nova/rpc/impl_fake.py b/nova/rpc/impl_fake.py
index 42ed7907d98c..43aed15c2643 100644
--- a/nova/rpc/impl_fake.py
+++ b/nova/rpc/impl_fake.py
@@ -77,12 +77,8 @@ class Consumer(object):
                     else:
                         res.append(rval)
                 done.send(res)
-            except Exception:
-                exc_info = sys.exc_info()
-                done.send_exception(
-                        rpc_common.RemoteError(exc_info[0].__name__,
-                            str(exc_info[1]),
-                            ''.join(traceback.format_exception(*exc_info))))
+            except Exception as e:
+                done.send_exception(e)
 
         thread = eventlet.greenthread.spawn(_inner)
 
@@ -161,7 +157,7 @@ def call(context, topic, msg, timeout=None):
 def cast(context, topic, msg):
     try:
         call(context, topic, msg)
-    except rpc_common.RemoteError:
+    except Exception:
         pass
 
 
@@ -184,5 +180,5 @@ def fanout_cast(context, topic, msg):
     for consumer in CONSUMERS.get(topic, []):
         try:
             consumer.call(context, method, args, None)
-        except rpc_common.RemoteError:
+        except Exception:
             pass
diff --git a/nova/scheduler/driver.py b/nova/scheduler/driver.py
index 91c5aa367c06..ad83bc10accd 100644
--- a/nova/scheduler/driver.py
+++ b/nova/scheduler/driver.py
@@ -362,7 +362,7 @@ class Scheduler(object):
                      {"method": 'compare_cpu',
                       "args": {'cpu_info': oservice_ref['cpu_info']}})
 
-        except rpc_common.RemoteError:
+        except exception.InvalidCPUInfo:
             src = instance_ref['host']
             LOG.exception(_("host %(dest)s is not compatible with "
                                 "original host %(src)s.") % locals())
@@ -446,17 +446,12 @@ class Scheduler(object):
         available = available_gb * (1024 ** 3)
 
         # Getting necessary disk size
-        try:
-            topic = db.queue_get_for(context, FLAGS.compute_topic,
-                                              instance_ref['host'])
-            ret = rpc.call(context, topic,
-                           {"method": 'get_instance_disk_info',
-                            "args": {'instance_name': instance_ref['name']}})
-            disk_infos = utils.loads(ret)
-        except rpc_common.RemoteError:
-            LOG.exception(_("host %(dest)s is not compatible with "
-                                "original host %(src)s.") % locals())
-            raise
+        topic = db.queue_get_for(context, FLAGS.compute_topic,
+                                          instance_ref['host'])
+        ret = rpc.call(context, topic,
+                       {"method": 'get_instance_disk_info',
+                        "args": {'instance_name': instance_ref['name']}})
+        disk_infos = utils.loads(ret)
 
         necessary = 0
         if disk_over_commit:
diff --git a/nova/tests/api/openstack/compute/contrib/test_floating_ips.py b/nova/tests/api/openstack/compute/contrib/test_floating_ips.py
index dd0165077b3d..452e0eef2d0a 100644
--- a/nova/tests/api/openstack/compute/contrib/test_floating_ips.py
+++ b/nova/tests/api/openstack/compute/contrib/test_floating_ips.py
@@ -20,6 +20,7 @@ import webob
 from nova.api.openstack.compute.contrib import floating_ips
 from nova import context
 from nova import db
+from nova import exception
 from nova import network
 from nova import compute
 from nova import rpc
@@ -184,6 +185,18 @@ class FloatingIpTest(test.TestCase):
         self.assertEqual(res_dict['floating_ip']['ip'], '10.10.10.10')
         self.assertEqual(res_dict['floating_ip']['instance_id'], None)
 
+    def test_floating_ip_show_not_found(self):
+        def fake_get_floating_ip(*args, **kwargs):
+            raise exception.FloatingIpNotFound()
+
+        self.stubs.Set(network.api.API, "get_floating_ip",
+                       fake_get_floating_ip)
+
+        req = fakes.HTTPRequest.blank('/v2/fake/os-floating-ips/9876')
+
+        self.assertRaises(webob.exc.HTTPNotFound,
+                          self.controller.show, req, 9876)
+
     def test_show_associated_floating_ip(self):
         def get_floating_ip(self, context, id):
             return {'id': 1, 'address': '10.10.10.10', 'pool': 'nova',
@@ -205,7 +218,7 @@ class FloatingIpTest(test.TestCase):
 # test floating ip allocate/release(deallocate)
     def test_floating_ip_allocate_no_free_ips(self):
         def fake_call(*args, **kwargs):
-            raise(rpc_common.RemoteError('NoMoreFloatingIps', '', ''))
+            raise exception.NoMoreFloatingIps()
 
         self.stubs.Set(rpc, "call", fake_call)
 
diff --git a/nova/tests/rpc/common.py b/nova/tests/rpc/common.py
index 87cb522c61d9..3524e5682800 100644
--- a/nova/tests/rpc/common.py
+++ b/nova/tests/rpc/common.py
@@ -25,6 +25,7 @@ from eventlet import greenthread
 import nose
 
 from nova import context
+from nova import exception
 from nova import log as logging
 from nova.rpc import amqp as rpc_amqp
 from nova.rpc import common as rpc_common
@@ -100,30 +101,6 @@ class BaseRpcTestCase(test.TestCase):
                                    "args": {"value": value}})
         self.assertEqual(self.context.to_dict(), result)
 
-    def test_call_exception(self):
-        """Test that exception gets passed back properly.
-
-        rpc.call returns a RemoteError object.  The value of the
-        exception is converted to a string, so we convert it back
-        to an int in the test.
-
-        """
-        value = 42
-        self.assertRaises(rpc_common.RemoteError,
-                          self.rpc.call,
-                          self.context,
-                          'test',
-                          {"method": "fail",
-                           "args": {"value": value}})
-        try:
-            self.rpc.call(self.context,
-                     'test',
-                     {"method": "fail",
-                      "args": {"value": value}})
-            self.fail("should have thrown RemoteError")
-        except rpc_common.RemoteError as exc:
-            self.assertEqual(int(exc.value), value)
-
     def test_nested_calls(self):
         """Test that we can do an rpc.call inside another call."""
         class Nested(object):
@@ -248,7 +225,12 @@ class TestReceiver(object):
     @staticmethod
     def fail(context, value):
         """Raises an exception with the value sent in."""
-        raise Exception(value)
+        raise NotImplementedError(value)
+
+    @staticmethod
+    def fail_converted(context, value):
+        """Raises an exception with the value sent in."""
+        raise exception.ConvertedException(explanation=value)
 
     @staticmethod
     def block(context, value):
diff --git a/nova/tests/rpc/test_common.py b/nova/tests/rpc/test_common.py
new file mode 100644
index 000000000000..6220bd01a134
--- /dev/null
+++ b/nova/tests/rpc/test_common.py
@@ -0,0 +1,147 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2012 OpenStack, LLC
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+"""
+Unit Tests for 'common' functons used through rpc code.
+"""
+
+import json
+import sys
+
+from nova import context
+from nova import exception
+from nova import flags
+from nova import log as logging
+from nova import test
+from nova.rpc import amqp as rpc_amqp
+from nova.rpc import common as rpc_common
+from nova.tests.rpc import common
+
+FLAGS = flags.FLAGS
+LOG = logging.getLogger(__name__)
+
+
+def raise_exception():
+    raise Exception("test")
+
+
+class FakeUserDefinedException(Exception):
+    def __init__(self):
+        Exception.__init__(self, "Test Message")
+
+
+class RpcCommonTestCase(test.TestCase):
+    def test_serialize_remote_exception(self):
+        expected = {
+            'class': 'Exception',
+            'module': 'exceptions',
+            'message': 'test',
+        }
+
+        try:
+            raise_exception()
+        except Exception as exc:
+            failure = rpc_common.serialize_remote_exception(sys.exc_info())
+
+        failure = json.loads(failure)
+        #assure the traceback was added
+        self.assertEqual(expected['class'], failure['class'])
+        self.assertEqual(expected['module'], failure['module'])
+        self.assertEqual(expected['message'], failure['message'])
+
+    def test_serialize_remote_nova_exception(self):
+        def raise_nova_exception():
+            raise exception.NovaException("test", code=500)
+
+        expected = {
+            'class': 'NovaException',
+            'module': 'nova.exception',
+            'kwargs': {'code': 500},
+            'message': 'test'
+        }
+
+        try:
+            raise_nova_exception()
+        except Exception as exc:
+            failure = rpc_common.serialize_remote_exception(sys.exc_info())
+
+        failure = json.loads(failure)
+        #assure the traceback was added
+        self.assertEqual(expected['class'], failure['class'])
+        self.assertEqual(expected['module'], failure['module'])
+        self.assertEqual(expected['kwargs'], failure['kwargs'])
+        self.assertEqual(expected['message'], failure['message'])
+
+    def test_deserialize_remote_exception(self):
+        failure = {
+            'class': 'NovaException',
+            'module': 'nova.exception',
+            'message': 'test message',
+            'tb': ['raise NovaException'],
+        }
+        serialized = json.dumps(failure)
+
+        after_exc = rpc_common.deserialize_remote_exception(serialized)
+        self.assertTrue(isinstance(after_exc, exception.NovaException))
+        self.assertTrue('test message' in unicode(after_exc))
+        #assure the traceback was added
+        self.assertTrue('raise NovaException' in unicode(after_exc))
+
+    def test_deserialize_remote_exception_bad_module(self):
+        failure = {
+            'class': 'popen2',
+            'module': 'os',
+            'kwargs': {'cmd': '/bin/echo failed'},
+            'message': 'foo',
+        }
+        serialized = json.dumps(failure)
+
+        after_exc = rpc_common.deserialize_remote_exception(serialized)
+        self.assertTrue(isinstance(after_exc, rpc_common.RemoteError))
+
+    def test_deserialize_remote_exception_user_defined_exception(self):
+        """Ensure a user defined exception can be deserialized."""
+        self.flags(allowed_rpc_exception_modules=[self.__class__.__module__])
+        failure = {
+            'class': 'FakeUserDefinedException',
+            'module': self.__class__.__module__,
+            'tb': ['raise FakeUserDefinedException'],
+        }
+        serialized = json.dumps(failure)
+
+        after_exc = rpc_common.deserialize_remote_exception(serialized)
+        self.assertTrue(isinstance(after_exc, FakeUserDefinedException))
+        #assure the traceback was added
+        self.assertTrue('raise FakeUserDefinedException' in unicode(after_exc))
+
+    def test_deserialize_remote_exception_cannot_recreate(self):
+        """Ensure a RemoteError is returned on initialization failure.
+
+        If an exception cannot be recreated with it's original class then a
+        RemoteError with the exception informations should still be returned.
+
+        """
+        self.flags(allowed_rpc_exception_modules=[self.__class__.__module__])
+        failure = {
+            'class': 'FakeIDontExistException',
+            'module': self.__class__.__module__,
+            'tb': ['raise FakeIDontExistException'],
+        }
+        serialized = json.dumps(failure)
+
+        after_exc = rpc_common.deserialize_remote_exception(serialized)
+        self.assertTrue(isinstance(after_exc, rpc_common.RemoteError))
+        #assure the traceback was added
+        self.assertTrue('raise FakeIDontExistException' in unicode(after_exc))
diff --git a/nova/tests/rpc/test_kombu.py b/nova/tests/rpc/test_kombu.py
index aa49b5d51003..966cb3a6905b 100644
--- a/nova/tests/rpc/test_kombu.py
+++ b/nova/tests/rpc/test_kombu.py
@@ -20,6 +20,7 @@ Unit Tests for remote procedure calls using kombu
 """
 
 from nova import context
+from nova import exception
 from nova import flags
 from nova import log as logging
 from nova import test
@@ -292,4 +293,54 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
 
         self.assertEqual(self.received_message, message)
         # Only called once, because our stub goes away during reconnection
-        self.assertEqual(info['called'], 1)
+
+    def test_call_exception(self):
+        """Test that exception gets passed back properly.
+
+        rpc.call returns an Exception object.  The value of the
+        exception is converted to a string.
+
+        """
+        self.flags(allowed_rpc_exception_modules=['exceptions'])
+        value = "This is the exception message"
+        self.assertRaises(NotImplementedError,
+                          self.rpc.call,
+                          self.context,
+                          'test',
+                          {"method": "fail",
+                           "args": {"value": value}})
+        try:
+            self.rpc.call(self.context,
+                     'test',
+                     {"method": "fail",
+                      "args": {"value": value}})
+            self.fail("should have thrown Exception")
+        except NotImplementedError as exc:
+            self.assertTrue(value in unicode(exc))
+            #Traceback should be included in exception message
+            self.assertTrue('raise NotImplementedError(value)' in unicode(exc))
+
+    def test_call_converted_exception(self):
+        """Test that exception gets passed back properly.
+
+        rpc.call returns an Exception object.  The value of the
+        exception is converted to a string.
+
+        """
+        value = "This is the exception message"
+        self.assertRaises(exception.ConvertedException,
+                          self.rpc.call,
+                          self.context,
+                          'test',
+                          {"method": "fail_converted",
+                           "args": {"value": value}})
+        try:
+            self.rpc.call(self.context,
+                     'test',
+                     {"method": "fail_converted",
+                      "args": {"value": value}})
+            self.fail("should have thrown Exception")
+        except exception.ConvertedException as exc:
+            self.assertTrue(value in unicode(exc))
+            #Traceback should be included in exception message
+            self.assertTrue('exception.ConvertedException' in unicode(exc))