Merge pull request ceph#45211 from phlogistonjohn/jjm-nfs-cleanups3

mgr/nfs: Additional fixes for NFS export management without orchestration

Reviewed-by: Adam King <[email protected]>
Reviewed-by: Michael Fritch <[email protected]>

Author: adk3798

src/pybind/mgr/nfs/export.py

@@ -20,8 +20,7 @@
 from mgr_module import NFS_POOL_NAME as POOL_NAME, NFS_GANESHA_SUPPORTED_FSALS
 
 from .export_utils import GaneshaConfParser, Export, RawBlock, CephFSFSAL, RGWFSAL
-from .exception import NFSException, NFSInvalidOperation, FSNotFound, \
-    ClusterNotFound
+from .exception import NFSException, NFSInvalidOperation, FSNotFound
 from .utils import (
     CONF_PREFIX,
     EXPORT_PREFIX,
@@ -40,6 +39,15 @@
 log = logging.getLogger(__name__)
 
 
+def known_cluster_ids(mgr: 'Module') -> Set[str]:
+    """Return the set of known cluster IDs."""
+    try:
+        clusters = set(available_clusters(mgr))
+    except NoOrchestrator:
+        clusters = nfs_rados_configs(mgr.rados)
+    return clusters
+
+
 def export_cluster_checker(func: FuncT) -> FuncT:
     def cluster_check(
             export: 'ExportMgr',
@@ -49,10 +57,7 @@ def cluster_check(
         """
         This method checks if cluster exists
         """
-        try:
-            clusters = set(available_clusters(export.mgr))
-        except NoOrchestrator:
-            clusters = nfs_rados_configs(export.mgr.rados)
+        clusters = known_cluster_ids(export.mgr)
         cluster_id: str = kwargs['cluster_id']
         log.debug("checking for %r in known nfs clusters: %r",
                   cluster_id, clusters)
@@ -187,10 +192,10 @@ def exports(self) -> Dict[str, List[Export]]:
         if self._exports is None:
             self._exports = {}
             log.info("Begin export parsing")
-            for cluster_id in available_clusters(self.mgr):
+            for cluster_id in known_cluster_ids(self.mgr):
                 self.export_conf_objs = []  # type: List[Export]
                 self._read_raw_config(cluster_id)
-                self.exports[cluster_id] = self.export_conf_objs
+                self._exports[cluster_id] = self.export_conf_objs
                 log.info("Exports parsed successfully %s", self.exports.items())
         return self._exports
 
@@ -710,8 +715,6 @@ def _apply_export(
         for k in ['path', 'pseudo']:
             if k not in new_export_dict:
                 raise NFSInvalidOperation(f'Export missing required field {k}')
-        if cluster_id not in available_clusters(self.mgr):
-            raise ClusterNotFound()
         if cluster_id not in self.exports:
             self.exports[cluster_id] = []