mgr/cephadm: move service wide config handling out of host parallelized section

In these portions we can have daemons of different services
which can cause us to mark the wrong spec for configuration.

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

Resolves: rhbz#2370541

Author: adk3798

src/pybind/mgr/cephadm/serve.py

@@ -683,7 +683,6 @@ async def _parallel_deploy_and_remove(
             removed_conflict_daemons, conflict_hosts_altered = self.remove_given_daemons(conflicts)
             if removed_conflict_daemons:
                 r = True
-                self.mgr.spec_store.mark_needs_configuration(spec.service_name())
                 hosts_altered.update(conflict_hosts_altered)
             daemons_placed, daemon_deployed_hosts, daemon_place_fails = await self.deploy_given_daemons(to_deploy)
             if daemons_placed:
@@ -692,12 +691,7 @@ async def _parallel_deploy_and_remove(
             removed_daemons, removed_daemon_hosts = self.remove_given_daemons(to_remove)
             if removed_daemons:
                 r = True
-                self.mgr.spec_store.mark_needs_configuration(spec.service_name())
                 hosts_altered.update(conflict_hosts_altered)
-            if self.mgr.spec_store.needs_configuration(spec.service_name()):
-                svc = service_registry.get_service(spec.service_type)
-                svc.config(spec)
-                self.mgr.spec_store.mark_configured(spec.service_name())
             return (r, hosts_altered, daemon_place_fails)
 
         async def _deploy_and_remove_all(
@@ -734,6 +728,11 @@ async def _deploy_and_remove_all(
         if placement_failures:
             self.mgr.set_health_warning('CEPHADM_DAEMON_PLACE_FAIL', f'Failed to place {len(placement_failures)} daemon(s)', len(
                 placement_failures), placement_failures)
+        for spec in specs:
+            if self.mgr.spec_store.needs_configuration(spec.service_name()):
+                svc = service_registry.get_service(spec.service_type)
+                svc.config(spec)
+                self.mgr.spec_store.mark_configured(spec.service_name())
         if self.mgr.use_agent:
             # can only send ack to agents if we know for sure port they bound to
             altered_hosts = set([h for h in altered_hosts if (h in self.mgr.agent_cache.agent_ports and not self.mgr.cache.is_host_draining(h))])
@@ -971,6 +970,9 @@ def has_interface_for_vip(host: str, sspec: ServiceSpec) -> bool:
                                         warnings)
             raise OrchestratorError(msg)
 
+        if slots_to_add or daemons_to_remove:
+            self.mgr.spec_store.mark_needs_configuration(spec.service_name())
+
         return slots_to_add, daemons_to_remove, rank_map
 
     def gather_conflicting_daemons_for_service(self, spec: ServiceSpec) -> List[orchestrator.DaemonDescription]: