mgr/cephadm: cache ssh connections

Instead of opening a fresh connection each time we run a command, keep
a handle for open connections.

- close the connection on any error
- close all connections if the ssh config changes
- close the connection when we remove a host

Signed-off-by: Sage Weil <sage@redhat.com>

diff --git a/src/pybind/mgr/cephadm/module.py b/src/pybind/mgr/cephadm/module.py
index 6b73e00a7c55481a3a7d79b4c1d25ee56bae6c9b..bd5490652bca58fcf9c1a79f19cea98235270dfa 100644 (file)
--- a/src/pybind/mgr/cephadm/module.py
+++ b/src/pybind/mgr/cephadm/module.py
@@ -7,7 +7,7 @@ from functools import wraps
 
 import string
 try:
-    from typing import List, Dict, Optional, Callable, TypeVar, Type, Any
+    from typing import List, Dict, Optional, Callable, Tuple, TypeVar, Type, Any
     from typing import TYPE_CHECKING
 except ImportError:
     TYPE_CHECKING = False  # just for type checking
@@ -344,6 +344,9 @@ class CephadmOrchestrator(MgrModule, orchestrator.OrchestratorClientMixin):
             self.container_image_base = ''
             self.warn_on_stray_hosts = True
             self.warn_on_stray_services = True
+
+        self._cons = {}  # type: Dict[str, Tuple[remoto.backends.BaseConnection,remoto.backends.LegacyModuleExecute]]
+
         self.config_notify()
 
         path = self.get_ceph_option('cephadm_path')
@@ -782,6 +785,22 @@ class CephadmOrchestrator(MgrModule, orchestrator.OrchestratorClientMixin):
         elif self.mode == 'cephadm-package':
             self.ssh_user = 'cephadm'
 
+        self._reset_cons()
+
+    def _reset_con(self, host):
+        conn, r = self._cons.get(host, (None, None))
+        if conn:
+            self.log.debug('_reset_con close %s' % host)
+            conn.exit()
+            del self._cons[host]
+
+    def _reset_cons(self):
+        for host, conn_and_r in self._cons.items():
+            self.log.debug('_reset_cons close %s' % host)
+            conn, r = conn_and_r
+            conn.exit()
+        self._cons = {}
+
     @staticmethod
     def can_run():
         if remoto is not None:
@@ -913,6 +932,10 @@ class CephadmOrchestrator(MgrModule, orchestrator.OrchestratorClientMixin):
         """
         Setup a connection for running commands on remote host.
         """
+        conn_and_r = self._cons.get(host)
+        if conn_and_r:
+            self.log.debug('Have connection to %s' % host)
+            return conn_and_r
         n = self.ssh_user + '@' + host
         self.log.info("Opening connection to {} with ssh options '{}'".format(
             n, self._ssh_options))
@@ -922,6 +945,7 @@ class CephadmOrchestrator(MgrModule, orchestrator.OrchestratorClientMixin):
             ssh_options=self._ssh_options)
 
         r = conn.import_module(remotes)
+        self._cons[host] = conn, r
 
         return conn, r
 
@@ -988,6 +1012,7 @@ class CephadmOrchestrator(MgrModule, orchestrator.OrchestratorClientMixin):
                         [python, '-u'],
                         stdin=script.encode('utf-8'))
                 except RuntimeError as e:
+                    self._reset_con(host)
                     if error_ok:
                         return '', str(e), 1
                     raise
@@ -998,6 +1023,7 @@ class CephadmOrchestrator(MgrModule, orchestrator.OrchestratorClientMixin):
                         ['sudo', '/usr/bin/cephadm'] + final_args,
                         stdin=stdin)
                 except RuntimeError as e:
+                    self._reset_con(host)
                     if error_ok:
                         return '', str(e), 1
                     raise
@@ -1014,9 +1040,6 @@ class CephadmOrchestrator(MgrModule, orchestrator.OrchestratorClientMixin):
             self.log.exception(ex)
             raise
 
-        finally:
-            conn.exit()
-
     def _get_hosts(self, wanted=None):
         return self.inventory_cache.items_filtered(wanted)
 
@@ -1050,6 +1073,7 @@ class CephadmOrchestrator(MgrModule, orchestrator.OrchestratorClientMixin):
         self._save_inventory()
         del self.inventory_cache[host]
         del self.service_cache[host]
+        self._reset_con(host)
         self.event.set()  # refresh stray health check
         return "Removed host '{}'".format(host)