Merge PR #42998 into master

* refs/pull/42998/head:
	qa/tasks/rook: add testing for host label add/rm
	mgr/rook: host add/rm label in rook orchestrator

Reviewed-by: Juan Miguel Olmo <jolmomar@redhat.com>

qa/suites/orch/rook/smoke/2-workload/radosbench.yaml

@@ -3,3 +3,10 @@ tasks:
     host.a:
 - radosbench:
     clients: [client.a]
+- rook.shell:
+    commands:
+        - |
+          ceph orch host label add `hostname` foo
+          ceph orch host ls | grep foo
+          ceph orch host label rm `hostname` foo
+          ceph orch host ls | grep -v foo

src/pybind/mgr/rook/module.py

@@ -222,7 +222,7 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
     @handle_orch_error
     def get_hosts(self):
         # type: () -> List[orchestrator.HostSpec]
-        return [orchestrator.HostSpec(n) for n in self.rook_cluster.get_node_names()]
+        return self.rook_cluster.get_hosts()
 
     @handle_orch_error
     def describe_service(self,
@@ -484,6 +484,11 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
         res = self._rook_cluster.remove_osds(osd_ids, replace, force, self.mon_command)
         return OrchResult(res)
 
+    def add_host_label(self, host: str, label: str) -> OrchResult[str]:
+        return self.rook_cluster.add_host_label(host, label)
+    
+    def remove_host_label(self, host: str, label: str) -> OrchResult[str]:
+        return self.rook_cluster.remove_host_label(host, label)
     """
     @handle_orch_error
     def create_osds(self, drive_group):

src/pybind/mgr/rook/rook_cluster.py

@@ -12,6 +12,7 @@ import logging
 from contextlib import contextmanager
 from time import sleep
 import re
+from orchestrator import OrchResult
 
 import jsonpatch
 from urllib.parse import urljoin
@@ -870,6 +871,29 @@ class RookCluster(object):
             cfs.CephFilesystem, 'cephfilesystems', spec.service_id,
             _update_fs, _create_fs)
 
+    def get_matching_node(self, host: str) -> Any:
+        matching_node = None
+        for node in self.nodes.items:
+            if node.metadata.labels['kubernetes.io/hostname'] == host:
+                matching_node = node
+        return matching_node
+
+    def add_host_label(self, host: str, label: str) -> OrchResult[str]:
+        matching_node = self.get_matching_node(host)
+        if matching_node == None:
+            return OrchResult(None, RuntimeError(f"Cannot add {label} label to {host}: host not found in cluster")) 
+        matching_node.metadata.labels['ceph-label/'+ label] = ""
+        self.coreV1_api.patch_node(host, matching_node)
+        return OrchResult(f'Added {label} label to {host}')
+
+    def remove_host_label(self, host: str, label: str) -> OrchResult[str]:
+        matching_node = self.get_matching_node(host)
+        if matching_node == None:
+            return OrchResult(None, RuntimeError(f"Cannot remove {label} label from {host}: host not found in cluster")) 
+        matching_node.metadata.labels.pop('ceph-label/' + label, None)
+        self.coreV1_api.patch_node(host, matching_node)
+        return OrchResult(f'Removed {label} label from {host}')
+
     def apply_objectstore(self, spec: RGWSpec) -> str:
         assert spec.service_id is not None
 
@@ -1040,6 +1064,17 @@ class RookCluster(object):
         )
         return self.remover.remove()
 
+    def get_hosts(self) -> List[orchestrator.HostSpec]:
+        ret = []
+        for node in self.nodes.items:
+            spec = orchestrator.HostSpec(
+                node.metadata.name, 
+                addr='/'.join([addr.address for addr in node.status.addresses]), 
+                labels=[label.split('/')[1] for label in node.metadata.labels if label.startswith('ceph-label')],
+            )
+            ret.append(spec)
+        return ret
+
     def _patch(self, crd: Type, crd_name: str, cr_name: str, func: Callable[[CrdClassT, CrdClassT], CrdClassT]) -> str:
         current_json = self.rook_api_get(
             "{}/{}".format(crd_name, cr_name)