ceph/src/pybind/mgr/cephadm/offline_watcher.py

   1 import logging
   2 from typing import List, Optional, TYPE_CHECKING
   3
   4 import multiprocessing as mp
   5 import threading
   6
   7 if TYPE_CHECKING:
   8     from cephadm.module import CephadmOrchestrator
   9
  10 logger = logging.getLogger(__name__)
  11
  12
  13 class OfflineHostWatcher(threading.Thread):
  14     def __init__(self, mgr: "CephadmOrchestrator") -> None:
  15         self.mgr = mgr
  16         self.hosts: Optional[List[str]] = None
  17         self.new_hosts: Optional[List[str]] = None
  18         self.stop = False
  19         self.event = threading.Event()
  20         super(OfflineHostWatcher, self).__init__(target=self.run)
  21
  22     def run(self) -> None:
  23         self.thread_pool = mp.pool.ThreadPool(10)
  24         while not self.stop:
  25             # only need to take action if we have hosts to check
  26             if self.hosts or self.new_hosts:
  27                 if self.new_hosts:
  28                     self.hosts = self.new_hosts
  29                     self.new_hosts = None
  30                 logger.debug(f'OfflineHostDetector: Checking if hosts: {self.hosts} are offline.')
  31                 assert self.hosts is not None
  32                 self.thread_pool.map(self.check_host, self.hosts)
  33             self.event.wait(20)
  34             self.event.clear()
  35         self.thread_pool.close()
  36         self.thread_pool.join()
  37
  38     def check_host(self, host: str) -> None:
  39         if host not in self.mgr.offline_hosts:
  40             try:
  41                 self.mgr.ssh.check_execute_command(host, ['true'])
  42             except Exception:
  43                 logger.debug(f'OfflineHostDetector: detected {host} to be offline')
  44                 # kick serve loop in case corrective action must be taken for offline host
  45                 self.mgr._kick_serve_loop()
  46
  47     def set_hosts(self, hosts: List[str]) -> None:
  48         hosts.sort()
  49         if (not self.hosts or self.hosts != hosts) and hosts:
  50             self.new_hosts = hosts
  51             logger.debug(
  52                 f'OfflineHostDetector: Hosts to check if offline swapped to: {self.new_hosts}.')
  53             self.wakeup()
  54
  55     def wakeup(self) -> None:
  56         self.event.set()
  57
  58     def shutdown(self) -> None:
  59         self.stop = True
  60         self.wakeup()