Merge PR ceph#59111 into main

* refs/pull/59111/head:
	doc: document earmark option for subvolume and new commands
	qa/cephfs: update tests for test_volumes & unit-test for earmarking
	mgr/volumes: add earmarking for subvol

Reviewed-by: Venky Shankar <[email protected]>
Reviewed-by: Kotresh Hiremath Ravishankar <[email protected]>
Reviewed-by: John Mulligan <[email protected]>

Author: vshankar

doc/cephfs/fs-volumes.rst

@@ -276,7 +276,7 @@ Use a command of the following form to create a subvolume:
 
 .. prompt:: bash #
 
-   ceph fs subvolume create <vol_name> <subvol_name> [--size <size_in_bytes>] [--group_name <subvol_group_name>] [--pool_layout <data_pool_name>] [--uid <uid>] [--gid <gid>] [--mode <octal_mode>] [--namespace-isolated]
+   ceph fs subvolume create <vol_name> <subvol_name> [--size <size_in_bytes>] [--group_name <subvol_group_name>] [--pool_layout <data_pool_name>] [--uid <uid>] [--gid <gid>] [--mode <octal_mode>] [--namespace-isolated] [--earmark <earmark>]
 
 
 The command succeeds even if the subvolume already exists.
@@ -289,6 +289,15 @@ The subvolume can be created in a separate RADOS namespace by specifying the
 default subvolume group with an octal file mode of ``755``, a uid of its
 subvolume group, a gid of its subvolume group, a data pool layout of its parent
 directory, and no size limit.
+You can also assign an earmark to a subvolume using the ``--earmark`` option.
+The earmark is a unique identifier that tags the subvolume for specific purposes,
+such as NFS or SMB services. By default, no earmark is set, allowing for flexible
+assignment based on administrative needs. An empty string ("") can be used to remove
+any existing earmark from a subvolume.
+
+The earmarking mechanism ensures that subvolumes are correctly tagged and managed,
+helping to avoid conflicts and ensuring that each subvolume is associated
+with the intended service or use case.
 
 Removing a subvolume
 ~~~~~~~~~~~~~~~~~~~~
@@ -418,6 +427,7 @@ The output format is JSON and contains the following fields.
 * ``pool_namespace``: RADOS namespace of the subvolume
 * ``features``: features supported by the subvolume
 * ``state``: current state of the subvolume
+* ``earmark``: earmark of the subvolume
 
 If a subvolume has been removed but its snapshots have been retained, the
 output contains only the following fields.
@@ -522,6 +532,33 @@ subvolume using the metadata key:
 Using the ``--force`` flag allows the command to succeed when it would
 otherwise fail (if the metadata key did not exist).
 
+Getting earmark of a subvolume
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Use a command of the following form to get the earmark of a subvolume:
+
+.. prompt:: bash #
+
+   ceph fs subvolume earmark get <vol_name> <subvol_name> [--group_name <subvol_group_name>]
+
+Setting earmark of a subvolume
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Use a command of the following form to set the earmark of a subvolume:
+
+.. prompt:: bash #
+
+   ceph fs subvolume earmark set <vol_name> <subvol_name> [--group_name <subvol_group_name>] <earmark>
+
+Removing earmark of a subvolume
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Use a command of the following form to remove the earmark of a subvolume:
+
+.. prompt:: bash #
+
+   ceph fs subvolume earmark rm <vol_name> <subvol_name> [--group_name <subvol_group_name>]
+
 Creating a Snapshot of a Subvolume
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

qa/tasks/cephfs/test_volumes.py

@@ -2367,6 +2367,124 @@ def test_subvolume_create_and_ls_providing_group_as_nogroup(self):
 
         # verify trash dir is clean.
         self._wait_for_trash_empty()
