Don't allow qpid receiving thread to die

This patch is a partial backport of
22ec8ff616a799085239e3e529daeeefea6366c4 in oslo-incubator.

https://review.openstack.org/#/c/32235/13

This patch ensures that the thread created by consume_in_thread() can
not be killed off by an unexpected exception.

Related-Bug: #1189711

Change-Id: I4370045b450b2b4b9b3bde1f6f3654cdecc722e2

neutron/openstack/common/excutils.py

@@ -22,6 +22,7 @@ Exception related utilities.
 import contextlib
 import logging
 import sys
+import time
 import traceback
 
 from neutron.openstack.common.gettextutils import _
@@ -49,3 +50,34 @@ def save_and_reraise_exception():
                       traceback.format_exception(type_, value, tb))
         raise
     raise type_, value, tb
+
+
+def forever_retry_uncaught_exceptions(infunc):
+    def inner_func(*args, **kwargs):
+        last_log_time = 0
+        last_exc_message = None
+        exc_count = 0
+        while True:
+            try:
+                return infunc(*args, **kwargs)
+            except Exception as exc:
+                this_exc_message = str(exc)
+                if this_exc_message == last_exc_message:
+                    exc_count += 1
+                else:
+                    exc_count = 1
+                # Do not log any more frequently than once a minute unless
+                # the exception message changes
+                cur_time = int(time.time())
+                if (cur_time - last_log_time > 60 or
+                        this_exc_message != last_exc_message):
+                    logging.exception(
+                        _('Unexpected exception occurred %d time(s)... '
+                          'retrying.') % exc_count)
+                    last_log_time = cur_time
+                    last_exc_message = this_exc_message
+                    exc_count = 0
+                # This should be a very rare event. In case it isn't, do
+                # a sleep.
+                time.sleep(1)
+    return inner_func

neutron/openstack/common/rpc/impl_kombu.py

@@ -30,6 +30,7 @@ import kombu.entity
 import kombu.messaging
 from oslo.config import cfg
 
+from neutron.openstack.common import excutils
 from neutron.openstack.common.gettextutils import _
 from neutron.openstack.common import network_utils
 from neutron.openstack.common.rpc import amqp as rpc_amqp
@@ -721,6 +722,7 @@ class Connection(object):
 
     def consume_in_thread(self):
         """Consumer from all queues/consumers in a greenthread."""
+        @excutils.forever_retry_uncaught_exceptions
         def _consumer_thread():
             try:
                 self.consume()

neutron/openstack/common/rpc/impl_qpid.py

@@ -23,6 +23,7 @@ import eventlet
 import greenlet
 from oslo.config import cfg
 
+from neutron.openstack.common import excutils
 from neutron.openstack.common.gettextutils import _
 from neutron.openstack.common import importutils
 from neutron.openstack.common import jsonutils
@@ -679,6 +680,7 @@ class Connection(object):
 
     def consume_in_thread(self):
         """Consumer from all queues/consumers in a greenthread."""
+        @excutils.forever_retry_uncaught_exceptions
         def _consumer_thread():
             try:
                 self.consume()