mgr/cephadm: add scheduled_daemon_actions to HostCache

Signed-off-by: Sebastian Wagner <sebastian.wagner@suse.com>
(cherry picked from commit ce4798ba0d33d2c8cfc7811553eed11f0f6af781)

diff --git a/src/pybind/mgr/cephadm/inventory.py b/src/pybind/mgr/cephadm/inventory.py
index c9b761f54e1e1f166dd3febf66833b66ccc58aea..c238cb7eba669ef72cece8ec380695a3553afaa7 100644 (file)
--- a/src/pybind/mgr/cephadm/inventory.py
+++ b/src/pybind/mgr/cephadm/inventory.py
@@ -186,6 +186,8 @@ class HostCache():
         self.last_etc_ceph_ceph_conf: Dict[str, datetime.datetime] = {}
         self.registry_login_queue: Set[str] = set()
 
+        self.scheduled_daemon_actions: Dict[str, Dict[str, str]] = {}
+
     def load(self):
         # type: () -> None
         for k, v in six.iteritems(self.mgr.get_store_prefix(HOST_CACHE_PREFIX)):
@@ -230,6 +232,8 @@ class HostCache():
                     self.last_etc_ceph_ceph_conf[host] = datetime.datetime.strptime(
                         j['last_etc_ceph_ceph_conf'], DATEFMT)
                 self.registry_login_queue.add(host)
+                self.scheduled_daemon_actions[host] = j.get('scheduled_daemon_actions', {})
+
                 self.mgr.log.debug(
                     'HostCache.load: host %s has %d daemons, '
                     '%d devices, %d networks' % (
@@ -293,9 +297,8 @@ class HostCache():
     def distribute_new_registry_login_info(self):
         self.registry_login_queue = set(self.mgr.inventory.keys())
 
-    def save_host(self, host):
-        # type: (str) -> None
-        j = {   # type: ignore
+    def save_host(self, host: str) -> None:
+        j: Dict[str, Any] = {
             'daemons': {},
             'devices': [],
             'osdspec_previews': [],
@@ -303,17 +306,17 @@ class HostCache():
         }
         if host in self.last_daemon_update:
             j['last_daemon_update'] = self.last_daemon_update[host].strftime(
-                DATEFMT)  # type: ignore
+                DATEFMT)
         if host in self.last_device_update:
             j['last_device_update'] = self.last_device_update[host].strftime(
-                DATEFMT)  # type: ignore
+                DATEFMT)
         for name, dd in self.daemons[host].items():
-            j['daemons'][name] = dd.to_json()  # type: ignore
+            j['daemons'][name] = dd.to_json()
         for d in self.devices[host]:
-            j['devices'].append(d.to_json())  # type: ignore
+            j['devices'].append(d.to_json())
         j['networks'] = self.networks[host]
         for name, depi in self.daemon_config_deps[host].items():
-            j['daemon_config_deps'][name] = {   # type: ignore
+            j['daemon_config_deps'][name] = {
                 'deps': depi.get('deps', []),
                 'last_config': depi['last_config'].strftime(DATEFMT),
             }
@@ -325,6 +328,8 @@ class HostCache():
 
         if host in self.last_etc_ceph_ceph_conf:
             j['last_etc_ceph_ceph_conf'] = self.last_etc_ceph_ceph_conf[host].strftime(DATEFMT)
+        if self.scheduled_daemon_actions.get(host, {}):
+            j['scheduled_daemon_actions'] = self.scheduled_daemon_actions[host]
 
         self.mgr.set_store(HOST_CACHE_PREFIX + host, json.dumps(j))
 
@@ -346,6 +351,8 @@ class HostCache():
             del self.last_device_update[host]
         if host in self.daemon_config_deps:
             del self.daemon_config_deps[host]
+        if host in self.scheduled_daemon_actions:
+            del self.scheduled_daemon_actions[host]
         self.mgr.set_store(HOST_CACHE_PREFIX + host, None)
 
     def get_hosts(self):
@@ -481,7 +488,7 @@ class HostCache():
             return
         self.last_etc_ceph_ceph_conf[host] = self.mgr.last_monmap
 
-    def host_needs_registry_login(self, host):
+    def host_needs_registry_login(self, host: str) -> bool:
         if host in self.mgr.offline_hosts:
             return False
         if host in self.registry_login_queue:
@@ -510,6 +517,34 @@ class HostCache():
         return all((h in self.last_daemon_update or h in self.mgr.offline_hosts)
                    for h in self.get_hosts())
 
+    def schedule_daemon_action(self, host: str, daemon_name: str, action: str):
+        priorities = {
+            'start': 1,
+            'restart': 2,
+            'reconfig': 3,
+            'redeploy': 4,
+            'stop': 5,
+        }
+        existing_action = self.scheduled_daemon_actions.get(host, {}).get(daemon_name, None)
+        if existing_action and priorities[existing_action] > priorities[action]:
+            logger.debug(
+                f'skipping {action}ing {daemon_name}, cause {existing_action} already scheduled.')
+            return
+
+        if host not in self.scheduled_daemon_actions:
+            self.scheduled_daemon_actions[host] = {}
+        self.scheduled_daemon_actions[host][daemon_name] = action
+
+    def rm_scheduled_daemon_action(self, host: str, daemon_name: str):
+        if host in self.scheduled_daemon_actions:
+            if daemon_name in self.scheduled_daemon_actions[host]:
+                del self.scheduled_daemon_actions[host][daemon_name]
+            if not self.scheduled_daemon_actions[host]:
+                del self.scheduled_daemon_actions[host]
+
+    def get_scheduled_daemon_action(self, host, daemon) -> Optional[str]:
+        return self.scheduled_daemon_actions.get(host, {}).get(daemon)
+
 
 class EventStore():
     def __init__(self, mgr):