+    
+    def test_subvolume_create_with_earmark(self):
+        # create subvolume with earmark
+        subvolume = self._gen_subvol_name()
+        earmark = "nfs.test"
+        self._fs_cmd("subvolume", "create", self.volname, subvolume, "--earmark", earmark)
+
+        # make sure it exists
+        subvolpath = self._get_subvolume_path(self.volname, subvolume)
+        self.assertNotEqual(subvolpath, None)
+
+        # verify the earmark
+        get_earmark = self._fs_cmd("subvolume", "earmark", "get", self.volname, subvolume)
+        self.assertEqual(get_earmark.rstrip('\n'), earmark)
+    
+    def test_subvolume_set_and_get_earmark(self):
+        # create subvolume
+        subvolume = self._gen_subvol_name()
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+
+        # set earmark
+        earmark = "smb.test"
+        self._fs_cmd("subvolume", "earmark", "set", self.volname, subvolume, "--earmark", earmark)
+
+        # get earmark
+        get_earmark = self._fs_cmd("subvolume", "earmark", "get", self.volname, subvolume)
+        self.assertEqual(get_earmark.rstrip('\n'), earmark)
+    
+    def test_subvolume_clear_earmark(self):
+        # create subvolume
+        subvolume = self._gen_subvol_name()
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+
+        # set earmark
+        earmark = "smb.test"
+        self._fs_cmd("subvolume", "earmark", "set", self.volname, subvolume, "--earmark", earmark)
+
+        # remove earmark
+        self._fs_cmd("subvolume", "earmark", "rm", self.volname, subvolume)
+
+        # get earmark
+        get_earmark = self._fs_cmd("subvolume", "earmark", "get", self.volname, subvolume)
+        self.assertEqual(get_earmark, "")
+
+    def test_earmark_on_non_existing_subvolume(self):
+        subvolume = "non_existing_subvol"
+        earmark = "nfs.test"
+        commands = [
+            ("set", earmark),
+            ("get", None),
+            ("rm", None),
+        ]
+
+        for action, arg in commands:
+            try:
+                # Build the command arguments
+                cmd_args = ["subvolume", "earmark", action, self.volname, subvolume]
+                if arg is not None:
+                    cmd_args.extend(["--earmark", arg])
+
+                # Execute the command with built arguments
+                self._fs_cmd(*cmd_args)
+            except CommandFailedError as ce:
+                self.assertEqual(ce.exitstatus, errno.ENOENT)
+
+    def test_get_remove_earmark_when_not_set(self):
+        # Create a subvolume without setting an earmark
+        subvolume = self._gen_subvol_name()
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+
+        # Attempt to get an earmark when it's not set
+        get_earmark = self._fs_cmd("subvolume", "earmark", "get", self.volname, subvolume)
+        self.assertEqual(get_earmark, "")
+
+        # Attempt to remove an earmark when it's not set
+        self._fs_cmd("subvolume", "earmark", "rm", self.volname, subvolume)
+    
+    def test_set_invalid_earmark(self):
+        # Create a subvolume
+        subvolume = self._gen_subvol_name()
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+
+        # Attempt to set an invalid earmark
+        invalid_earmark = "invalid_format"
+        expected_message = (
+            f"Invalid earmark specified: '{invalid_earmark}'. A valid earmark should "
+            "either be empty or start with 'nfs' or 'smb', followed by dot-separated "
+            "non-empty components."
+        )
+        try:
+            self._fs_cmd("subvolume", "earmark", "set", self.volname, subvolume, "--earmark", invalid_earmark)
+        except CommandFailedError as ce:
+            self.assertEqual(ce.exitstatus, errno.EINVAL, expected_message)
+
+    def test_earmark_on_deleted_subvolume_with_retained_snapshot(self):
+        subvolume = self._gen_subvol_name()
+        snapshot = self._gen_subvol_snap_name()
+
+        # Create subvolume and snapshot
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+        self._fs_cmd("subvolume", "snapshot", "create", self.volname, subvolume, snapshot)
+
+        # Delete subvolume while retaining the snapshot
+        self._fs_cmd("subvolume", "rm", self.volname, subvolume, "--retain-snapshots")
+
+        # Define the expected error message
+        error_message = f'subvolume "{subvolume}" is removed and has only snapshots retained'
+
+        # Test cases for setting, getting, and removing earmarks
+        for operation in ["get", "rm", "set"]:
+            try:
+                extra_arg = "smb" if operation == "set" else None
+                if operation == "set":
+                    self._fs_cmd("subvolume", "earmark", operation, self.volname, subvolume, "--earmark", extra_arg)
+                else:
+                    self._fs_cmd("subvolume", "earmark", operation, self.volname, subvolume)
+            except CommandFailedError as ce:
+                self.assertEqual(ce.exitstatus, errno.ENOENT, error_message)
 
     def test_subvolume_expand(self):
         """
@@ -2440,6 +2558,14 @@ def test_subvolume_info(self):
         for feature in ['snapshot-clone', 'snapshot-autoprotect', 'snapshot-retention']:
             self.assertIn(feature, subvol_info["features"], msg="expected feature '{0}' in subvolume".format(feature))
 
+        # set earmark
+        earmark = "smb.test"
+        self._fs_cmd("subvolume", "earmark", "set", self.volname, subvolume, "--earmark", earmark)
+
+        subvol_info = json.loads(self._get_subvolume_info(self.volname, subvolume))
+
+        self.assertEqual(subvol_info["earmark"], earmark)
+        
         # remove subvolumes
         self._fs_cmd("subvolume", "rm", self.volname, subvolume)
 

src/pybind/mgr/volumes/fs/operations/subvolume.py

@@ -5,7 +5,7 @@
 from .template import SubvolumeOpType
 from .versions import loaded_subvolumes
 
-def create_subvol(mgr, fs, vol_spec, group, subvolname, size, isolate_nspace, pool, mode, uid, gid):
+def create_subvol(mgr, fs, vol_spec, group, subvolname, size, isolate_nspace, pool, mode, uid, gid, earmark):
     """
     create a subvolume (create a subvolume with the max known version).
 
