Merge "engine: delete workers from etcd after removing"

qinling/engine/default_engine.py

@@ -245,5 +245,6 @@ class DefaultEngine(object):
         for worker in workers:
             LOG.debug('Removing worker %s', worker)
             self.orchestrator.delete_worker(worker)
+            etcd_util.delete_worker(function_id, worker)
 
         LOG.info('Finished scaling down function %s.', function_id)

qinling/tests/unit/engine/test_default_engine.py

@@ -430,23 +430,33 @@ class TestDefaultEngine(base.DbTestCase):
         etcd_util_create_service_url_mock.assert_called_once_with(
             function_id, 'url')
 
+    @mock.patch('qinling.utils.etcd_util.delete_worker')
     @mock.patch('qinling.utils.etcd_util.get_workers')
-    def test_scaledown_function(self, etcd_util_get_workers_mock):
+    def test_scaledown_function(
+            self, etcd_util_get_workers_mock, etcd_util_delete_workers_mock
+    ):
         function_id = common.generate_unicode_uuid()
-        etcd_util_get_workers_mock.return_value = range(4)
+        etcd_util_get_workers_mock.return_value = [
+            'worker_%d' % i for i in range(4)
+        ]
 
         self.default_engine.scaledown_function(mock.Mock(), function_id)
 
         etcd_util_get_workers_mock.assert_called_once_with(
             function_id)
-        self.orchestrator.delete_worker.assert_called_once_with(0)
+        self.orchestrator.delete_worker.assert_called_once_with('worker_0')
+        etcd_util_delete_workers_mock.assert_called_once_with(
+            function_id, 'worker_0')
 
+    @mock.patch('qinling.utils.etcd_util.delete_worker')
     @mock.patch('qinling.utils.etcd_util.get_workers')
     def test_scaledown_function_multiple_workers(
-            self, etcd_util_get_workers_mock
+            self, etcd_util_get_workers_mock, etcd_util_delete_workers_mock
     ):
         function_id = common.generate_unicode_uuid()
-        etcd_util_get_workers_mock.return_value = range(4)
+        etcd_util_get_workers_mock.return_value = [
+            'worker_%d' % i for i in range(4)
+        ]
 
         self.default_engine.scaledown_function(
             mock.Mock(), function_id, count=2)
@@ -454,16 +464,25 @@ class TestDefaultEngine(base.DbTestCase):
         etcd_util_get_workers_mock.assert_called_once_with(
             function_id)
         # First two workers will be deleted.
-        expected = [mock.call(0), mock.call(1)]
+        expected = [mock.call('worker_0'), mock.call('worker_1')]
         self.orchestrator.delete_worker.assert_has_calls(expected)
         self.assertEqual(2, self.orchestrator.delete_worker.call_count)
+        expected = [
+            mock.call(function_id, 'worker_0'),
+            mock.call(function_id, 'worker_1')
+        ]
+        etcd_util_delete_workers_mock.assert_has_calls(expected)
+        self.assertEqual(2, etcd_util_delete_workers_mock.call_count)
 
+    @mock.patch('qinling.utils.etcd_util.delete_worker')
     @mock.patch('qinling.utils.etcd_util.get_workers')
     def test_scaledown_function_leaving_one_worker(
-            self, etcd_util_get_workers_mock
+            self, etcd_util_get_workers_mock, etcd_util_delete_workers_mock
     ):
         function_id = common.generate_unicode_uuid()
-        etcd_util_get_workers_mock.return_value = range(4)
+        etcd_util_get_workers_mock.return_value = [
+            'worker_%d' % i for i in range(4)
+        ]
 
         self.default_engine.scaledown_function(
             mock.Mock(), function_id, count=5)  # count > len(workers)
@@ -471,6 +490,15 @@ class TestDefaultEngine(base.DbTestCase):
         etcd_util_get_workers_mock.assert_called_once_with(
             function_id)
         # Only the first three workers will be deleted
-        expected = [mock.call(0), mock.call(1), mock.call(2)]
+        expected = [
+            mock.call('worker_0'), mock.call('worker_1'), mock.call('worker_2')
+        ]
         self.orchestrator.delete_worker.assert_has_calls(expected)
         self.assertEqual(3, self.orchestrator.delete_worker.call_count)
+        expected = [
+            mock.call(function_id, 'worker_0'),
+            mock.call(function_id, 'worker_1'),
+            mock.call(function_id, 'worker_2')
+        ]
+        etcd_util_delete_workers_mock.assert_has_calls(expected)
+        self.assertEqual(3, etcd_util_delete_workers_mock.call_count)

qinling/utils/etcd_util.py

@@ -14,7 +14,6 @@
 
 import etcd3gw
 from oslo_config import cfg
-from oslo_utils import uuidutils
 
 CONF = cfg.CONF
 CLIENT = None
@@ -36,13 +35,25 @@ def get_worker_lock():
 
 
 def create_worker(function_id, worker):
+    """Create the worker info in etcd.
+
+    The worker parameter is assumed to be unique.
+    """
+    # NOTE(huntxu): for the kubernetes orchestrator, which is the the only
+    # available orchestrator at the moment, the value of the worker param
+    # is the name of the pod so it is unique.
     client = get_client()
     client.create(
-        '%s/worker_%s' % (function_id, uuidutils.generate_uuid()),
+        '%s/worker_%s' % (function_id, worker),
         worker
     )
 
 
+def delete_worker(function_id, worker):
+    client = get_client()
+    client.delete('%s/worker_%s' % (function_id, worker))
+
+
 def get_workers(function_id, conf=None):
     client = get_client(conf)
     values = client.get_prefix('%s/worker' % function_id)