Add WBE worker expiry

When a worker hasn't responded to a notification
request for a given amount of time remove it from
being a useable worker that we can match task submissions
to.

Change-Id: I596bccc1c42f83ee79136dd27bc87039154ff7b1

2 files changed, 20 insertions, 5 deletions

diff --git a/taskflow/tests/unit/worker_based/test_creation.py b/taskflow/tests/unit/worker_based/test_creation.py
index 5b689de..7d2b75d 100644
--- a/taskflow/tests/unit/worker_based/test_creation.py
+++ b/taskflow/tests/unit/worker_based/test_creation.py
@@ -50,7 +50,8 @@ class TestWorkerBasedActionEngine(test.MockTestCase):
                                      transport=None,
                                      transport_options=None,
                                      transition_timeout=mock.ANY,
-                                     retry_options=None)
+                                     retry_options=None,
+                                     worker_expiry=mock.ANY)
         ]
         self.assertEqual(expected_calls, self.master_mock.mock_calls)
 
@@ -66,7 +67,8 @@ class TestWorkerBasedActionEngine(test.MockTestCase):
             transport_options={},
             transition_timeout=200,
             topics=topics,
-            retry_options={})
+            retry_options={},
+            worker_expiry=1)
         expected_calls = [
             mock.call.executor_class(uuid=eng.storage.flow_uuid,
                                      url=broker_url,
@@ -75,7 +77,8 @@ class TestWorkerBasedActionEngine(test.MockTestCase):
                                      transport='memory',
                                      transport_options={},
                                      transition_timeout=200,
-                                     retry_options={})
+                                     retry_options={},
+                                     worker_expiry=1)
         ]
         self.assertEqual(expected_calls, self.master_mock.mock_calls)
 
diff --git a/taskflow/tests/unit/worker_based/test_types.py b/taskflow/tests/unit/worker_based/test_types.py
index 6809aef..e81aa79 100644
--- a/taskflow/tests/unit/worker_based/test_types.py
+++ b/taskflow/tests/unit/worker_based/test_types.py
@@ -33,12 +33,24 @@ class TestTopicWorker(test.TestCase):
 
 
 class TestProxyFinder(test.TestCase):
+
+    @mock.patch("oslo_utils.timeutils.now")
+    def test_expiry(self, mock_now):
+        finder = worker_types.ProxyWorkerFinder('me', mock.MagicMock(), [],
+                                                worker_expiry=60)
+        w, emit = finder._add('dummy-topic', [utils.DummyTask])
+        w.last_seen = 0
+        mock_now.side_effect = [120]
+        gone = finder.clean()
+        self.assertEqual(0, finder.total_workers)
+        self.assertEqual(1, gone)
+
     def test_single_topic_worker(self):
         finder = worker_types.ProxyWorkerFinder('me', mock.MagicMock(), [])
         w, emit = finder._add('dummy-topic', [utils.DummyTask])
         self.assertIsNotNone(w)
         self.assertTrue(emit)
-        self.assertEqual(1, finder.total_workers())
+        self.assertEqual(1, finder.total_workers)
         w2 = finder.get_worker_for_task(utils.DummyTask)
         self.assertEqual(w.identity, w2.identity)
 
@@ -60,7 +72,7 @@ class TestProxyFinder(test.TestCase):
         added.append(finder._add('dummy-topic', [utils.DummyTask]))
         added.append(finder._add('dummy-topic-2', [utils.DummyTask]))
         added.append(finder._add('dummy-topic-3', [utils.NastyTask]))
-        self.assertEqual(3, finder.total_workers())
+        self.assertEqual(3, finder.total_workers)
         w = finder.get_worker_for_task(utils.NastyTask)
         self.assertEqual(added[-1][0].identity, w.identity)
         w = finder.get_worker_for_task(utils.DummyTask)