@@ -18,10 +18,11 @@ def create_subvol(mgr, fs, vol_spec, group, subvolname, size, isolate_nspace, po
     :param mode: the user permissions
     :param uid: the user identifier
     :param gid: the group identifier
+    :param earmark: metadata string to identify if subvolume is associated with nfs/smb
     :return: None
     """
     subvolume = loaded_subvolumes.get_subvolume_object_max(mgr, fs, vol_spec, group, subvolname)
-    subvolume.create(size, isolate_nspace, pool, mode, uid, gid)
+    subvolume.create(size, isolate_nspace, pool, mode, uid, gid, earmark)
 
 
 def create_clone(mgr, fs, vol_spec, group, subvolname, pool, source_volume, source_subvolume, snapname):

src/pybind/mgr/volumes/fs/operations/template.py

@@ -68,6 +68,9 @@ class SubvolumeOpType(Enum):
     SNAP_METADATA_GET     = 'snap-metadata-get'
     SNAP_METADATA_LIST    = 'snap-metadata-ls'
     SNAP_METADATA_REMOVE  = 'snap-metadata-rm'
+    EARMARK_GET           = 'earmark-get'
+    EARMARK_SET           = 'earmark-set'
+    EARMARK_CLEAR          = 'earmark-clear'
 
 class SubvolumeTemplate(object):
     VERSION = None # type: int

src/pybind/mgr/volumes/fs/operations/versions/subvolume_base.py

@@ -18,6 +18,8 @@
 from .auth_metadata import AuthMetadataManager
 from .subvolume_attrs import SubvolumeStates
 
+from ceph.fs.earmarking import CephFSVolumeEarmarking, EarmarkException  # type: ignore
+
 log = logging.getLogger(__name__)
 
 
@@ -192,6 +194,14 @@ def get_attrs(self, pathname):
         except cephfs.NoData:
             attrs["quota"] = None
 
+        try:
+            fs_earmark = CephFSVolumeEarmarking(self.fs, pathname)
+            attrs["earmark"] = fs_earmark.get_earmark()
+        except cephfs.NoData:
+            attrs["earmark"] = ''
+        except EarmarkException:
+            attrs["earmark"] = ''
+
         return attrs
 
     def set_attrs(self, path, attrs):
@@ -277,6 +287,12 @@ def set_attrs(self, path, attrs):
         if mode is not None:
             self.fs.lchmod(path, mode)
 
+        # set earmark
+        earmark = attrs.get("earmark")
+        if earmark is not None:
+            fs_earmark = CephFSVolumeEarmarking(self.fs, path)
+            fs_earmark.set_earmark(earmark)
+
     def _resize(self, path, newsize, noshrink):
         try:
             newsize = int(newsize)
@@ -418,6 +434,14 @@ def info(self):
         except cephfs.Error as e:
             raise VolumeException(-e.args[0], e.args[1])
 
+        try:
+            fs_earmark = CephFSVolumeEarmarking(self.fs, subvolpath)
+            earmark = fs_earmark.get_earmark()
+        except cephfs.NoData:
+            earmark = ''
+        except EarmarkException:
+            earmark = ''
+
         return {'path': subvolpath,
                 'type': etype.value,
                 'uid': int(st["uid"]),
@@ -434,7 +458,9 @@ def info(self):
                 if nsize == 0
                 else '{0:.2f}'.format((float(usedbytes) / nsize) * 100.0),
                 'pool_namespace': pool_namespace,
-                'features': self.features, 'state': self.state.value}
+                'features': self.features,
+                'state': self.state.value,
+                'earmark': earmark}
 
     def set_user_metadata(self, keyname, value):
         try:

src/pybind/mgr/volumes/fs/operations/versions/subvolume_v1.py

@@ -85,7 +85,7 @@ def snapshot_data_path(self, snapname):
         """ Path to user data directory within a subvolume snapshot named 'snapname' """
         return self.snapshot_path(snapname)
 
-    def create(self, size, isolate_nspace, pool, mode, uid, gid):
+    def create(self, size, isolate_nspace, pool, mode, uid, gid, earmark):
         subvolume_type = SubvolumeTypes.TYPE_NORMAL
         try:
             initial_state = SubvolumeOpSm.get_init_state(subvolume_type)
@@ -103,7 +103,8 @@ def create(self, size, isolate_nspace, pool, mode, uid, gid):
                 'gid': gid,
                 'data_pool': pool,
                 'pool_namespace': self.namespace if isolate_nspace else None,
-                'quota': size
+                'quota': size,
+                'earmark': earmark
             }
             self.set_attrs(subvol_path, attrs)
 

src/pybind/mgr/volumes/fs/operations/versions/subvolume_v2.py

@@ -154,7 +154,7 @@ def _set_incarnation_metadata(self, subvolume_type, qpath, initial_state):
         self.metadata_mgr.update_global_section(MetadataManager.GLOBAL_META_KEY_PATH, qpath)
         self.metadata_mgr.update_global_section(MetadataManager.GLOBAL_META_KEY_STATE, initial_state.value)
 
-    def create(self, size, isolate_nspace, pool, mode, uid, gid):
+    def create(self, size, isolate_nspace, pool, mode, uid, gid, earmark):
         subvolume_type = SubvolumeTypes.TYPE_NORMAL
         try:
             initial_state = SubvolumeOpSm.get_init_state(subvolume_type)
@@ -175,7 +175,8 @@ def create(self, size, isolate_nspace, pool, mode, uid, gid):
                 'gid': gid,
                 'data_pool': pool,
                 'pool_namespace': self.namespace if isolate_nspace else None,
-                'quota': size
+                'quota': size,
+                'earmark': earmark
             }
             self.set_attrs(subvol_path, attrs)