mgr/cephadm: validating TunedProfileSpec placement

fixes: https://tracker.ceph.com/issues/57192

Signed-off-by: Redouane Kachach <rkachach@redhat.com>

src/pybind/mgr/cephadm/module.py

@@ -2535,10 +2535,43 @@ def _apply(self, spec: GenericSpec) -> str:
 
         return self._apply_service_spec(cast(ServiceSpec, spec))
 
+    def _get_candidate_hosts(self, placement: PlacementSpec) -> List[str]:
+        """Return a list of candidate hosts according to the placement specification."""
+        all_hosts = self.cache.get_schedulable_hosts()
+        draining_hosts = [dh.hostname for dh in self.cache.get_draining_hosts()]
+        candidates = []
+        if placement.hosts:
+            candidates = [h.hostname for h in placement.hosts if h.hostname in placement.hosts]
+        elif placement.label:
+            candidates = [x.hostname for x in [h for h in all_hosts if placement.label in h.labels]]
+        elif placement.host_pattern:
+            candidates = [x for x in placement.filter_matching_hostspecs(all_hosts)]
+        elif (placement.count is not None or placement.count_per_host is not None):
+            candidates = [x.hostname for x in all_hosts]
+        return [h for h in candidates if h not in draining_hosts]
+
+    def _validate_one_shot_placement_spec(self, spec: PlacementSpec) -> None:
+        """Validate placement specification for TunedProfileSpec and ClientKeyringSpec."""
+        if spec.count is not None:
+            raise OrchestratorError("Placement 'count' field is no supported for this specification.")
+        if spec.count_per_host is not None:
+            raise OrchestratorError("Placement 'count_per_host' field is no supported for this specification.")
+        if spec.hosts:
+            all_hosts = [h.hostname for h in self.inventory.all_specs()]
+            invalid_hosts = [h.hostname for h in spec.hosts if h.hostname not in all_hosts]
+            if invalid_hosts:
+                raise OrchestratorError(f"Found invalid host(s) in placement section: {invalid_hosts}. "
+                                        f"Please check 'ceph orch host ls' for available hosts.")
+        elif not self._get_candidate_hosts(spec):
+            raise OrchestratorError("Invalid placement specification (no candidate hosts). "
+                                    "Please check 'ceph orch host ls' for available hosts."
+                                    "Note: draining hots are excluded from the candidate list.")
+
     @handle_orch_error
     def apply_tuned_profiles(self, specs: List[TunedProfileSpec]) -> str:
         outs = []
         for spec in specs:
+            self._validate_one_shot_placement_spec(spec.placement)
             self.tuned_profiles.add_profile(spec)
             outs.append(f'Saved tuned profile {spec.profile_name}')
         self._kick_serve_loop()

src/python-common/ceph/deployment/hostspec.py

@@ -128,6 +128,8 @@ def __str__(self) -> str:
         return self.hostname
 
     def __eq__(self, other: Any) -> bool:
+        if not isinstance(other, HostSpec):
+            return NotImplemented
         # Let's omit `status` for the moment, as it is still the very same host.
         return self.hostname == other.hostname and \
                self.addr == other.addr and \