Merge PR #43046 into master

* refs/pull/43046/head:
	mgr/rook: get running pods, auth rm, better error checking for orch nfs
	qa/tasks/rook: add apply nfs to rook qa task
	mgr/rook: prevent creation of NFS clusters not in .nfs rados pool
	mgr/rook, mgr/nfs: update rook orchestrator to create and use .nfs pool

Reviewed-by: Juan Miguel Olmo <jolmomar@redhat.com>
Reviewed-by: Varsha Rao <rvarsha016@gmail.com>

qa/suites/orch/rook/smoke/3-final.yaml

@@ -9,3 +9,4 @@ tasks:
       - ceph orch apply rgw foo
       - ceph orch apply mds foo
       - ceph orch apply rbd-mirror
+      - ceph orch apply nfs foo --port 12777

src/pybind/mgr/rook/module.py

@@ -1,5 +1,6 @@
-from logging import error
+import datetime
 import logging
+import re
 import threading
 import functools
 import os
@@ -33,7 +34,7 @@ except ImportError:
     client = None
     config = None
 
-from mgr_module import MgrModule, Option
+from mgr_module import MgrModule, Option, NFS_POOL_NAME
 import orchestrator
 from orchestrator import handle_orch_error, OrchResult, raise_if_exception
 
@@ -348,12 +349,16 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
         if service_type == 'nfs' or service_type is None:
             # CephNFSes
             all_nfs = self.rook_cluster.get_resource("cephnfses")
+            nfs_pods = self.rook_cluster.describe_pods('nfs', None, None)
             for nfs in all_nfs:
+                if nfs['spec']['rados']['pool'] != NFS_POOL_NAME:
+                    continue
                 nfs_name = nfs['metadata']['name']
                 svc = 'nfs.' + nfs_name
                 if svc in spec:
                     continue
                 active = nfs['spec'].get('server', {}).get('active')
+                creation_timestamp = datetime.datetime.strptime(nfs['metadata']['creationTimestamp'], '%Y-%m-%dT%H:%M:%SZ')
                 spec[svc] = orchestrator.ServiceDescription(
                     spec=NFSServiceSpec(
                         service_id=nfs_name,
@@ -361,6 +366,8 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
                     ),
                     size=active,
                     last_refresh=now,
+                    running=len([1 for pod in nfs_pods if pod['labels']['ceph_nfs'] == nfs_name]),
+                    created=creation_timestamp.astimezone(tz=datetime.timezone.utc)
                 )
         if service_type == 'osd' or service_type is None:
             # OSDs
@@ -504,6 +511,15 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
         elif service_type == 'rgw':
             return self.rook_cluster.rm_service('cephobjectstores', service_id)
         elif service_type == 'nfs':
+            ret, out, err = self.mon_command({
+                'prefix': 'auth ls'
+            })
+            matches = re.findall(rf'client\.nfs-ganesha\.{service_id}\..*', out)
+            for match in matches:
+                self.check_mon_command({
+                    'prefix': 'auth rm',
+                    'entity': match
+                })
             return self.rook_cluster.rm_service('cephnfses', service_id)
         elif service_type == 'rbd-mirror':
             return self.rook_cluster.rm_service('cephrbdmirrors', service_id)
@@ -512,6 +528,9 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
                 del self._drive_group_map[service_id]
                 self._save_drive_groups()
             return f'Removed {service_name}'
+        elif service_type == 'ingress':
+            self.log.info("{0} service '{1}' does not exist".format('ingress', service_id))
+            return 'The Rook orchestrator does not currently support ingress'
         else:
             raise orchestrator.OrchestratorError(f'Service type {service_type} not supported')
 
@@ -553,7 +572,11 @@ class RookOrchestrator(MgrModule, orchestrator.Orchestrator):
     @handle_orch_error
     def apply_nfs(self, spec):
         # type: (NFSServiceSpec) -> str
-        return self.rook_cluster.apply_nfsgw(spec)
+        try:
+            return self.rook_cluster.apply_nfsgw(spec, self)
+        except Exception as e:
+            logging.error(e)
+            return "Unable to create NFS daemon, check logs for more traceback\n" + str(e.with_traceback(None))
 
     @handle_orch_error
     def remove_daemons(self, names: List[str]) -> List[str]:

src/pybind/mgr/rook/rook_cluster.py

@@ -26,6 +26,9 @@ from ceph.deployment.drive_group import DriveGroupSpec
 from ceph.deployment.service_spec import ServiceSpec, NFSServiceSpec, RGWSpec, PlacementSpec, HostPlacementSpec
 from ceph.utils import datetime_now
 from ceph.deployment.drive_selection.matchers import SizeMatcher
+from nfs.cluster import create_ganesha_pool
+from nfs.module import Module
+from nfs.export import NFSRados
 from mgr_module import NFS_POOL_NAME
 from mgr_util import merge_dicts
 
@@ -49,7 +52,7 @@ from .rook_client._helper import CrdClass
 import orchestrator
 
 try:
-    from rook.module import RookEnv
+    from rook.module import RookEnv, RookOrchestrator
 except ImportError:
     pass  # just used for type checking.
 
@@ -773,7 +776,7 @@ class RookCluster(object):
                             "osd": ("ceph-osd-id", service_id),
                             "mon": ("mon", service_id),
                             "mgr": ("mgr", service_id),
-                            "ceph_nfs": ("ceph_nfs", service_id),
+                            "nfs": ("nfs", service_id),
                             "rgw": ("ceph_rgw", service_id),
                         }[service_type]
                     except KeyError:
@@ -1027,12 +1030,15 @@ class RookCluster(object):
             cos.CephObjectStore, 'cephobjectstores', name,
             _update_zone, _create_zone)
 
-    def apply_nfsgw(self, spec: NFSServiceSpec) -> str:
+    def apply_nfsgw(self, spec: NFSServiceSpec, mgr: 'RookOrchestrator') -> str:
         # TODO use spec.placement
         # TODO warn if spec.extended has entries we don't kow how
         #      to action.
         # TODO Number of pods should be based on the list of hosts in the
         #      PlacementSpec.
+        assert spec.service_id, "service id in NFS service spec cannot be an empty string or None " # for mypy typing
+        service_id = spec.service_id
+        mgr_module = cast(Module, mgr)
         count = spec.placement.count or 1
         def _update_nfs(new: cnfs.CephNFS) -> cnfs.CephNFS:
             new.spec.server.active = count
@@ -1047,7 +1053,7 @@ class RookCluster(object):
                         ),
                     spec=cnfs.Spec(
                         rados=cnfs.Rados(
-                            namespace=self.rook_env.namespace,
+                            namespace=service_id,
                             pool=NFS_POOL_NAME,
                             ),
                         server=cnfs.Server(
@@ -1056,12 +1062,12 @@ class RookCluster(object):
                         )
                     )
 
-            rook_nfsgw.spec.rados.namespace = cast(str, spec.service_id)
 
             return rook_nfsgw
 
-        assert spec.service_id is not None
-        return self._create_or_patch(cnfs.CephNFS, 'cephnfses', spec.service_id,
+        create_ganesha_pool(mgr)
+        NFSRados(mgr_module, service_id).write_obj('', f'conf-nfs.{spec.service_id}')
+        return self._create_or_patch(cnfs.CephNFS, 'cephnfses', service_id,
                 _update_nfs, _create_nfs)
 
     def rm_service(self, rooktype: str, service_id: str) -> str: