mgr/volumes: set, get, list and remove custom metadata for subvolume

If CephFS in ODF configured in external mode, user like to use
volume / subvolume metadata to store some Openshift specific
information, as the PVC / PV / namespace the volumes / subvolumes
are coming from. For RBD volumes, it's possible to add metadata
information to the images using the 'rbd image-meta' command.
However, this feature is not available for CephFS volumes.
We'd like to request this capability.

Adding following commands:
    ceph fs subvolume metadata set <vol_name> <sub_name> <key_name> <value> [<group_name>]
    ceph fs subvolume metadata get <vol_name> <sub_name> <key_name> [<group_name>]
    ceph fs subvolume metadata ls <vol_name> <sub_name> [<group_name>]
    ceph fs subvolume metadata rm <vol_name> <sub_name> <key_name> [<group_name>] [--force]

Fixes: https://tracker.ceph.com/issues/54472
Signed-off-by: Nikhilkumar Shelke <nshelke@redhat.com>

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

@@ -37,29 +37,33 @@ def list_snapshots(self):
 
 @unique
 class SubvolumeOpType(Enum):
-    CREATE          = 'create'
-    REMOVE          = 'rm'
-    REMOVE_FORCE    = 'rm-force'
-    PIN             = 'pin'
-    LIST            = 'ls'
-    GETPATH         = 'getpath'
-    INFO            = 'info'
-    RESIZE          = 'resize'
-    SNAP_CREATE     = 'snap-create'
-    SNAP_REMOVE     = 'snap-rm'
-    SNAP_LIST       = 'snap-ls'
-    SNAP_INFO       = 'snap-info'
-    SNAP_PROTECT    = 'snap-protect'
-    SNAP_UNPROTECT  = 'snap-unprotect'
-    CLONE_SOURCE    = 'clone-source'
-    CLONE_CREATE    = 'clone-create'
-    CLONE_STATUS    = 'clone-status'
-    CLONE_CANCEL    = 'clone-cancel'
-    CLONE_INTERNAL  = 'clone_internal'
-    ALLOW_ACCESS    = 'allow-access'
-    DENY_ACCESS     = 'deny-access'
-    AUTH_LIST       = 'auth-list'
-    EVICT           = 'evict'
+    CREATE                = 'create'
+    REMOVE                = 'rm'
+    REMOVE_FORCE          = 'rm-force'
+    PIN                   = 'pin'
+    LIST                  = 'ls'
+    GETPATH               = 'getpath'
+    INFO                  = 'info'
+    RESIZE                = 'resize'
+    SNAP_CREATE           = 'snap-create'
+    SNAP_REMOVE           = 'snap-rm'
+    SNAP_LIST             = 'snap-ls'
+    SNAP_INFO             = 'snap-info'
+    SNAP_PROTECT          = 'snap-protect'
+    SNAP_UNPROTECT        = 'snap-unprotect'
+    CLONE_SOURCE          = 'clone-source'
+    CLONE_CREATE          = 'clone-create'
+    CLONE_STATUS          = 'clone-status'
+    CLONE_CANCEL          = 'clone-cancel'
+    CLONE_INTERNAL        = 'clone_internal'
+    ALLOW_ACCESS          = 'allow-access'
+    DENY_ACCESS           = 'deny-access'
+    AUTH_LIST             = 'auth-list'
+    EVICT                 = 'evict'
+    USER_METADATA_SET     = 'metadata-set'
+    USER_METADATA_GET     = 'metadata-get'
+    USER_METADATA_LIST    = 'metadata-ls'
+    USER_METADATA_REMOVE  = 'metadata-rm'
 
 class SubvolumeTemplate(object):
     VERSION = None # type: int

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

@@ -21,6 +21,7 @@
 
 class MetadataManager(object):
     GLOBAL_SECTION = "GLOBAL"
+    USER_METADATA_SECTION   = "USER_METADATA"
     GLOBAL_META_KEY_VERSION = "version"
     GLOBAL_META_KEY_TYPE    = "type"
     GLOBAL_META_KEY_PATH    = "path"
@@ -109,7 +110,7 @@ def add_section(self, section):
     def remove_option(self, section, key):
         if not self.config.has_section(section):
             raise MetadataMgrException(-errno.ENOENT, "section '{0}' does not exist".format(section))
-        self.config.remove_option(section, key)
+        return self.config.remove_option(section, key)
 
     def remove_section(self, section):
         self.config.remove_section(section)
@@ -138,6 +139,14 @@ def get_option(self, section, key):
     def get_global_option(self, key):
         return self.get_option(MetadataManager.GLOBAL_SECTION, key)
 
+    def list_all_options_from_section(self, section):
+        metadata_dict = {}
+        if self.config.has_section(section):
+            options = self.config.options(section)
+            for option in options:
+                metadata_dict[option] = self.config.get(section,option)
+        return metadata_dict
+
     def section_has_item(self, section, item):
         if not self.config.has_section(section):
             raise MetadataMgrException(-errno.ENOENT, "section '{0}' does not exist".format(section))

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

