Optimize RPC client instantiations.

coriolis/conductor/rpc/server.py

@@ -20,7 +20,7 @@ from coriolis.db.sqlalchemy import models
 from coriolis import exception
 from coriolis import keystone
 from coriolis.licensing import client as licensing_client
-from coriolis.minion_manager.rpc import client as minion_manager_client
+from coriolis.minion_manager.rpc import client as rpc_minion_manager_client
 from coriolis.replica_cron.rpc import client as rpc_cron_client
 from coriolis.scheduler.rpc import client as rpc_scheduler_client
 from coriolis import schemas
@@ -175,6 +175,7 @@ class ConductorServerEndpoint(object):
         self._worker_client_instance = None
         self._scheduler_client_instance = None
         self._replica_cron_client_instance = None
+        self._minion_manager_client_instance = None
 
     # NOTE(aznashwan): it is unsafe to fork processes with pre-instantiated
     # oslo_messaging clients as the underlying eventlet thread queues will
@@ -204,7 +205,10 @@ class ConductorServerEndpoint(object):
 
     @property
     def _minion_manager_client(self):
-        return minion_manager_client.MinionManagerClient()
+        if not getattr(self, '_minion_manager_client_instance'):
+            self._minion_manager_client_instance = (
+                rpc_minion_manager_client.MinionManagerClient())
+        return self._minion_manager_client_instance
 
     def get_all_diagnostics(self, ctxt):
         client_objects = {

coriolis/minion_manager/rpc/server.py

@@ -65,6 +65,11 @@ class MinionManagerServerEndpoint(object):
 
     def __init__(self):
         self._admin_ctxt = context.get_admin_context()
+        self._scheduler_client_instance = None
+        self._worker_client_instance = None
+        self._conductor_client_instance = None
+        self._replica_cron_client_instance = None
+        self._minion_manager_client_instance = None
         try:
             self._cron = cron.Cron()
             self._init_pools_refresh_cron_jobs()
@@ -180,22 +185,34 @@ class MinionManagerServerEndpoint(object):
     # oslo_messaging clients as the underlying eventlet thread queues will
     # be invalidated. Considering this class both serves from a "main
     # process" as well as forking child processes, it is safest to
-    # re-instantiate the clients every time:
+    # instantiate the clients only when needed:
     @property
     def _rpc_worker_client(self):
-        return rpc_worker_client.WorkerClient()
+        if not getattr(self, '_worker_client_instance', None):
+            self._worker_client_instance = (
+                rpc_worker_client.WorkerClient())
+        return self._worker_client_instance
 
     @property
     def _rpc_scheduler_client(self):
-        return rpc_scheduler_client.SchedulerClient()
+        if not getattr(self, '_scheduler_client_instance', None):
+            self._scheduler_client_instance = (
+                rpc_scheduler_client.SchedulerClient())
+        return self._scheduler_client_instance
 
     @property
     def _rpc_conductor_client(self):
-        return rpc_conductor_client.ConductorClient()
+        if not getattr(self, '_conductor_client_instance', None):
+            self._conductor_client_instance = (
+                rpc_conductor_client.ConductorClient())
+        return self._conductor_client_instance
 
     @property
     def _rpc_minion_manager_client(self):
-        return rpc_minion_manager_client.MinionManagerClient()
+        if not getattr(self, '_minion_manager_client_instance', None):
+            self._minion_manager_client_instance = (
+                rpc_minion_manager_client.MinionManagerClient())
+        return self._minion_manager_client_instance
 
     def get_diagnostics(self, ctxt):
         return utils.get_diagnostics_info()

coriolis/minion_manager/rpc/tasks.py

@@ -79,11 +79,17 @@ class MinionManagerTaskEventMixin(object):
     # re-instantiate the clients every time:
     @property
     def _conductor_client(self):
-        return rpc_conductor_client.ConductorClient()
+        if not getattr(self, '_conductor_client_instance', None):
+            self._conductor_client_instance = (
+                rpc_conductor_client.ConductorClient())
+        return self._conductor_client_instance
 
     @property
     def _minion_manager_client(self):
-        return rpc_minion_manager_client.MinionManagerClient()
+        if not getattr(self, '_minion_manager_client_instance', None):
+            self._minion_manager_client_instance = (
+                rpc_minion_manager_client.MinionManagerClient())
+        return self._minion_manager_client_instance
 
     def _add_minion_pool_event(
             self, context, message, level=constants.TASK_EVENT_INFO):

coriolis/taskflow/base.py

@@ -114,12 +114,16 @@ class BaseRunWorkerTask(BaseCoriolisTaskflowTask):
         self._main_task_runner_type = main_task_runner_type
         self._cleanup_task_runner_type = cleanup_task_runner_type
         self._raise_on_cleanup_failure = raise_on_cleanup_failure
+        self._scheduler_client_instance = None
 
         super(BaseRunWorkerTask, self).__init__(name=task_name, **kwargs)
 
     @property
     def _scheduler_client(self):
-        return rpc_scheduler_client.SchedulerClient()
+        if not getattr(self, '_scheduler_client_instance', None):
+            self._scheduler_client_instance = (
+                rpc_scheduler_client.SchedulerClient())
+        return self._scheduler_client_instance
 
     def _set_provides_for_dependencies(self, kwargs):
         dep = TASK_RETURN_VALUE_FORMAT % self._task_name