Merge "Prevent leaked eventlets to send notifications" into stable/xena

Author: Zuul

nova/test.py

@@ -177,6 +177,9 @@ def setUp(self):
         with fixtures.EnvironmentVariable('OS_LOG_CAPTURE', '0'):
             super(TestCase, self).setUp()
 
+        self.useFixture(
+            nova_fixtures.PropagateTestCaseIdToChildEventlets(self.id()))
+
         # How many of which service we've started. {$service-name: $count}
         self._service_fixture_count = collections.defaultdict(int)
 

nova/tests/fixtures/notifications.py

@@ -14,6 +14,7 @@
 import functools
 import threading
 
+import eventlet
 import fixtures
 from oslo_log import log as logging
 import oslo_messaging
@@ -76,6 +77,7 @@ class FakeNotifier(object):
 
     def __init__(
         self, transport, publisher_id, serializer=None, parent=None,
+        test_case_id=None
     ):
         self.transport = transport
         self.publisher_id = publisher_id
@@ -92,13 +94,16 @@ def __init__(
                 functools.partial(self._notify, priority.upper()),
             )
 
+        self.test_case_id = test_case_id
+
     def prepare(self, publisher_id=None):
         if publisher_id is None:
             publisher_id = self.publisher_id
 
         return self.__class__(
             self.transport, publisher_id,
             serializer=self._serializer, parent=self,
+            test_case_id=self.test_case_id
         )
 
     def _notify(self, priority, ctxt, event_type, payload):
@@ -130,8 +135,10 @@ def reset(self):
 class FakeVersionedNotifier(FakeNotifier):
     def __init__(
         self, transport, publisher_id, serializer=None, parent=None,
+        test_case_id=None
     ):
-        super().__init__(transport, publisher_id, serializer)
+        super().__init__(
+            transport, publisher_id, serializer, test_case_id=test_case_id)
         if parent:
             self.versioned_notifications = parent.versioned_notifications
         else:
