Merge pull request ceph#59726 from avanthakkar/earmark-nfs

mgr/nfs: earmark resolver for subvolume

Reviewed-by: Adam King <[email protected]>
Reviewed-by: John Mulligan <[email protected]>

src/pybind/mgr/nfs/export.py

@@ -13,8 +13,10 @@
     Set,
     cast)
 from os.path import normpath
+from ceph.fs.earmarking import EarmarkTopScope
 import cephfs
 
+from mgr_util import CephFSEarmarkResolver
 from rados import TimedOut, ObjectNotFound, Rados
 
 from object_format import ErrorResponse
@@ -535,7 +537,8 @@ def get_export_by_pseudo(
 
     # This method is used by the dashboard module (../dashboard/controllers/nfs.py)
     # Do not change interface without updating the Dashboard code
-    def apply_export(self, cluster_id: str, export_config: str) -> AppliedExportResults:
+    def apply_export(self, cluster_id: str, export_config: str,
+                     earmark_resolver: Optional[CephFSEarmarkResolver] = None) -> AppliedExportResults:
         try:
             exports = self._read_export_config(cluster_id, export_config)
         except Exception as e:
@@ -544,7 +547,7 @@ def apply_export(self, cluster_id: str, export_config: str) -> AppliedExportResu
 
         aeresults = AppliedExportResults()
         for export in exports:
-            changed_export = self._change_export(cluster_id, export)
+            changed_export = self._change_export(cluster_id, export, earmark_resolver)
             # This will help figure out which export blocks in conf/json file
             # are problematic.
             if changed_export.get("state", "") == "error":
@@ -573,9 +576,10 @@ def _read_export_config(self, cluster_id: str, export_config: str) -> List[Dict]
             return j  # j is already a list object
         return [j]  # return a single object list, with j as the only item
 
-    def _change_export(self, cluster_id: str, export: Dict) -> Dict[str, Any]:
+    def _change_export(self, cluster_id: str, export: Dict,
+                       earmark_resolver: Optional[CephFSEarmarkResolver] = None) -> Dict[str, Any]:
         try:
-            return self._apply_export(cluster_id, export)
+            return self._apply_export(cluster_id, export, earmark_resolver)
         except NotImplementedError:
             # in theory, the NotImplementedError here may be raised by a hook back to
             # an orchestration module. If the orchestration module supports it the NFS
@@ -651,10 +655,34 @@ def _create_user_key(self, cluster_id: str, entity: str, path: str, fs_name: str
         log.info(f"Export user created is {json_res[0]['entity']}")
         return json_res[0]['key']
 
+    def _check_earmark(self, earmark_resolver: CephFSEarmarkResolver, path: str,
+                       fs_name: str) -> None:
+        earmark = earmark_resolver.get_earmark(
+            path,
+            fs_name,
+        )
+        if not earmark:
+            earmark_resolver.set_earmark(
+                path,
+                fs_name,
+                EarmarkTopScope.NFS.value,
+            )
+        else:
+            if not earmark_resolver.check_earmark(
+                earmark, EarmarkTopScope.NFS
+            ):
+                raise NFSException(
+                    'earmark has already been set by ' + earmark.split('.')[0],
+                    -errno.EAGAIN
+                )
+        return None
+
     def create_export_from_dict(self,
                                 cluster_id: str,
                                 ex_id: int,
-                                ex_dict: Dict[str, Any]) -> Export:
+                                ex_dict: Dict[str, Any],
+                                earmark_resolver: Optional[CephFSEarmarkResolver] = None
+                                ) -> Export:
         pseudo_path = ex_dict.get("pseudo")
         if not pseudo_path:
             raise NFSInvalidOperation("export must specify pseudo path")
@@ -677,6 +705,11 @@ def create_export_from_dict(self,
                 raise FSNotFound(fs_name)
 
             validate_cephfs_path(self.mgr, fs_name, path)
+
+            # Check if earmark is set for the path, given path is of subvolume
+            if earmark_resolver:
+                self._check_earmark(earmark_resolver, path, fs_name)
+
             if fsal["cmount_path"] != "/":
                 _validate_cmount_path(fsal["cmount_path"], path)  # type: ignore
 
@@ -707,7 +740,9 @@ def create_cephfs_export(self,
                              access_type: str,
                              clients: list = [],
                              sectype: Optional[List[str]] = None,
-                             cmount_path: Optional[str] = "/") -> Dict[str, Any]:
+                             cmount_path: Optional[str] = "/",
+                             earmark_resolver: Optional[CephFSEarmarkResolver] = None
+                             ) -> Dict[str, Any]:
 
         validate_cephfs_path(self.mgr, fs_name, path)
         if cmount_path != "/":
@@ -731,7 +766,8 @@ def create_cephfs_export(self,
                     },
                     "clients": clients,
                     "sectype": sectype,
-                }
+                },
+                earmark_resolver
             )
             log.debug("creating cephfs export %s", export)
             self._ensure_cephfs_export_user(export)
