From 02a2e6d380db01098a57e2e4d1269d28caba47dd Mon Sep 17 00:00:00 2001
From: Davanum Srinivas <davanum@gmail.com>
Date: Sun, 16 Aug 2015 17:30:46 -0400
Subject: [PATCH] Log warning instead of raising RuntimeError
Changes introduced in:
I0fc1717e3118bc1cd7b9cd0ccc072251cfb2c038
are causing Neutron to fail as summarized by Doug in:
http://markmail.org/message/2xlclp7gqnqpkted
So we should log warnings instead and make sure we
find which projects are affected (using logstash) and
fix them. Once we do that, then we can switch this
back to raise RuntimeError(s)
Change-Id: I9dce272246b89f3ec63aefcf11d385fd2d21da6e
---
oslo_messaging/server.py | 21 ++++++++++++++++-----
oslo_messaging/tests/rpc/test_server.py | 25 ++++++++++++++++++++-----
2 files changed, 36 insertions(+), 10 deletions(-)
diff --git a/oslo_messaging/server.py b/oslo_messaging/server.py
index c5b4093d4..fac0815ea 100644
--- a/oslo_messaging/server.py
+++ b/oslo_messaging/server.py
@@ -23,14 +23,19 @@ __all__ = [
'ServerListenError',
]
+import logging
+
from oslo_service import service
from stevedore import driver
from oslo_messaging._drivers import base as driver_base
from oslo_messaging._i18n import _
+from oslo_messaging._i18n import _LW
from oslo_messaging import _utils
from oslo_messaging import exceptions
+LOG = logging.getLogger(__name__)
+
class MessagingServerError(exceptions.MessagingException):
"""Base class for all MessageHandlingServer exceptions."""
@@ -135,8 +140,11 @@ class MessageHandlingServer(service.ServiceBase):
if self._thread_id is None:
self._thread_id = self._get_thread_id()
elif self._thread_id != self._get_thread_id():
- raise RuntimeError(_("start/stop/wait must be called in the "
- "same thread"))
+ # NOTE(dims): Need to change this to raise RuntimeError after
+ # verifying/fixing other openstack projects (like Neutron)
+ # work ok with this change
+ LOG.warn(_LW("start/stop/wait must be called in the "
+ "same thread"))
def stop(self):
"""Stop handling incoming messages.
@@ -165,9 +173,12 @@ class MessageHandlingServer(service.ServiceBase):
self._check_same_thread_id()
if self._running:
- raise RuntimeError(_("wait() should be called after stop() as it "
- "waits for existing messages to finish "
- "processing"))
+ # NOTE(dims): Need to change this to raise RuntimeError after
+ # verifying/fixing other openstack projects (like Neutron)
+ # work ok with this change
+ LOG.warn(_("wait() should be called after stop() as it "
+ "waits for existing messages to finish "
+ "processing"))
if self._executor is not None:
self._executor.wait()
diff --git a/oslo_messaging/tests/rpc/test_server.py b/oslo_messaging/tests/rpc/test_server.py
index 2601f4f8a..dfa900cdf 100644
--- a/oslo_messaging/tests/rpc/test_server.py
+++ b/oslo_messaging/tests/rpc/test_server.py
@@ -18,6 +18,7 @@ import threading
from oslo_config import cfg
import testscenarios
+import mock
import oslo_messaging
from oslo_messaging.tests import utils as test_utils
from six.moves import mock
@@ -130,7 +131,9 @@ class TestRPCServer(test_utils.BaseTestCase, ServerSetupMixin):
self.assertIsNone(server._executor)
self.assertEqual(1, listener.cleanup.call_count)
- def test_server_invalid_wait_running_server(self):
+ @mock.patch('oslo_messaging._executors.impl_pooledexecutor.'
+ 'PooledExecutor.wait')
+ def test_server_invalid_wait_running_server(self, mock_wait):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
target = oslo_messaging.Target(topic='foo', server='bar')
endpoints = [object()]
@@ -142,9 +145,15 @@ class TestRPCServer(test_utils.BaseTestCase, ServerSetupMixin):
self.addCleanup(server.wait)
self.addCleanup(server.stop)
server.start()
- self.assertRaises(RuntimeError, server.wait)
+ with mock.patch('logging.Logger.warn') as warn:
+ server.wait()
+ warn.assert_called_with('wait() should be called after '
+ 'stop() as it waits for existing '
+ 'messages to finish processing')
- def test_server_invalid_stop_from_other_thread(self):
+ @mock.patch('oslo_messaging._executors.impl_pooledexecutor.'
+ 'PooledExecutor.wait')
+ def test_server_invalid_stop_from_other_thread(self, mock_wait):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
target = oslo_messaging.Target(topic='foo', server='bar')
endpoints = [object()]
@@ -158,8 +167,14 @@ class TestRPCServer(test_utils.BaseTestCase, ServerSetupMixin):
t.start()
self.addCleanup(t.join)
self.addCleanup(t.stop)
- self.assertRaises(RuntimeError, server.stop)
- self.assertRaises(RuntimeError, server.wait)
+ with mock.patch('logging.Logger.warn') as warn:
+ server.stop()
+ warn.assert_called_with('start/stop/wait must be called '
+ 'in the same thread')
+ with mock.patch('logging.Logger.warn') as warn:
+ server.wait()
+ warn.assert_called_with('start/stop/wait must be called '
+ 'in the same thread')
def test_no_target_server(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')