@@ -142,7 +149,31 @@ def __init__(
         else:
             self.subscriptions = collections.defaultdict(_Sub)
 
+    @staticmethod
+    def _get_sender_test_case_id():
+        current = eventlet.getcurrent()
+        # NOTE(gibi) not all eventlet spawn is under our control, so there can
+        # be senders without test_case_id set, find the first ancestor that
+        # was spawned from nova.utils.spawn[_n] and therefore has the id set.
+        while not getattr(current, 'test_case_id', None):
+            current = current.parent
+        return current.test_case_id
+
     def _notify(self, priority, ctxt, event_type, payload):
+        sender_test_case_id = self._get_sender_test_case_id()
+        # NOTE(gibi): this is here to prevent late notifications from already
+        # finished test cases to break the currently running test case. See
+        # more in https://bugs.launchpad.net/nova/+bug/1946339
+        if sender_test_case_id != self.test_case_id:
+            raise RuntimeError(
+                'FakeVersionedNotifier received %s notification emitted by %s '
+                'test case which is different from the currently running test '
+                'case %s. This notification is ignored. The sender test case '
+                'probably leaked a running eventlet that emitted '
+                'notifications after the test case finished. Now this eventlet'
+                'is terminated by raising this exception.' %
+                (event_type, sender_test_case_id, self.test_case_id))
+
         payload = self._serializer.serialize_entity(ctxt, payload)
         notification = {
             'publisher_id': self.publisher_id,
@@ -180,7 +211,9 @@ def setUp(self):
         self.fake_versioned_notifier = FakeVersionedNotifier(
             rpc.NOTIFIER.transport,
             rpc.NOTIFIER.publisher_id,
-            serializer=getattr(rpc.NOTIFIER, '_serializer', None))
+            serializer=getattr(rpc.NOTIFIER, '_serializer', None),
+            test_case_id=self.test.id()
+        )
         if rpc.LEGACY_NOTIFIER and rpc.NOTIFIER:
             self.test.stub_out('nova.rpc.LEGACY_NOTIFIER', self.fake_notifier)
             self.test.stub_out(

nova/tests/fixtures/nova.py

@@ -18,10 +18,12 @@
 
 import collections
 from contextlib import contextmanager
+import functools
 import logging as std_logging
 import os
 import warnings
 
+import eventlet
 import fixtures
 import futurist
 import mock
@@ -56,6 +58,7 @@
 from nova.scheduler import weights
 from nova import service
 from nova.tests.functional.api import client
+from nova import utils
 
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
@@ -1516,3 +1519,84 @@ def fail(*a, **k):
             except ImportError:
                 self.useFixture(fixtures.MonkeyPatch(
                     meth, poison_configure(meth, why)))
+
+
+class PropagateTestCaseIdToChildEventlets(fixtures.Fixture):
+    """A fixture that adds the currently running test case id to each spawned
+    eventlet. This information then later used by the NotificationFixture to
+    detect if a notification was emitted by an eventlet that was spawned by a
+    previous test case so such late notification can be ignored. For more
+    background about what issues this can prevent see
+    https://bugs.launchpad.net/nova/+bug/1946339
+
+    """
+    def __init__(self, test_case_id):
+        self.test_case_id = test_case_id
+
+    def setUp(self):
+        super().setUp()
+
+        # set the id on the main eventlet
+        c = eventlet.getcurrent()
+        c.test_case_id = self.test_case_id
+
+        orig_spawn = utils.spawn
+
+        def wrapped_spawn(func, *args, **kwargs):
+            # This is still runs before the eventlet.spawn so read the id for
+            # propagation
+            caller = eventlet.getcurrent()
+            # If there is no id set on us that means we were spawned with other
+            # than nova.utils.spawn or spawn_n so the id propagation chain got
+            # broken. We fall back to self.test_case_id from the fixture which
+            # is good enough
+            caller_test_case_id = getattr(
+                caller, 'test_case_id', None) or self.test_case_id
+
+            @functools.wraps(func)
+            def test_case_id_wrapper(*args, **kwargs):
+                # This runs after the eventlet.spawn in the new child.
+                # Propagate the id from our caller eventlet
+                current = eventlet.getcurrent()
+                current.test_case_id = caller_test_case_id
+                return func(*args, **kwargs)
+
+            # call the original spawn to create the child but with our
+            # new wrapper around its target
+            return orig_spawn(test_case_id_wrapper, *args, **kwargs)
+
+        # let's replace nova.utils.spawn with the wrapped one that injects
+        # our initialization to the child eventlet
+        self.useFixture(
+            fixtures.MonkeyPatch('nova.utils.spawn', wrapped_spawn))
+
+        # now do the same with spawn_n
+        orig_spawn_n = utils.spawn_n
+
+        def wrapped_spawn_n(func, *args, **kwargs):
+            # This is still runs before the eventlet.spawn so read the id for
+            # propagation
+            caller = eventlet.getcurrent()
+            # If there is no id set on us that means we were spawned with other
+            # than nova.utils.spawn or spawn_n so the id propagation chain got
+            # broken. We fall back to self.test_case_id from the fixture which
+            # is good enough
+            caller_test_case_id = getattr(
+                caller, 'test_case_id', None) or self.test_case_id
+
+            @functools.wraps(func)
+            def test_case_id_wrapper(*args, **kwargs):
+                # This runs after the eventlet.spawn in the new child.
+                # Propagate the id from our caller eventlet
+                current = eventlet.getcurrent()
+                current.test_case_id = caller_test_case_id
+                return func(*args, **kwargs)
+
+            # call the original spawn_n to create the child but with our
+            # new wrapper around its target
+            return orig_spawn_n(test_case_id_wrapper, *args, **kwargs)
+
+        # let's replace nova.utils.spawn_n with the wrapped one that injects
+        # our initialization to the child eventlet
+        self.useFixture(
+            fixtures.MonkeyPatch('nova.utils.spawn_n', wrapped_spawn_n))