@@ -795,6 +831,7 @@ def _apply_export(
             self,
             cluster_id: str,
             new_export_dict: Dict,
+            earmark_resolver: Optional[CephFSEarmarkResolver] = None
     ) -> Dict[str, str]:
         for k in ['path', 'pseudo']:
             if k not in new_export_dict:
@@ -834,7 +871,8 @@ def _apply_export(
         new_export = self.create_export_from_dict(
             cluster_id,
             new_export_dict.get('export_id', self._gen_export_id(cluster_id)),
-            new_export_dict
+            new_export_dict,
+            earmark_resolver
         )
 
         if not old_export:

src/pybind/mgr/nfs/module.py

@@ -6,6 +6,7 @@
 import object_format
 import orchestrator
 from orchestrator.module import IngressType
+from mgr_util import CephFSEarmarkResolver
 
 from .export import ExportMgr, AppliedExportResults
 from .cluster import NFSCluster
@@ -41,6 +42,7 @@ def _cmd_nfs_export_create_cephfs(
             cmount_path: Optional[str] = "/"
     ) -> Dict[str, Any]:
         """Create a CephFS export"""
+        earmark_resolver = CephFSEarmarkResolver(self)
         return self.export_mgr.create_export(
             fsal_type='cephfs',
             fs_name=fsname,
@@ -51,7 +53,8 @@ def _cmd_nfs_export_create_cephfs(
             squash=squash,
             addr=client_addr,
             sectype=sectype,
-            cmount_path=cmount_path
+            cmount_path=cmount_path,
+            earmark_resolver=earmark_resolver
         )
 
     @CLICommand('nfs export create rgw', perm='rw')
@@ -114,8 +117,10 @@ def _cmd_nfs_export_get(self, cluster_id: str, pseudo_path: str) -> Dict[str, An
     @CLICheckNonemptyFileInput(desc='Export JSON or Ganesha EXPORT specification')
     @object_format.Responder()
     def _cmd_nfs_export_apply(self, cluster_id: str, inbuf: str) -> AppliedExportResults:
+        earmark_resolver = CephFSEarmarkResolver(self)
         """Create or update an export by `-i <json_or_ganesha_export_file>`"""
-        return self.export_mgr.apply_export(cluster_id, export_config=inbuf)
+        return self.export_mgr.apply_export(cluster_id, export_config=inbuf,
+                                            earmark_resolver=earmark_resolver)
 
     @CLICommand('nfs cluster create', perm='rw')
     @object_format.EmptyResponder()

src/python-common/ceph/fs/earmarking.py

@@ -61,11 +61,16 @@ def _handle_cephfs_error(self, e: Exception, action: str) -> None:
         if isinstance(e, ValueError):
             raise EarmarkException(errno.EINVAL, f"Invalid earmark specified: {e}") from e
         elif isinstance(e, OSError):
-            log.error(f"Error {action} earmark: {e}")
-            raise EarmarkException(-e.errno, e.strerror) from e
+            if e.errno == errno.ENODATA:
+                # Return empty string when earmark is not set
+                log.info(f"No earmark set for the path while {action}. Returning empty result.")
+                return ''
+            else:
+                log.error(f"Error {action} earmark: {e}")
+                raise EarmarkException(-e.errno, e.strerror) from e
         else:
             log.error(f"Unexpected error {action} earmark: {e}")
-            raise EarmarkException
+            raise EarmarkException(errno.EFAULT, f"Unexpected error {action} earmark: {e}") from e
 
     @staticmethod
     def parse_earmark(value: str) -> Optional[EarmarkContents]:
@@ -128,8 +133,7 @@ def get_earmark(self) -> Optional[str]:
             )
             return earmark_value
         except Exception as e:
-            self._handle_cephfs_error(e, "getting")
-            return None
+            return self._handle_cephfs_error(e, "getting")
 
     def set_earmark(self, earmark: str):
         # Validate the earmark before attempting to set it