mgr/cephadm: swap _rm_daemon to use cephadm _orch rm-daemons

As part of this

1) _remove_daemon is now async
2) remove_given_daemons is now async
3) _remove_daemon now calls `cephadm _orch rm-daemons` instead of `cephadm rm-daemon`
4) fixed an issue where we weren't properly building daemons to remove list

While _remove_daemon is now built for removing multiple daemons,
as of this call we still only ever pass it one at a time.

Signed-off-by: Adam King <[email protected]>
(cherry picked from commit 40a89d1)

Conflicts:
	src/pybind/mgr/cephadm/serve.py

Resolves: rhbz#2364414

Author: adk3798

src/cephadm/cephadm.py

@@ -3196,10 +3196,7 @@ def command_deploy(ctx):
     # type: (CephadmContext) -> None
     lock = FileLock(ctx, ctx.fsid)
     lock.acquire()
-    try:
-        _common_deploy(ctx)
-    except DaemonStartException:
-        sys.exit(DAEMON_FAILED_ERROR)
+    _common_deploy(ctx)
 
 
 def apply_deploy_config_to_ctx(
@@ -3249,8 +3246,6 @@ def _deploy_daemon(ctx: CephadmContext) -> Tuple[str, int]:
         rc: int = 0
         try:
             _common_deploy(ctx)
-        except DaemonStartException:
-            rc = DAEMON_FAILED_ERROR
         except Exception:
             # TODO: better rc based on exception?
             rc = -1
@@ -5931,6 +5926,7 @@ def _get_parser():
         default=False,
         action='store_true',
         help='Remove max coredump size drop-in and LimitCORE=infinity from unit file'
+    )
 
     parser_rm_daemon_from = subparsers_orch.add_parser(
         'rm-daemons', help='remove daemons')

src/pybind/mgr/cephadm/inventory.py