@@ -405,3 +405,31 @@ def info(self):
                 else '{0:.2f}'.format((float(usedbytes) / nsize) * 100.0),
                 'pool_namespace': pool_namespace,
                 'features': self.features, 'state': self.state.value}
+
+    def set_user_metadata(self, keyname, value):
+        self.metadata_mgr.add_section(MetadataManager.USER_METADATA_SECTION)
+        self.metadata_mgr.update_section(MetadataManager.USER_METADATA_SECTION, keyname, str(value))
+        self.metadata_mgr.flush()
+
+    def get_user_metadata(self, keyname):
+        try:
+            value = self.metadata_mgr.get_option(MetadataManager.USER_METADATA_SECTION, keyname)
+        except MetadataMgrException as me:
+            if me.errno == -errno.ENOENT:
+                raise VolumeException(-errno.ENOENT, "key '{0}' does not exist.".format(keyname))
+            raise VolumeException(-me.args[0], me.args[1])
+        return value
+
+    def list_user_metadata(self):
+        return self.metadata_mgr.list_all_options_from_section(MetadataManager.USER_METADATA_SECTION)
+
+    def remove_user_metadata(self, keyname):
+        try:
+            ret = self.metadata_mgr.remove_option(MetadataManager.USER_METADATA_SECTION, keyname)
+            if not ret:
+                raise VolumeException(-errno.ENOENT, "key '{0}' does not exist.".format(keyname))
+            self.metadata_mgr.flush()
+        except MetadataMgrException as me:
+            if me.errno == -errno.ENOENT:
+                raise VolumeException(-errno.ENOENT, "subvolume metadata not does not exist")
+            raise VolumeException(-me.args[0], me.args[1])

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

@@ -372,6 +372,7 @@ def remove(self, retainsnaps=False, internal_cleanup=False):
                 return
         if self.state != SubvolumeStates.STATE_RETAINED:
             self.trash_incarnation_dir()
+            self.metadata_mgr.remove_section(MetadataManager.USER_METADATA_SECTION)
             self.metadata_mgr.update_global_section(MetadataManager.GLOBAL_META_KEY_PATH, "")
             self.metadata_mgr.update_global_section(MetadataManager.GLOBAL_META_KEY_STATE, SubvolumeStates.STATE_RETAINED.value)
             self.metadata_mgr.flush()

src/pybind/mgr/volumes/fs/volume.py

@@ -383,6 +383,74 @@ def subvolume_info(self, **kwargs):
             ret = self.volume_exception_to_retval(ve)
         return ret
 
+    def set_user_metadata(self, **kwargs):
+        ret        = 0, "", ""
+        volname    = kwargs['vol_name']
+        subvolname = kwargs['sub_name']
+        groupname  = kwargs['group_name']
+        keyname    = kwargs['key_name']
+        value      = kwargs['value']
+
+        try:
+            with open_volume(self, volname) as fs_handle:
+                with open_group(fs_handle, self.volspec, groupname) as group:
+                    with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.USER_METADATA_SET) as subvolume:
+                        subvolume.set_user_metadata(keyname, value)
+        except VolumeException as ve:
+            ret = self.volume_exception_to_retval(ve)
+        return ret
+
+    def get_user_metadata(self, **kwargs):
+        ret        = 0, "", ""
+        volname    = kwargs['vol_name']
+        subvolname = kwargs['sub_name']
+        groupname  = kwargs['group_name']
+        keyname    = kwargs['key_name']
+
+        try:
+            with open_volume(self, volname) as fs_handle:
+                with open_group(fs_handle, self.volspec, groupname) as group:
+                    with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.USER_METADATA_GET) as subvolume:
+                        value = subvolume.get_user_metadata(keyname)
+                        ret = 0, value, ""
+        except VolumeException as ve:
+            ret = self.volume_exception_to_retval(ve)
+        return ret
+
+    def list_user_metadata(self, **kwargs):
+        ret        = 0, "", ""
+        volname    = kwargs['vol_name']
+        subvolname = kwargs['sub_name']
+        groupname  = kwargs['group_name']
+
+        try:
+            with open_volume(self, volname) as fs_handle:
+                with open_group(fs_handle, self.volspec, groupname) as group:
+                    with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.USER_METADATA_LIST) as subvolume:
+                        subvol_metadata_dict = subvolume.list_user_metadata()
+                        ret = 0, json.dumps(subvol_metadata_dict, indent=4, sort_keys=True), ""
+        except VolumeException as ve:
+            ret = self.volume_exception_to_retval(ve)
+        return ret
+
+    def remove_user_metadata(self, **kwargs):
+        ret        = 0, "", ""
+        volname    = kwargs['vol_name']
+        subvolname = kwargs['sub_name']
+        groupname  = kwargs['group_name']
+        keyname    = kwargs['key_name']
+        force      = kwargs['force']
+
+        try:
+            with open_volume(self, volname) as fs_handle:
+                with open_group(fs_handle, self.volspec, groupname) as group:
+                    with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.USER_METADATA_REMOVE) as subvolume:
+                        subvolume.remove_user_metadata(keyname)
+        except VolumeException as ve:
+            if not (ve.errno == -errno.ENOENT and force):
+                ret = self.volume_exception_to_retval(ve)
+        return ret
+
     def list_subvolumes(self, **kwargs):
         ret        = 0, "", ""
         volname    = kwargs['vol_name']

src/pybind/mgr/volumes/module.py

@@ -189,10 +189,51 @@ class Module(orchestrator.OrchestratorClientMixin, MgrModule):
                    'name=vol_name,type=CephString '
                    'name=sub_name,type=CephString '
                    'name=group_name,type=CephString,req=false ',
-            'desc': "Get the metadata of a CephFS subvolume in a volume, "
+            'desc': "Get the information of a CephFS subvolume in a volume, "
                     "and optionally, in a specific subvolume group",
             'perm': 'r'
         },
+        {
+            'cmd': 'fs subvolume metadata set '
+                   'name=vol_name,type=CephString '
+                   'name=sub_name,type=CephString '
+                   'name=key_name,type=CephString '
+                   'name=value,type=CephString '
+                   'name=group_name,type=CephString,req=false ',
+            'desc': "Set custom metadata (key-value) for a CephFS subvolume in a volume, "
+                    "and optionally, in a specific subvolume group",
+            'perm': 'rw'
+        },
+        {
+            'cmd': 'fs subvolume metadata get '
+                   'name=vol_name,type=CephString '
+                   'name=sub_name,type=CephString '
+                   'name=key_name,type=CephString '
+                   'name=group_name,type=CephString,req=false ',
+            'desc': "Get custom metadata associated with the key of a CephFS subvolume in a volume, "
+                    "and optionally, in a specific subvolume group",
+            'perm': 'r'
+        },
+        {
+            'cmd': 'fs subvolume metadata ls '
+                   'name=vol_name,type=CephString '
+                   'name=sub_name,type=CephString '
+                   'name=group_name,type=CephString,req=false ',
+            'desc': "List custom metadata (key-value pairs) of a CephFS subvolume in a volume, "
+                    "and optionally, in a specific subvolume group",
+            'perm': 'r'
+        },
+        {
+            'cmd': 'fs subvolume metadata rm '
+                   'name=vol_name,type=CephString '
+                   'name=sub_name,type=CephString '
+                   'name=key_name,type=CephString '
+                   'name=group_name,type=CephString,req=false '
+                   'name=force,type=CephBool,req=false ',
+            'desc': "Remove custom metadata (key-value) associated with the key of a CephFS subvolume in a volume, "
+                    "and optionally, in a specific subvolume group",
+            'perm': 'rw'
+        },
         {
             'cmd': 'fs subvolumegroup pin'
                    ' name=vol_name,type=CephString'
@@ -543,6 +584,35 @@ def _cmd_fs_subvolume_info(self, inbuf, cmd):
                                       sub_name=cmd['sub_name'],
                                       group_name=cmd.get('group_name', None))
 
+    @mgr_cmd_wrap
+    def _cmd_fs_subvolume_metadata_set(self, inbuf, cmd):
+        return self.vc.set_user_metadata(vol_name=cmd['vol_name'],
+                                      sub_name=cmd['sub_name'],
+                                      key_name=cmd['key_name'],
+                                      value=cmd['value'],
+                                      group_name=cmd.get('group_name', None))
+
+    @mgr_cmd_wrap
+    def _cmd_fs_subvolume_metadata_get(self, inbuf, cmd):
+        return self.vc.get_user_metadata(vol_name=cmd['vol_name'],
+                                      sub_name=cmd['sub_name'],
+                                      key_name=cmd['key_name'],
+                                      group_name=cmd.get('group_name', None))
+
+    @mgr_cmd_wrap
+    def _cmd_fs_subvolume_metadata_ls(self, inbuf, cmd):
+        return self.vc.list_user_metadata(vol_name=cmd['vol_name'],
+                                      sub_name=cmd['sub_name'],
+                                      group_name=cmd.get('group_name', None))
+
+    @mgr_cmd_wrap
+    def _cmd_fs_subvolume_metadata_rm(self, inbuf, cmd):
+        return self.vc.remove_user_metadata(vol_name=cmd['vol_name'],
+                                      sub_name=cmd['sub_name'],
+                                      key_name=cmd['key_name'],
+                                      group_name=cmd.get('group_name', None),
+                                      force=cmd.get('force', False))
+
     @mgr_cmd_wrap
     def _cmd_fs_subvolumegroup_pin(self, inbuf, cmd):
         return self.vc.pin_subvolume_group(vol_name=cmd['vol_name'],