Support magnum-conductor multiple process workers

Multiple process workers support for magnum-conductor.
Adds new option 'workers' to group [conductor] of magnum.conf.

Change-Id: If4d47769c97f756dbf5f45ac4413df7971731f21
Implements: blueprint magnum-multiple-process-workers

magnum/cmd/conductor.py

@@ -17,6 +17,7 @@
 import os
 import sys
 
+from oslo_concurrency import processutils
 from oslo_log import log as logging
 from oslo_reports import guru_meditation_report as gmr
 from oslo_service import service
@@ -55,5 +56,8 @@ def main():
     server = rpc_service.Service.create(CONF.conductor.topic,
                                         conductor_id, endpoints,
                                         binary='magnum-conductor')
-    launcher = service.launch(CONF, server)
+    workers = CONF.conductor.workers
+    if not workers:
+        workers = processutils.get_worker_count()
+    launcher = service.launch(CONF, server, workers=workers)
     launcher.wait()

magnum/conf/conductor.py

@@ -24,6 +24,9 @@ conductor_service_opts = [
                default=4,
                help=('RPC timeout for the conductor liveness check that is '
                      'used for cluster locking.')),
+    cfg.IntOpt('workers',
+               help='Number of magnum-conductor processes to fork and run. '
+                    'Default to number of CPUs on the host.')
 ]
 
 

magnum/tests/functional/python_client_base.py

@@ -263,6 +263,14 @@ extendedKeyUsage = clientAuth
         cls.cluster = cls._create_cluster(cls.__name__,
                                           cls.cluster_template.uuid)
         if not cls.cluster_template_kwargs.get('tls_disabled', False):
+            # NOTE (wangbo) with multiple mangum-conductor processes, client
+            # ca files should be created after completion of cluster ca_cert
+            cls._wait_on_status(
+                cls.cluster,
+                [None, "CREATE_IN_PROGRESS"],
+                ["CREATE_FAILED", "CREATE_COMPLETE"],
+                timeout=cls.cluster_complete_timeout
+            )
             cls._create_tls_ca_files(cls.config_contents)
 
     @classmethod

magnum/tests/unit/cmd/test_conductor.py

@@ -14,6 +14,8 @@
 
 import mock
 
+from oslo_concurrency import processutils
+
 from magnum.cmd import conductor
 from magnum.tests import base
 
@@ -32,5 +34,25 @@ class TestMagnumConductor(base.TestCase):
         mock_rpc.Service.create.assert_called_once_with(
             base.CONF.conductor.topic,
             mock.ANY, mock.ANY, binary='magnum-conductor')
-        mock_launch.assert_called_once_with(base.CONF, server)
+        workers = processutils.get_worker_count()
+        mock_launch.assert_called_once_with(base.CONF, server,
+                                            workers=workers)
+        launcher.wait.assert_called_once_with()
+
+    @mock.patch('oslo_service.service.launch')
+    @mock.patch.object(conductor, 'rpc_service')
+    @mock.patch('magnum.common.service.prepare_service')
+    def test_conductor_config_workers(self, mock_prep, mock_rpc, mock_launch):
+        fake_workers = 8
+        self.config(workers=fake_workers, group='conductor')
+        conductor.main()
+
+        server = mock_rpc.Service.create.return_value
+        launcher = mock_launch.return_value
+        mock_prep.assert_called_once_with(mock.ANY)
+        mock_rpc.Service.create.assert_called_once_with(
+            base.CONF.conductor.topic,
+            mock.ANY, mock.ANY, binary='magnum-conductor')
+        mock_launch.assert_called_once_with(base.CONF, server,
+                                            workers=fake_workers)
         launcher.wait.assert_called_once_with()