@@ -67,7 +67,6 @@ def __init__(
         self.hostname = hostname
 
 
-<<<<<<< HEAD
 class CoredumpctlOverrides:
     """
     Class to track supported coredumpctl override parameters
@@ -112,7 +111,8 @@ def __eq__(self, other: Any) -> bool:
                     return False
             return True
         return NotImplemented
-=======
+
+
 class DaemonDeployQueue:
     """
     Queue of daemons cephadm thinks it should deploy
@@ -166,7 +166,6 @@ def get_queued_daemon_placements_by_service(self, service_name: str) -> List[Dae
 
     def clear_queued_daemons(self) -> None:
         self._daemons = {}
->>>>>>> 9c228c014be (mgr/cephadm: add DaemonDeployQueue class)
 
 
 class DaemonRemovalQueue:

src/pybind/mgr/cephadm/module.py

@@ -2814,7 +2814,7 @@ def remove_daemons(self, names):
         if not args:
             raise OrchestratorError('Unable to find daemon(s) %s' % (names))
         self.log.info('Remove daemons %s' % ' '.join([a[0] for a in args]))
-        return self._remove_daemons(args)
+        return [msg for msgs in self._remove_daemons(args) for msg in msgs]
 
     @handle_orch_error
     def remove_service(self, service_name: str, force: bool = False) -> str:
@@ -3173,8 +3173,8 @@ def _calc_daemon_deps(self,
         return sorted(deps)
 
     @forall_hosts
-    def _remove_daemons(self, name: str, host: str) -> str:
-        return CephadmServe(self)._remove_daemon(name, host)
+    def _remove_daemons(self, name: str, host: str) -> List[str]:
+        return self.wait_async(CephadmServe(self)._remove_daemon([name], host))
 
     def _check_pool_exists(self, pool: str, service_name: str) -> None:
         logger.info(f'Checking pool "{pool}" exists for service {service_name}')

src/pybind/mgr/cephadm/serve.py

@@ -611,7 +611,7 @@ def _check_for_moved_osds(self) -> None:
             for e in error:
                 assert e.hostname
                 try:
-                    self._remove_daemon(e.name(), e.hostname, no_post_remove=True)
+                    self.mgr.wait_async(self._remove_daemon([e.name()], e.hostname, no_post_removals={e.name(): True}))
                     self.mgr.events.for_daemon(
                         e.name(), 'INFO', f"Removed duplicated daemon on host '{e.hostname}'")
                 except OrchestratorError as ex:
@@ -709,7 +709,7 @@ async def _parallel_deploy_and_remove(
             if daemons_placed:
                 r = True
                 hosts_altered.update(daemon_deployed_hosts)
-            removed_daemons, removed_daemon_hosts = self.remove_given_daemons(to_remove)
+            removed_daemons, removed_daemon_hosts = await self.remove_given_daemons(to_remove)
             if removed_daemons:
                 r = True
                 hosts_altered.update(conflict_hosts_altered)
@@ -1648,8 +1648,7 @@ async def _create_daemon(self,
                         ),
                         init_containers=init_containers,
                     ),
-                    config_blobs=daemon_spec.final_config,
-                    use_current_daemon_image=reconfig,
+                    config_blobs=daemon_spec.final_config
                 )
             )
 
@@ -1798,50 +1797,80 @@ def _setup_init_containers(
             ic_params.append(ic.to_json(flatten_args=True))
         return ic_meta
 
-    def _remove_daemon(self, name: str, host: str, no_post_remove: bool = False) -> str:
+    async def _remove_daemon(
+        self,
+        names: List[str],
+        host: str,
+        no_post_removals: Optional[Dict[str, bool]] = None
+    ) -> List[str]:
         """
-        Remove a daemon
+        Remove daemons
         """
-        dd = self.mgr.cache.get_daemon(name)
-        daemon_type = dd.daemon_type
-        daemon_id = dd.daemon_id
-        assert (daemon_type is not None and daemon_id is not None)
-        daemon = orchestrator.DaemonDescription(
-            daemon_type=daemon_type,
-            daemon_id=daemon_id,
-            service_name=dd.service_name(),
-            hostname=host)
-
-        with set_exception_subject('service', daemon.service_id(), overwrite=True):
-
-            service_registry.get_service(daemon_type_to_service(daemon_type)).pre_remove(daemon)
-            # NOTE: we are passing the 'force' flag here, which means
-            # we can delete a mon instances data.
-            if dd.ports:
-                args = ['--name', name, '--force', '--tcp-ports', ' '.join(map(str, dd.ports))]
-            else:
-                args = ['--name', name, '--force']
-
+        if not no_post_removals:
+            no_post_removals = {}
+        # Json blob specifying the removal info for each daemon.
+        # Currently that just consists of a name and list of ports.
+        # If it gets more complex we should add a class to
+        # exchange.Deploy to encapsulate it
+        daemon_rm_info: List[Dict[str, Union[str, List[str]]]] = []
+        for name in names:
+            dd = self.mgr.cache.get_daemon(name)
+            daemon_type = dd.daemon_type
+            daemon_id = dd.daemon_id
+            assert (daemon_type is not None and daemon_id is not None)
+            daemon = orchestrator.DaemonDescription(
+                daemon_type=daemon_type,
+                daemon_id=daemon_id,
+                service_name=dd.service_name(),
+                hostname=host)
+            with set_exception_subject('service', daemon.service_id(), overwrite=True):
+                service_registry.get_service(daemon_type_to_service(daemon_type)).pre_remove(daemon)
+            daemon_rm_info.append({'name': name, 'tcp_ports': [str(port) for port in (dd.ports or [])]})
             self.log.info('Removing daemon %s from %s -- ports %s' % (name, host, dd.ports))
-            with self.mgr.async_timeout_handler(host, f'cephadm rm-daemon (daemon {name})'):
-                out, err, code = self.mgr.wait_async(self._run_cephadm(
-                    host, name, 'rm-daemon', args))
-            if not code:
+
+        out, err, code = await self._run_cephadm(
+            host,
+            name,
+            ['_orch', 'rm-daemons'],
+            [],
+            stdin=json.dumps(daemon_rm_info),
+            error_ok=True
+        )
+
+        if len(out) != 1:  # _run_cephadm puts the output into a list, we should only get one thing here
+            raise OrchestratorError(f'Got unexpected non-length 1 list of output in _remove_daemon: {out}')
+        results: Dict[str, int] = json.loads(out[0])
+        msgs: List[str] = []
+        for name in names:
+            dd = self.mgr.cache.get_daemon(name)
+            daemon_type = dd.daemon_type
+            daemon_id = dd.daemon_id
+            assert (daemon_type is not None and daemon_id is not None)
+            daemon = orchestrator.DaemonDescription(
+                daemon_type=daemon_type,
+                daemon_id=daemon_id,
+                service_name=dd.service_name(),
+                hostname=host)
+            if results[name] == 0:
                 # remove item from cache
                 self.mgr.cache.rm_daemon(host, name)
-            self.mgr.cache.invalidate_host_daemons(host)
 
-            if not no_post_remove:
+            if name not in no_post_removals or not no_post_removals[name]:
                 if daemon_type not in ['iscsi']:
                     service_registry.get_service(daemon_type_to_service(
                         daemon_type)).post_remove(daemon, is_failed_deploy=False)
                 else:
                     self.mgr.scheduled_async_actions.append(lambda: service_registry.get_service(daemon_type_to_service(
                                                             daemon_type)).post_remove(daemon, is_failed_deploy=False))
                     self.mgr._kick_serve_loop()
-
             self.mgr.recently_altered_daemons[name] = datetime_now()
-            return "Removed {} from host '{}'".format(name, host)
+            msg = "Removed {} from host '{}'".format(name, host)
+            self.mgr.log.info(msg)
+            msgs.append(msg)
+
+        self.mgr.cache.invalidate_host_daemons(host)
+
+        return msgs
 
     async def _run_cephadm_json(self,
                                 host: str,

src/pybind/mgr/cephadm/services/osd.py

@@ -892,7 +892,7 @@ def process_removal_queue(self) -> bool:
             assert osd.hostname is not None
 
             if self.mgr.cache.has_daemon(f'osd.{osd.osd_id}'):
-                CephadmServe(self.mgr)._remove_daemon(f'osd.{osd.osd_id}', osd.hostname)
+                self.mgr.wait_async(CephadmServe(self.mgr)._remove_daemon([f'osd.{osd.osd_id}'], osd.hostname))
                 logger.info(f"Successfully removed {osd} on {osd.hostname}")
                 result = True
             else: