mgr/smb: earmark resolver for subvolume

Signed-off-by: Avan Thakkar <[email protected]>

Author: avanthakkar

src/pybind/mgr/mgr_util.py

@@ -1,5 +1,12 @@
 import os
 
+from ceph.fs.earmarking import (
+    CephFSVolumeEarmarking,
+    EarmarkParseError,
+    EarmarkTopScope,
+    EarmarkException
+)
+
 if 'UNITTEST' in os.environ:
     import tests  # noqa
 
@@ -335,6 +342,93 @@ def get_all_filesystems(self) -> List[str]:
         return fs_list
 
 
+class CephFSEarmarkResolver:
+    def __init__(self, mgr: Module_T, *, client: Optional[CephfsClient] = None) -> None:
+        self._mgr = mgr
+        self._cephfs_client = client or CephfsClient(mgr)
+
+    def _extract_path_component(self, path: str, index: int) -> Optional[str]:
+        """
+        Extracts a specific component from the path based on the given index.
+
+        :param path: The path in the format '/volumes/{subvolumegroup}/{subvolume}/..'
+        :param index: The index of the component to extract (1 for subvolumegroup, 2 for subvolume)
+        :return: The component at the specified index
+        """
+        parts = path.strip('/').split('/')
+        if len(parts) >= 3 and parts[0] == "volumes":
+            return parts[index]
+        return None
+
+    def _fetch_subvolumegroup_from_path(self, path: str) -> Optional[str]:
+        """
+        Extracts and returns the subvolume group name from the given path.
+
+        :param path: The path in the format '/volumes/{subvolumegroup}/{subvolume}/..'
+        :return: The subvolume group name
+        """
+        return self._extract_path_component(path, 1)
+
+    def _fetch_subvolume_from_path(self, path: str) -> Optional[str]:
+        """
+        Extracts and returns the subvolume name from the given path.
+
+        :param path: The path in the format '/volumes/{subvolumegroup}/{subvolume}/..'
+        :return: The subvolume name
+        """
+        return self._extract_path_component(path, 2)
+
+    def _manage_earmark(self, path: str, volume: str, operation: str, earmark: Optional[str] = None) -> Optional[str]:
+        """
+        Manages (get or set) the earmark for a subvolume based on the provided parameters.
+
+        :param path: The path of the subvolume
+        :param volume: The volume name
+        :param earmark: The earmark to set (None if only getting the earmark)
+        :return: The earmark if getting, otherwise None
+        """
+        with open_filesystem(self._cephfs_client, volume) as fs:
+            earmark_manager = CephFSVolumeEarmarking(fs, path)
+            try:
+                if operation == 'set' and earmark is not None:
+                    earmark_manager.set_earmark(earmark)
+                    return None
+                elif operation == 'get':
+                    return earmark_manager.get_earmark()
+            except EarmarkException as e:
+                logger.error(f"Failed to manage earmark: {e}")
+                return None
+        return None
+
+    def get_earmark(self, path: str, volume: str) -> Optional[str]:
+        """
+        Get earmark for a subvolume.
+        """
+        return self._manage_earmark(path, volume, 'get')
+
+    def set_earmark(self, path: str, volume: str, earmark: str) -> None:
+        """
+        Set earmark for a subvolume.
+        """
+        self._manage_earmark(path, volume, 'set', earmark)
+
+    def check_earmark(self, earmark: str, top_level_scope: EarmarkTopScope) -> bool:
+        """
+        Check if the earmark belongs to the mentioned top level scope.
+
+        :param earmark: The earmark string to check.
+        :param top_level_scope: The expected top level scope.
+        :return: True if the earmark matches the top level scope, False otherwise.
+        """
+        try:
+            parsed = CephFSVolumeEarmarking.parse_earmark(earmark)
+            if parsed is None:
+                return False
+            return parsed.top == top_level_scope
+        except EarmarkParseError:
+            return False
+
+
 @contextlib.contextmanager
 def open_filesystem(fsc: CephfsClient, fs_name: str) -> Generator["cephfs.LibCephFS", None, None]:
     """

src/pybind/mgr/smb/handler.py

@@ -18,6 +18,7 @@
 import time
 
 from ceph.deployment.service_spec import SMBSpec
+from ceph.fs.earmarking import EarmarkTopScope
 
 from . import config_store, external, resources
 from .enums import (
@@ -43,6 +44,7 @@
     AccessAuthorizer,
     ConfigEntry,
     ConfigStore,
+    EarmarkResolver,
     EntryKey,
     OrchSubmitter,
     PathResolver,
@@ -112,6 +114,22 @@ def resolve(
     resolve_exists = resolve
 
 
+class _FakeEarmarkResolver:
+    """A stub EarmarkResolver for unit testing."""
+
+    def __init__(self) -> None:
+        self._earmarks: Dict[Tuple[str, str], str] = {}
+
+    def get_earmark(self, path: str, volume: str) -> Optional[str]:
+        return None
+
+    def set_earmark(self, path: str, volume: str, earmark: str) -> None:
+        pass
+
+    def check_earmark(self, earmark: str, top_level_scope: str) -> bool:
+        return True
+
+
 class _FakeAuthorizer:
     """A stub AccessAuthorizer for unit testing."""
 
@@ -325,6 +343,7 @@ def __init__(
         path_resolver: Optional[PathResolver] = None,
         authorizer: Optional[AccessAuthorizer] = None,
         orch: Optional[OrchSubmitter] = None,
+        earmark_resolver: Optional[EarmarkResolver] = None,
     ) -> None:
         self.internal_store = internal_store
         self.public_store = public_store
@@ -336,14 +355,18 @@ def __init__(
             authorizer = _FakeAuthorizer()
         self._authorizer: AccessAuthorizer = authorizer
         self._orch = orch  # if None, disables updating the spec via orch
+        if earmark_resolver is None:
+            earmark_resolver = cast(EarmarkResolver, _FakeEarmarkResolver())
+        self._earmark_resolver = earmark_resolver
         log.info(
             'Initialized new ClusterConfigHandler with'
             f' internal store {self.internal_store!r},'
             f' public store {self.public_store!r},'
             f' priv store {self.priv_store!r},'
             f' path resolver {self._path_resolver!r},'
             f' authorizer {self._authorizer!r},'
-            f' orch {self._orch!r}'
+            f' orch {self._orch!r},'
+            f' earmark resolver {self._earmark_resolver!r}'
         )
 
     def apply(
@@ -474,7 +497,12 @@ def _check(
             elif isinstance(
                 resource, (resources.Share, resources.RemovedShare)
             ):
-                _check_share(resource, staging, self._path_resolver)
+                _check_share(
+                    resource,
+                    staging,
+                    self._path_resolver,
+                    self._earmark_resolver,
+                )
             elif isinstance(resource, resources.JoinAuth):
                 _check_join_auths(resource, staging)
             elif isinstance(resource, resources.UsersAndGroups):
@@ -807,7 +835,10 @@ def _check_cluster(cluster: ClusterRef, staging: _Staging) -> None:
 
 
 def _check_share(
-    share: ShareRef, staging: _Staging, resolver: PathResolver
+    share: ShareRef,
+    staging: _Staging,
+    resolver: PathResolver,
+    earmark_resolver: EarmarkResolver,
 ) -> None:
     """Check that the share resource can be updated."""
     if share.intent == Intent.REMOVED:
@@ -822,7 +853,7 @@ def _check_share(
         )
     assert share.cephfs is not None
     try:
-        resolver.resolve_exists(
+        volpath = resolver.resolve_exists(
             share.cephfs.volume,
             share.cephfs.subvolumegroup,
             share.cephfs.subvolume,
@@ -832,6 +863,34 @@ def _check_share(
         raise ErrorResult(
             share, msg="path is not a valid directory in volume"
         )
+    if earmark_resolver:
+        earmark = earmark_resolver.get_earmark(
+            volpath,
+            share.cephfs.volume,
+        )
+        if not earmark:
+            smb_earmark = (
+                f"{EarmarkTopScope.SMB.value}.cluster.{share.cluster_id}"
+            )
+            earmark_resolver.set_earmark(
+                volpath,
+                share.cephfs.volume,
+                smb_earmark,
+            )
+        else:
+            if not earmark_resolver.check_earmark(
+                earmark, EarmarkTopScope.SMB
+            ):
+                raise ErrorResult(
+                    share,
+                    msg=f"earmark has already been set by {earmark.split('.')[0]}",
+                )
+            # Check if earmark is set by same cluster
+            if earmark.split('.')[2] != share.cluster_id:
+                raise ErrorResult(
+                    share,
+                    msg=f"earmark has already been set by smb cluster {earmark.split('.')[2]}",
+                )
     name_used_by = _share_name_in_use(staging, share)
     if name_used_by:
         raise ErrorResult(

src/pybind/mgr/smb/module.py

@@ -5,6 +5,7 @@
 import orchestrator
 from ceph.deployment.service_spec import PlacementSpec, SMBSpec
 from mgr_module import MgrModule, Option, OptionLevel
+from mgr_util import CephFSEarmarkResolver
 
 from . import (
     cli,
@@ -55,6 +56,7 @@ def __init__(self, *args: str, **kwargs: Any) -> None:
         path_resolver = kwargs.pop('path_resolver', None)
         authorizer = kwargs.pop('authorizer', None)
         uo = kwargs.pop('update_orchestration', None)
+        earmark_resolver = kwargs.pop('earmark_resolver', None)
         super().__init__(*args, **kwargs)
         if internal_store is not None:
             self._internal_store = internal_store
@@ -69,6 +71,7 @@ def __init__(self, *args: str, **kwargs: Any) -> None:
             public_store or rados_store.RADOSConfigStore.init(self)
         )
         path_resolver = path_resolver or fs.CachingCephFSPathResolver(self)
+        earmark_resolver = earmark_resolver or CephFSEarmarkResolver(self)
         # Why the honk is the cast needed but path_resolver doesn't need it??
         # Sometimes mypy drives me batty.
         authorizer = cast(
@@ -81,6 +84,7 @@ def __init__(self, *args: str, **kwargs: Any) -> None:
             path_resolver=path_resolver,
             authorizer=authorizer,
             orch=self._orch_backend(enable_orch=uo),
+            earmark_resolver=earmark_resolver,
         )
 
     def _backend_store(self, store_conf: str = '') -> ConfigStore:

src/pybind/mgr/smb/proto.py

@@ -15,6 +15,7 @@
 import sys
 
 from ceph.deployment.service_spec import SMBSpec
+from ceph.fs.earmarking import EarmarkTopScope
 
 # this uses a version check as opposed to a try/except because this
 # form makes mypy happy and try/except doesn't.
@@ -185,3 +186,18 @@ def authorize_entity(
         self, volume: str, entity: str, caps: str = ''
     ) -> None:
         ...  # pragma: no cover
+
+
+class EarmarkResolver(Protocol):
+    """A protocol for a type that can resolve earmarks for subvolumes."""
+
+    def get_earmark(self, path: str, volume: str) -> Optional[str]:
+        ...  # pragma: no cover
+
+    def set_earmark(self, path: str, volume: str, earmark: str) -> None:
+        ...  # pragma: no cover
+
+    def check_earmark(
+        self, earmark: str, top_level_scope: EarmarkTopScope
+    ) -> bool:
+        ...  # pragma: no cover

src/pybind/mgr/smb/tests/test_handler.py

@@ -1,6 +1,7 @@
 import pytest
 
 import smb
+from smb.handler import _FakeEarmarkResolver
 
 
 def _cluster(**kwargs):
@@ -880,6 +881,7 @@ def remove_smb_service(self, service_name):
             self.deployed.remove(service_name)
 
     thandler._orch = FakeOrch()
+    thandler._earmark_resolver = _FakeEarmarkResolver()
     test_apply_full_cluster_create(thandler)
 
     to_apply = [

src/pybind/mgr/smb/tests/test_smb.py

@@ -26,6 +26,7 @@ def tmodule():
         path_resolver=smb.handler._FakePathResolver(),
         authorizer=smb.handler._FakeAuthorizer(),
         update_orchestration=False,
+        earmark_resolver=smb.handler._FakeEarmarkResolver(),
     )